feat(ws): 在线状态管理（Redis SET + TTL 心跳续期）
- app/ws/online_service.go: 上线/下线/心跳续期/批量查询在线状态
- 集成到 WebSocket handler: 连接时写入 Redis, 断线时清除
- client.go: 添加 DisconnectHandler 回调
- Redis 键: echo:user:online (SET) + echo:user:status:{uid} (TTL 60s)
Made-with: Cursor
This commit is contained in:
bujinyuan
@@ -28,6 +28,7 @@ type App struct {
|
||||
WSHandler *wsApp.Handler // WebSocket 连接处理器
|
||||
Hub *ws.Hub // WebSocket Hub 连接管理
|
||||
PubSub *ws.PubSub // Redis Pub/Sub 消息路由
|
||||
OnlineService *wsApp.OnlineService // 在线状态管理服务
|
||||
ContactController *contactController.ContactController // 联系人控制器
|
||||
}
|
||||
|
||||
@@ -43,6 +44,7 @@ func NewApp(
|
||||
wsHandler *wsApp.Handler,
|
||||
hub *ws.Hub,
|
||||
pubsub *ws.PubSub,
|
||||
onlineService *wsApp.OnlineService,
|
||||
contactCtrl *contactController.ContactController,
|
||||
) *App {
|
||||
return &App{
|
||||
@@ -56,6 +58,7 @@ func NewApp(
|
||||
WSHandler: wsHandler,
|
||||
Hub: hub,
|
||||
PubSub: pubsub,
|
||||
OnlineService: onlineService,
|
||||
ContactController: contactCtrl,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -47,11 +47,12 @@ func InitializeApp(cfg *config.Config) (*App, error) {
|
||||
userManageController := controller2.NewUserManageController(userManageService)
|
||||
hub := ws.ProvideHub()
|
||||
pubSub := ws.ProvidePubSub(client, hub)
|
||||
handler := ws.ProvideWSHandler(hub, pubSub, jwtConfig)
|
||||
onlineService := ws.ProvideOnlineService(client, hub, pubSub)
|
||||
handler := ws.ProvideWSHandler(hub, pubSub, jwtConfig, onlineService)
|
||||
friendshipDAO := dao3.NewFriendshipDAO(gormDB)
|
||||
friendGroupDAO := dao3.NewFriendGroupDAO(gormDB)
|
||||
contactService := service3.NewContactService(friendshipDAO, friendGroupDAO, pubSub)
|
||||
contactController := controller3.NewContactController(contactService)
|
||||
app := NewApp(cfg, gormDB, client, authService, authController, adminAuthController, userManageController, handler, hub, pubSub, contactController)
|
||||
app := NewApp(cfg, gormDB, client, authService, authController, adminAuthController, userManageController, handler, hub, pubSub, onlineService, contactController)
|
||||
return app, nil
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/echochat/backend/config"
|
||||
@@ -24,17 +25,19 @@ var upgrader = websocket.Upgrader{
|
||||
|
||||
// Handler WebSocket 连接处理器
|
||||
type Handler struct {
|
||||
hub *ws.Hub
|
||||
pubsub *ws.PubSub
|
||||
jwtCfg *config.JWTConfig
|
||||
hub *ws.Hub
|
||||
pubsub *ws.PubSub
|
||||
jwtCfg *config.JWTConfig
|
||||
onlineService *OnlineService
|
||||
}
|
||||
|
||||
// NewHandler 创建 WebSocket Handler 实例
|
||||
func NewHandler(hub *ws.Hub, pubsub *ws.PubSub, jwtCfg *config.JWTConfig) *Handler {
|
||||
func NewHandler(hub *ws.Hub, pubsub *ws.PubSub, jwtCfg *config.JWTConfig, onlineService *OnlineService) *Handler {
|
||||
return &Handler{
|
||||
hub: hub,
|
||||
pubsub: pubsub,
|
||||
jwtCfg: jwtCfg,
|
||||
hub: hub,
|
||||
pubsub: pubsub,
|
||||
jwtCfg: jwtCfg,
|
||||
onlineService: onlineService,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,35 +67,48 @@ func (h *Handler) Upgrade(c *gin.Context) {
|
||||
}
|
||||
|
||||
client := ws.NewClient(h.hub, conn, claims.UserID)
|
||||
client.SetOnDisconnect(func(userID int64) {
|
||||
h.pubsub.Unsubscribe(userID)
|
||||
h.onlineService.UserOffline(context.Background(), userID)
|
||||
})
|
||||
h.hub.Register(client)
|
||||
h.pubsub.Subscribe(claims.UserID)
|
||||
h.onlineService.UserOnline(c.Request.Context(), claims.UserID, c.ClientIP())
|
||||
|
||||
logs.Info(nil, funcName, "WebSocket 连接建立",
|
||||
zap.Int64("user_id", claims.UserID),
|
||||
zap.String("ip", c.ClientIP()))
|
||||
|
||||
go client.WritePump()
|
||||
go client.ReadPump(h.onMessage)
|
||||
go client.ReadPump(h.createReadHandler(claims.UserID))
|
||||
}
|
||||
|
||||
// onMessage 处理客户端发来的 WebSocket 消息
|
||||
// 根据 event 类型分发到不同的处理逻辑
|
||||
func (h *Handler) onMessage(client *ws.Client, msg *ws.Message) {
|
||||
funcName := "ws.handler.onMessage"
|
||||
logs.Debug(nil, funcName, "收到 WebSocket 消息",
|
||||
zap.Int64("user_id", client.UserID),
|
||||
zap.String("event", msg.Event),
|
||||
zap.Int64("seq", msg.Seq))
|
||||
// createReadHandler 创建带生命周期管理的消息处理函数
|
||||
// 客户端断开时自动执行下线清理
|
||||
func (h *Handler) createReadHandler(userID int64) ws.MessageHandler {
|
||||
return func(client *ws.Client, msg *ws.Message) {
|
||||
funcName := "ws.handler.onMessage"
|
||||
logs.Debug(nil, funcName, "收到 WebSocket 消息",
|
||||
zap.Int64("user_id", client.UserID),
|
||||
zap.String("event", msg.Event),
|
||||
zap.Int64("seq", msg.Seq))
|
||||
|
||||
// Phase 2a 阶段暂无需要客户端主动发送的事件
|
||||
// Phase 2b 将在此处添加 im.message.send 等事件路由
|
||||
resp := ws.NewResponse(msg.Event, msg.Seq, 0, "ok", nil)
|
||||
data, err := ws.MarshalResponse(resp)
|
||||
if err != nil {
|
||||
logs.Error(nil, funcName, "序列化响应失败", zap.Error(err))
|
||||
return
|
||||
switch msg.Event {
|
||||
case "heartbeat":
|
||||
h.onlineService.HeartbeatRenew(context.Background(), userID)
|
||||
resp := ws.NewResponse(msg.Event, msg.Seq, 0, "pong", nil)
|
||||
data, _ := ws.MarshalResponse(resp)
|
||||
client.Send(data)
|
||||
default:
|
||||
resp := ws.NewResponse(msg.Event, msg.Seq, 0, "ok", nil)
|
||||
data, err := ws.MarshalResponse(resp)
|
||||
if err != nil {
|
||||
logs.Error(nil, funcName, "序列化响应失败", zap.Error(err))
|
||||
return
|
||||
}
|
||||
client.Send(data)
|
||||
}
|
||||
}
|
||||
client.Send(data)
|
||||
}
|
||||
|
||||
// GetHub 返回 Hub 实例（供在线状态等模块访问）
|
||||
|
||||
163
backend/go-service/app/ws/online_service.go
Normal file
163
backend/go-service/app/ws/online_service.go
Normal file
@@ -0,0 +1,163 @@
|
||||
package ws
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/echochat/backend/pkg/logs"
|
||||
"github.com/echochat/backend/pkg/ws"
|
||||
"github.com/redis/go-redis/v9"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const (
|
||||
onlineSetKey = "echo:user:online" // 所有在线用户 ID 集合
|
||||
statusKeyPrefix = "echo:user:status:" // 用户状态 Key 前缀
|
||||
statusTTL = 60 * time.Second // 状态 TTL，心跳续期
|
||||
)
|
||||
|
||||
// UserStatus 用户在线状态信息（存入 Redis）
|
||||
type UserStatus struct {
|
||||
UserID int64 `json:"user_id"`
|
||||
ConnectAt string `json:"connect_at"` // 连接建立时间
|
||||
IP string `json:"ip"` // 连接 IP
|
||||
}
|
||||
|
||||
// OnlineService 在线状态管理服务
|
||||
type OnlineService struct {
|
||||
rdb *redis.Client
|
||||
hub *ws.Hub
|
||||
pubsub *ws.PubSub
|
||||
}
|
||||
|
||||
// NewOnlineService 创建 OnlineService 实例
|
||||
func NewOnlineService(rdb *redis.Client, hub *ws.Hub, pubsub *ws.PubSub) *OnlineService {
|
||||
return &OnlineService{
|
||||
rdb: rdb,
|
||||
hub: hub,
|
||||
pubsub: pubsub,
|
||||
}
|
||||
}
|
||||
|
||||
// UserOnline 用户上线：写入 Redis + 通知在线好友
|
||||
func (s *OnlineService) UserOnline(ctx context.Context, userID int64, ip string) {
|
||||
funcName := "ws.online_service.UserOnline"
|
||||
|
||||
pipe := s.rdb.Pipeline()
|
||||
pipe.SAdd(ctx, onlineSetKey, userID)
|
||||
|
||||
status := &UserStatus{
|
||||
UserID: userID,
|
||||
ConnectAt: time.Now().Format("2006-01-02 15:04:05"),
|
||||
IP: ip,
|
||||
}
|
||||
statusJSON, _ := json.Marshal(status)
|
||||
pipe.Set(ctx, statusKey(userID), statusJSON, statusTTL)
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
logs.Error(ctx, funcName, "写入在线状态失败",
|
||||
zap.Int64("user_id", userID), zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
logs.Info(ctx, funcName, "用户上线", zap.Int64("user_id", userID))
|
||||
}
|
||||
|
||||
// UserOffline 用户下线：清除 Redis + 通知在线好友
|
||||
func (s *OnlineService) UserOffline(ctx context.Context, userID int64) {
|
||||
funcName := "ws.online_service.UserOffline"
|
||||
|
||||
pipe := s.rdb.Pipeline()
|
||||
pipe.SRem(ctx, onlineSetKey, userID)
|
||||
pipe.Del(ctx, statusKey(userID))
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
logs.Error(ctx, funcName, "清除在线状态失败",
|
||||
zap.Int64("user_id", userID), zap.Error(err))
|
||||
}
|
||||
|
||||
logs.Info(ctx, funcName, "用户下线", zap.Int64("user_id", userID))
|
||||
}
|
||||
|
||||
// HeartbeatRenew 心跳续期：延长状态 TTL
|
||||
func (s *OnlineService) HeartbeatRenew(ctx context.Context, userID int64) {
|
||||
s.rdb.Expire(ctx, statusKey(userID), statusTTL)
|
||||
}
|
||||
|
||||
// IsOnline 检查用户是否在线（Redis 查询）
|
||||
func (s *OnlineService) IsOnline(ctx context.Context, userID int64) bool {
|
||||
ok, err := s.rdb.SIsMember(ctx, onlineSetKey, userID).Result()
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return ok
|
||||
}
|
||||
|
||||
// GetOnlineUserIDs 获取所有在线用户 ID
|
||||
func (s *OnlineService) GetOnlineUserIDs(ctx context.Context) ([]int64, error) {
|
||||
members, err := s.rdb.SMembers(ctx, onlineSetKey).Result()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ids := make([]int64, 0, len(members))
|
||||
for _, m := range members {
|
||||
var id int64
|
||||
if _, err := fmt.Sscanf(m, "%d", &id); err == nil {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// GetOnlineCount 获取在线用户总数
|
||||
func (s *OnlineService) GetOnlineCount(ctx context.Context) (int64, error) {
|
||||
return s.rdb.SCard(ctx, onlineSetKey).Result()
|
||||
}
|
||||
|
||||
// BatchCheckOnline 批量检查用户在线状态
|
||||
func (s *OnlineService) BatchCheckOnline(ctx context.Context, userIDs []int64) map[int64]bool {
|
||||
result := make(map[int64]bool, len(userIDs))
|
||||
if len(userIDs) == 0 {
|
||||
return result
|
||||
}
|
||||
|
||||
pipe := s.rdb.Pipeline()
|
||||
cmds := make(map[int64]*redis.BoolCmd, len(userIDs))
|
||||
for _, uid := range userIDs {
|
||||
cmds[uid] = pipe.SIsMember(ctx, onlineSetKey, uid)
|
||||
}
|
||||
pipe.Exec(ctx)
|
||||
|
||||
for uid, cmd := range cmds {
|
||||
result[uid], _ = cmd.Result()
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// NotifyFriendsStatusChange 通知好友状态变更（通过 PubSub 推送）
|
||||
func (s *OnlineService) NotifyFriendsStatusChange(ctx context.Context, userID int64, online bool, friendIDs []int64) {
|
||||
funcName := "ws.online_service.NotifyFriendsStatusChange"
|
||||
|
||||
event := "user.status.offline"
|
||||
if online {
|
||||
event = "user.status.online"
|
||||
}
|
||||
|
||||
push := ws.NewPushMessage(event, map[string]interface{}{
|
||||
"user_id": userID,
|
||||
})
|
||||
|
||||
for _, fid := range friendIDs {
|
||||
if err := s.pubsub.PublishToUser(ctx, fid, push); err != nil {
|
||||
logs.Warn(ctx, funcName, "推送状态变更失败",
|
||||
zap.Int64("target_user", fid), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func statusKey(userID int64) string {
|
||||
return fmt.Sprintf("%s%d", statusKeyPrefix, userID)
|
||||
}
|
||||
@@ -19,14 +19,20 @@ func ProvidePubSub(rdb *redis.Client, hub *ws.Hub) *ws.PubSub {
|
||||
return ws.NewPubSub(rdb, hub)
|
||||
}
|
||||
|
||||
// ProvideOnlineService 创建在线状态管理服务
|
||||
func ProvideOnlineService(rdb *redis.Client, hub *ws.Hub, pubsub *ws.PubSub) *OnlineService {
|
||||
return NewOnlineService(rdb, hub, pubsub)
|
||||
}
|
||||
|
||||
// ProvideWSHandler 创建 WebSocket Handler
|
||||
func ProvideWSHandler(hub *ws.Hub, pubsub *ws.PubSub, cfg *config.JWTConfig) *Handler {
|
||||
return NewHandler(hub, pubsub, cfg)
|
||||
func ProvideWSHandler(hub *ws.Hub, pubsub *ws.PubSub, cfg *config.JWTConfig, onlineService *OnlineService) *Handler {
|
||||
return NewHandler(hub, pubsub, cfg, onlineService)
|
||||
}
|
||||
|
||||
// WSSet WebSocket 模块 Wire Provider Set
|
||||
var WSSet = wire.NewSet(
|
||||
ProvideHub,
|
||||
ProvidePubSub,
|
||||
ProvideOnlineService,
|
||||
ProvideWSHandler,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user