From 54b26cdece63138d8fbf5059af4b075acf86c7eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E7=90=A6?= Date: Sun, 8 Feb 2026 00:28:49 +0800 Subject: [PATCH] 1 --- controller/websocket_controller.go | 104 ++++++++++++++++++++++++++--- go.mod | 2 +- models/messages.go | 21 +++--- 3 files changed, 108 insertions(+), 19 deletions(-) diff --git a/controller/websocket_controller.go b/controller/websocket_controller.go index 4b6e6ce..b302a84 100644 --- a/controller/websocket_controller.go +++ b/controller/websocket_controller.go @@ -598,9 +598,23 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt if err := json.Unmarshal(message, &payload); err == nil && payload.RequestType != "" { log.Printf("📦📦📦📦📦📦📦📦 解析JSON消息成功: Type=%s", payload.RequestType) + // 处理unbind请求（清理旧连接） + // 注意：unbind主要用于清理其他旧连接，不应该关闭当前连接 + // 如果前端需要完全断开，应该直接关闭WebSocket连接 + if payload.RequestType == "unbind" && payload.SenderUserID != "" { + log.Printf("🔓🔓🔓🔓🔓🔓🔓🔓 处理解绑请求: \n ClientID=%s \n UserID=%s", senderID, payload.SenderUserID) + + // 清理该用户的所有旧连接（除了当前连接） + // 这样可以确保旧连接被清理，但当前连接仍然可用 + c.cleanOldUserConnections(payload.SenderUserID, senderID) + + log.Printf("✅ 解绑完成: ClientID=%s | UserID=%s (已清理旧连接，当前连接保持)", senderID, payload.SenderUserID) + return + } + if payload.RequestType == "bind" && payload.SenderUserID != "" { - log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理绑定请求: \n ClientID=%s \n UserID=%s \n UserType=%s \n Token=%s", - senderID, payload.SenderUserID, payload.UserType, maskToken(payload.Token)) + log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理绑定请求: \n ClientID=%s \n UserID=%s \n UserType=%s \n Token=%s \n CleanOldConnections=%v", + senderID, payload.SenderUserID, payload.UserType, maskToken(payload.Token), payload.CleanOldConnections) // 验证Token if payload.Token == "" { @@ -617,12 +631,13 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt return } - // 绑定用户 - if err := c.bindClientToUser(senderID, payload.SenderUserID); err != nil { + // 绑定用户（检查是否需要清理旧连接） + cleanOldConnections := payload.CleanOldConnections + if err := c.bindClientToUser(senderID, payload.SenderUserID, cleanOldConnections); err != nil { log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err) c.SendAuthResponse(senderID, "failed", "绑定失败") } else { - log.Printf("✅ 绑定成功: \n ClientID=%s \n UserID=%s", senderID, payload.SenderUserID) + log.Printf("✅ 绑定成功: \n ClientID=%s \n UserID=%s \n CleanOldConnections=%v", senderID, payload.SenderUserID, cleanOldConnections) // 保存客户端认证信息 clientInfo := &models.ClientInfo{ @@ -1171,7 +1186,8 @@ func (c *WebSocketController) BindHandler(ctx *gin.Context) { log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理用户绑定请求: \n UserID=%s \n ClientID=%s", req.UserID, req.ClientID) - if err := c.bindClientToUser(req.ClientID, req.UserID); err != nil { + // API绑定请求默认清理旧连接 + if err := c.bindClientToUser(req.ClientID, req.UserID, true); err != nil { log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err) ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return @@ -1416,13 +1432,85 @@ func (c *WebSocketController) cleanupUserBinding(clientID string) { log.Printf("🧹🧹🧹🧹🧹🧹🧹🧹 清理绑定关系: ClientID=%s | UserID=%s", clientID, userID) } +/** + * cleanOldUserConnections + * 功能：清理用户的所有旧连接（除了当前clientID） + * @param userID string 用户ID + * @param currentClientID string 当前客户端ID（保留此连接） + */ +func (c *WebSocketController) cleanOldUserConnections(userID, currentClientID string) { + // 获取该用户的所有客户端ID + userClientsKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID) + clientIDs, err := c.RedisCli.SMembers(c.RedisCtx, userClientsKey).Result() + if err != nil { + log.Printf("⚠️ 获取用户客户端列表失败: %v | UserID=%s", err, userID) + return + } + + if len(clientIDs) == 0 { + log.Printf("ℹ️ 用户无旧连接需要清理: UserID=%s", userID) + return + } + + log.Printf("🧹🧹🧹🧹🧹🧹🧹🧹 开始清理用户旧连接: \n UserID=%s \n 当前ClientID=%s \n 旧连接数=%d", userID, currentClientID, len(clientIDs)) + + cleanedCount := 0 + for _, oldClientID := range clientIDs { + // 跳过当前客户端ID + if oldClientID == currentClientID { + continue + } + + // 检查旧连接是否还在本节点 + c.ClientsMux.RLock() + conn, exists := c.Clients[oldClientID] + c.ClientsMux.RUnlock() + + if exists { + // 如果连接还在本节点，先关闭WebSocket连接 + log.Printf("🔌 关闭旧连接: ClientID=%s | UserID=%s", oldClientID, userID) + if conn != nil { + conn.Close() + } + // 然后调用removeClient清理（会同时清理Redis映射和本地连接） + c.removeClient(oldClientID) + } else { + // 如果连接不在本节点，只清理Redis映射 + log.Printf("🧹 清理Redis映射: ClientID=%s | UserID=%s", oldClientID, userID) + // 从用户-客户端映射中移除 + c.RedisCli.SRem(c.RedisCtx, userClientsKey, oldClientID) + // 从客户端-用户映射中移除 + clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, oldClientID) + c.RedisCli.Del(c.RedisCtx, clientUserKey) + } + + cleanedCount++ + } + + // 清空用户的所有客户端映射（稍后会在bindClientToUser中重新添加当前clientID） + if cleanedCount > 0 { + // 先移除当前clientID（如果存在），然后清空整个集合 + c.RedisCli.SRem(c.RedisCtx, userClientsKey, currentClientID) + c.RedisCli.Del(c.RedisCtx, userClientsKey) + log.Printf("✅ 清理完成: \n UserID=%s \n 清理连接数=%d", userID, cleanedCount) + } +} + /** * bindClientToUser * 功能：在 Redis 中建立 UserID 和 ClientID 的双向映射。 * 映射1：client_user_mapping:clientID -> userID (string) * 映射2：user_client_mapping:userID -> [clientID1, clientID2] (set) + * @param clientID string 客户端ID + * @param userID string 用户ID + * @param cleanOldConnections bool 是否清理该用户之前的旧连接（默认false） */ -func (c *WebSocketController) bindClientToUser(clientID, userID string) error { +func (c *WebSocketController) bindClientToUser(clientID, userID string, cleanOldConnections bool) error { + // 如果需要清理旧连接，先清理该用户的所有旧连接 + if cleanOldConnections { + c.cleanOldUserConnections(userID, clientID) + } + // 设置客户端->用户映射 clientUserKey := fmt.Sprintf("%s:%s", models.ClientUserKey, clientID) if err := c.RedisCli.Set(c.RedisCtx, clientUserKey, userID, 0).Err(); err != nil { @@ -1435,7 +1523,7 @@ func (c *WebSocketController) bindClientToUser(clientID, userID string) error { return fmt.Errorf("添加用户客户端集合失败: %v", err) } - log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 绑定成功: \n ClientID=%s \n UserID=%s", clientID, userID) + log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 绑定成功: \n ClientID=%s \n UserID=%s \n CleanOldConnections=%v", clientID, userID, cleanOldConnections) return nil } diff --git a/go.mod b/go.mod index 2bc25a9..38ba9f6 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.24.4 require ( github.com/gin-gonic/gin v1.10.1 github.com/go-redis/redis/v8 v8.11.5 + github.com/golang-jwt/jwt/v5 v5.3.0 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 github.com/joho/godotenv v1.5.1 @@ -27,7 +28,6 @@ require ( github.com/go-playground/validator/v10 v10.20.0 // indirect github.com/go-sql-driver/mysql v1.8.1 // indirect github.com/goccy/go-json v0.10.2 // indirect - github.com/golang-jwt/jwt/v5 v5.3.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/models/messages.go b/models/messages.go index 4d20984..e72e078 100644 --- a/models/messages.go +++ b/models/messages.go @@ -13,16 +13,17 @@ import "time" */ // 统一消息结构（发送方使用） type SendMessagePayload struct { - RequestType string `json:"request_type"` // 消息类型 - TargetClientID string `json:"target_client_id"` // 目标客户端ID - SenderUserID string `json:"sender_user_id"` // 发送者用户ID - ReceiverUserID string `json:"receiver_user_id"` // 接收者用户ID - MessageType int `json:"message_type"` // 消息类型（0-13） - MessageContent string `json:"message_content"` // 消息内容 - RoomId string `json:"room_id"` // 房间ID - Duration int `json:"duration"` // 时长（音频、视频、音视频通话） - Token string `json:"token,omitempty"` // 认证Token - UserType string `json:"user_type,omitempty"` // 用户类型: user/doctor + RequestType string `json:"request_type"` // 消息类型 + TargetClientID string `json:"target_client_id"` // 目标客户端ID + SenderUserID string `json:"sender_user_id"` // 发送者用户ID + ReceiverUserID string `json:"receiver_user_id"` // 接收者用户ID + MessageType int `json:"message_type"` // 消息类型（0-13） + MessageContent string `json:"message_content"` // 消息内容 + RoomId string `json:"room_id"` // 房间ID + Duration int `json:"duration"` // 时长（音频、视频、音视频通话） + Token string `json:"token,omitempty"` // 认证Token + UserType string `json:"user_type,omitempty"` // 用户类型: user/doctor + CleanOldConnections bool `json:"clean_old_connections"` // 是否清理旧连接（bind时使用） // 通话专用字段 CallID string `json:"call_id,omitempty"` // 通话唯一ID CallStatus string `json:"call_status,omitempty"` // 通话状态:invite/accepted/rejected/ended/candidate