修改部分逻辑

2025-07-05 15:57:28 +08:00
parent 9f267927c6
commit 1706db2532
7 changed files with 533 additions and 56 deletions
--- a/controller/websocket_controller.go
+++ b/controller/websocket_controller.go
@@ -29,6 +29,7 @@ var upgrader = websocket.Upgrader{
 type WebSocketController struct {
 	Clients    map[string]*websocket.Conn
 	ClientsMux sync.RWMutex
+	WriteMutex sync.Mutex // 添加写锁
 	RedisCli   *redis.Client
 	NodeID     string
 	Port       string
@@ -50,7 +51,7 @@ func (c *WebSocketController) ConfigureSystem() {
 	// 只有当端口未设置时才从环境变量获取
 	if c.Port == "" {
 		c.Port = utils.GetEnv("PORT", "12080")
-		log.Printf("📡📡📡📡📡📡📡📡 使用环境变量设置端口: %s", c.Port)
+		log.Printf("📡📡📡📡📡📡📡📡📡📡📡📡📡📡📡📡 使用环境变量设置端口: %s", c.Port)
 	}

 	log.SetPrefix(fmt.Sprintf("[Node:%s] ", c.NodeID))
@@ -79,7 +80,7 @@ func (c *WebSocketController) registerNode() {
 		if err := c.RedisCli.Set(c.RedisCtx, key, value, 30*time.Second).Err(); err != nil {
 			log.Printf("⚠️ 节点注册失败: %v", err)
 		} else {
-			log.Printf("📌📌📌📌📌📌📌📌 节点已注册: %s = %s", key, value)
+			log.Printf("📌📌📌📌📌📌📌📌📌📌📌📌📌📌📌📌 节点已注册: %s = %s", key, value)
 		}

 		time.Sleep(20 * time.Second)
@@ -90,7 +91,7 @@ func (c *WebSocketController) registerNode() {
 func (c *WebSocketController) configureLogger() {
 	logDir := utils.GetEnv("LOG_DIR", "./logs")
 	if err := os.MkdirAll(logDir, 0755); err != nil {
-		log.Fatalf("❌❌❌❌ 创建日志目录失败: %v", err)
+		log.Fatalf("❌❌❌❌❌❌❌❌ 创建日志目录失败: %v", err)
 	}

 	logFreq := utils.GetEnv("LOG_ROTATE_FREQ", "daily")
@@ -118,7 +119,7 @@ func (c *WebSocketController) InitRedisClient() {
 	})

 	if err := c.checkRedisConnection(); err != nil {
-		log.Fatalf("❌❌❌❌ Redis连接失败: %v", err)
+		log.Fatalf("❌❌❌❌❌❌❌❌ Redis连接失败: %v", err)
 	}
 }

@@ -131,16 +132,16 @@ func (c *WebSocketController) checkRedisConnection() error {
 // PrintStartupInfo 打印启动信息
 func (c *WebSocketController) PrintStartupInfo() {
 	hostname, _ := os.Hostname()
-	log.Printf("🚀🚀🚀🚀 WebSocket服务启动: NodeID=%s", c.NodeID)
-	log.Printf("🌐🌐🌐🌐 监听端口: %s", c.Port)
-	log.Printf("📡📡📡📡 Redis地址: %s", utils.GetEnv("REDIS_ADDR", "localhost:6379"))
-	log.Printf("💻💻💻💻 主机: %s", hostname)
-	log.Printf("🕒🕒🕒🕒🕒🕒🕒🕒🕒 启动时间: %s", time.Now().Format("2006-01-02 15:04:05"))
-	log.Printf("🔔🔔🔔🔔 支持消息类型: \n  0:文本\n  1:图片\n  2:音频\n  3:视频\n  4:处方\n  5:病例\n  6:视频通话")
-	log.Printf("🔑🔑🔑🔑 用户绑定键: \n  %s\n  %s", models.ClientUserKey, models.UserClientKey)
-	log.Printf("📬📬📬📬 新增接口: POST /send-to-user (直接通过用户ID发送消息)")
-	log.Printf("🔗🔗🔗🔗 集群节点注册: %s:%s", utils.GetOutboundIP(), c.Port)
-	log.Println("🔗🔗🔗🔗 等待客户端连接...")
+	log.Printf("🚀🚀🚀🚀🚀🚀🚀🚀 WebSocket服务启动: NodeID=%s", c.NodeID)
+	log.Printf("🌐🌐🌐🌐🌐🌐🌐🌐 监听端口: %s", c.Port)
+	log.Printf("📡📡📡📡📡📡📡📡 Redis地址: %s", utils.GetEnv("REDIS_ADDR", "localhost:6379"))
+	log.Printf("💻💻💻💻💻💻💻💻 主机: %s", hostname)
+	log.Printf("🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒🕒 启动时间: %s", time.Now().Format("2006-01-02 15:04:05"))
+	log.Printf("🔔🔔🔔🔔🔔🔔🔔🔔 支持消息类型: \n  0:文本\n  1:图片\n  2:音频\n  3:视频\n  4:处方\n  5:病例\n  6:视频通话\n  7:语音通话\n  8:文件消息")
+	log.Printf("🔑🔑🔑🔑🔑🔑🔑🔑 用户绑定键: \n  %s\n  %s", models.ClientUserKey, models.UserClientKey)
+	log.Printf("📬📬📬📬📬📬📬📬 新增接口: POST /send-to-user (直接通过用户ID发送消息)")
+	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 集群节点注册: %s:%s", utils.GetOutboundIP(), c.Port)
+	log.Println("🔗🔗🔗🔗🔗🔗🔗🔗 等待客户端连接...")
 }

 // HealthHandler 健康检查处理器
@@ -157,7 +158,7 @@ func (c *WebSocketController) HealthHandler(ctx *gin.Context) {
 func (c *WebSocketController) HandleWebSocket(ctx *gin.Context) {
 	start := time.Now()
 	clientIP := ctx.ClientIP()
-	log.Printf("👤👤👤👤 客户端连接中: IP=%s", clientIP)
+	log.Printf("👤👤👤👤👤👤👤👤 客户端连接中: IP=%s", clientIP)

 	conn, err := upgrader.Upgrade(ctx.Writer, ctx.Request, nil)
 	if err != nil {
@@ -188,7 +189,7 @@ func (c *WebSocketController) HandleWebSocket(ctx *gin.Context) {
 		messageType, message, err := conn.ReadMessage()
 		if err != nil {
 			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
-				log.Printf("❌❌❌❌ 连接意外断开: %v | ClientID=%s", err, clientID)
+				log.Printf("❌❌❌❌❌❌❌❌ 连接意外断开: %v | ClientID=%s", err, clientID)
 			} else {
 				log.Printf("⚠️ 连接正常关闭: ClientID=%s", clientID)
 			}
@@ -203,16 +204,16 @@ func (c *WebSocketController) HandleWebSocket(ctx *gin.Context) {

 // 处理客户端消息
 func (c *WebSocketController) handleClientMessage(senderID string, message []byte) {
-	log.Printf("📥📥📥📥 收到客户端消息: \n  SenderID=%s \n  Size=%d bytes", senderID, len(message))
+	log.Printf("📥📥📥📥📥📥📥📥 收到客户端消息: \n  SenderID=%s \n  Size=%d bytes", senderID, len(message))

 	var payload models.SendMessagePayload
 	if err := json.Unmarshal(message, &payload); err == nil && payload.RequestType != "" {
-		log.Printf("📦📦📦📦 解析JSON消息成功: Type=%s", payload.RequestType)
+		log.Printf("📦📦📦📦📦📦📦📦 解析JSON消息成功: Type=%s", payload.RequestType)

 		if payload.RequestType == "bind" && payload.SenderUserID != "" {
-			log.Printf("🔗🔗🔗🔗 处理绑定请求: \n  ClientID=%s \n  UserID=%s", senderID, payload.SenderUserID)
+			log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理绑定请求: \n  ClientID=%s \n  UserID=%s", senderID, payload.SenderUserID)
 			if err := c.bindClientToUser(senderID, payload.SenderUserID); err != nil {
-				log.Printf("❌❌❌❌ 绑定失败: %v", err)
+				log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err)
 			} else {
 				log.Printf("✅ 绑定成功: \n  ClientID=%s \n  UserID=%s", senderID, payload.SenderUserID)
 			}
@@ -221,7 +222,7 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt

 		if payload.RequestType == "send_message" {
 			if payload.TargetClientID != "" {
-				log.Printf("📨📨📨📨 处理客户端发起的发送请求: \n  TargetID=%s \n  MsgType=%d",
+				log.Printf("📨📨📨📨📨📨📨📨 处理客户端发起的发送请求: \n  TargetID=%s \n  MsgType=%d",
 					payload.TargetClientID, payload.MessageType)

 				clientMsg := models.ClientReceivedMessage{
@@ -239,7 +240,7 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt
 			}

 			if payload.ReceiverUserID != "" {
-				log.Printf("📨📨📨📨 处理客户端发起的用户发送请求: \n  ReceiverUserID=%s \n  MsgType=%d",
+				log.Printf("📨📨📨📨📨📨📨📨 处理客户端发起的用户发送请求: \n  ReceiverUserID=%s \n  MsgType=%d",
 					payload.ReceiverUserID, payload.MessageType)

 				clientMsg := models.ClientReceivedMessage{
@@ -260,7 +261,7 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt

 	var clientMsg models.ClientReceivedMessage
 	if err := json.Unmarshal(message, &clientMsg); err == nil && clientMsg.ReceiverID != "" {
-		log.Printf("📨📨📨📨 处理客户端封装消息: \n  TargetID=%s \n  MsgType=%d",
+		log.Printf("📨📨📨📨📨📨📨📨 处理客户端封装消息: \n  TargetID=%s \n  MsgType=%d",
 			clientMsg.ReceiverID, clientMsg.MessageType)

 		clientMsg.SenderID = senderID
@@ -268,16 +269,159 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt
 		return
 	}

+	// 修复：正确解析通话信令并分发处理
+	var callSignal models.CallSignal
+	if err := json.Unmarshal(message, &callSignal); err == nil && callSignal.Action != "" {
+		// 获取发送者用户ID
+		senderUserID, _ := c.getUserIDByClientID(senderID)
+
+		log.Printf("📞📞 处理通话信令: \n  Action=%s \n  CallID=%s \n  CallType=%d \n  SenderUserID=%s \n  CallerID=%s \n  CalleeID=%s",
+			callSignal.Action, callSignal.CallID, callSignal.CallType, senderUserID, callSignal.CallerID, callSignal.CalleeID)
+
+		// 处理不同类型的通话动作
+		switch callSignal.Action {
+		case "invite":
+			c.handleCallInvite(senderID, senderUserID, callSignal)
+		case "accept":
+			c.handleCallAccept(senderID, senderUserID, callSignal)
+		case "reject":
+			c.handleCallReject(senderID, senderUserID, callSignal)
+		case "end":
+			c.handleCallEnd(senderID, senderUserID, callSignal)
+		case "candidate":
+			c.handleCallCandidate(senderID, senderUserID, callSignal)
+		default:
+			log.Printf("⚠️ 未知的通话动作: %s", callSignal.Action)
+		}
+		return
+	}
+
 	log.Printf("⚠️ 无法识别的消息格式: \n  Size=%d bytes \n  Message=%s", len(message), string(message))
 }

+// 处理通话邀请
+func (c *WebSocketController) handleCallInvite(senderID, senderUserID string, signal models.CallSignal) {
+	log.Printf("📞 处理通话邀请: \n  CallID=%s \n  CallType=%d \n  From=%s \n  To=%s",
+		signal.CallID, signal.CallType, senderUserID, signal.CalleeID)
+
+	// 创建通话消息
+	callMsg := models.ClientReceivedMessage{
+		SenderID:       senderID,
+		ReceiverID:     "", // 通过用户ID发送
+		SenderUserID:   senderUserID,
+		ReceiverUserID: signal.CalleeID,
+		MessageType:    signal.CallType, // 6=视频, 7=语音
+		Content:        signal.Data,     // 包含SDP offer
+		SendTime:       time.Now().Format(time.RFC3339),
+		CallID:         signal.CallID,
+		CallStatus:     "invite",
+	}
+
+	// 发送邀请给被叫方
+	if err := c.sendMessageToUser(signal.CalleeID, callMsg); err != nil {
+		log.Printf("❌ 发送通话邀请失败: %v | CalleeID=%s", err, signal.CalleeID)
+	}
+}
+
+// 处理通话接受
+func (c *WebSocketController) handleCallAccept(senderID, senderUserID string, signal models.CallSignal) {
+	log.Printf("✅ 处理通话接受: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)
+
+	// 创建通话接受消息
+	acceptMsg := models.ClientReceivedMessage{
+		SenderID:       senderID,
+		ReceiverID:     "", // 通过用户ID发送
+		SenderUserID:   senderUserID,
+		ReceiverUserID: signal.CallerID,
+		MessageType:    signal.CallType, // 6=视频, 7=语音
+		Content:        signal.Data,     // 包含SDP answer
+		SendTime:       time.Now().Format(time.RFC3339),
+		CallID:         signal.CallID,
+		CallStatus:     "accepted",
+	}
+
+	// 发送接受消息给主叫方
+	if err := c.sendMessageToUser(signal.CallerID, acceptMsg); err != nil {
+		log.Printf("❌ 发送通话接受失败: %v | CallerID=%s", err, signal.CallerID)
+	}
+}
+
+// 处理通话拒绝
+func (c *WebSocketController) handleCallReject(senderID, senderUserID string, signal models.CallSignal) {
+	log.Printf("❌ 处理通话拒绝: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CallerID)
+
+	// 创建通话拒绝消息
+	rejectMsg := models.ClientReceivedMessage{
+		SenderID:       senderID,
+		ReceiverID:     "", // 通过用户ID发送
+		SenderUserID:   senderUserID,
+		ReceiverUserID: signal.CallerID,
+		MessageType:    signal.CallType, // 6=视频, 7=语音
+		Content:        signal.Data,     // 可包含拒绝原因
+		SendTime:       time.Now().Format(time.RFC3339),
+		CallID:         signal.CallID,
+		CallStatus:     "rejected",
+	}
+
+	// 发送拒绝消息给主叫方
+	if err := c.sendMessageToUser(signal.CallerID, rejectMsg); err != nil {
+		log.Printf("❌ 发送通话拒绝失败: %v | CallerID=%s", err, signal.CallerID)
+	}
+}
+
+// 处理通话结束
+func (c *WebSocketController) handleCallEnd(senderID, senderUserID string, signal models.CallSignal) {
+	log.Printf("📞 处理通话结束: \n  CallID=%s \n  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)
+
+	// 创建通话结束消息
+	endMsg := models.ClientReceivedMessage{
+		SenderID:       senderID,
+		ReceiverID:     "", // 通过用户ID发送
+		SenderUserID:   senderUserID,
+		ReceiverUserID: signal.CalleeID,
+		MessageType:    signal.CallType, // 6=视频, 7=语音
+		Content:        signal.Data,     // 可包含结束原因
+		SendTime:       time.Now().Format(time.RFC3339),
+		CallID:         signal.CallID,
+		CallStatus:     "ended",
+	}
+
+	// 发送结束消息给对端
+	if err := c.sendMessageToUser(signal.CalleeID, endMsg); err != nil {
+		log.Printf("❌ 发送通话结束失败: %v | CalleeID=%s", err, signal.CalleeID)
+	}
+}
+
+// 处理通话候选
+func (c *WebSocketController) handleCallCandidate(senderID, senderUserID string, signal models.CallSignal) {
+	log.Printf("📶 处理通话候选: \n  CallID=%s  From=%s \n  To=%s", signal.CallID, senderUserID, signal.CalleeID)
+
+	// 创建候选消息
+	candidateMsg := models.ClientReceivedMessage{
+		SenderID:       senderID,
+		ReceiverID:     "", // 通过用户ID发送
+		SenderUserID:   senderUserID,
+		ReceiverUserID: signal.CalleeID,
+		MessageType:    signal.CallType, // 6=视频, 7=语音
+		Content:        signal.Data,     // 包含ICE候选
+		SendTime:       time.Now().Format(time.RFC3339),
+		CallID:         signal.CallID,
+		CallStatus:     "candidate",
+	}
+
+	// 发送候选消息给对端
+	if err := c.sendMessageToUser(signal.CalleeID, candidateMsg); err != nil {
+		log.Printf("❌ 发送候选失败: %v | CalleeID=%s", err, signal.CalleeID)
+	}
+}
+
 // SendMessageHandler API消息发送处理器
 func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) {
 	start := time.Now()

 	var payload models.SendMessagePayload
 	if err := ctx.ShouldBindJSON(&payload); err != nil {
-		log.Printf("⚠️ 无效的JSON请求格式: %v", err)
+		log.Printf("⚠️ 无效的JSON请求格式: %极简", err)
 		ctx.JSON(http.StatusBadRequest, gin.H{"error": "无效的JSON格式"})
 		return
 	}
@@ -291,7 +435,7 @@ func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) {
 	}

 	if payload.RequestType == "send_message" {
-		log.Printf("📤📤📤📤 处理API发送请求: \n  SenderUser=%s \n  MsgType=%d",
+		log.Printf("📤📤📤📤📤📤📤📤 处理API发送请求: \n  SenderUser=%s \n  MsgType=%d",
 			payload.SenderUserID, payload.MessageType)

 		clientMsg := models.ClientReceivedMessage{
@@ -308,11 +452,11 @@ func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) {
 		var sendErr error

 		if payload.TargetClientID != "" {
-			log.Printf("🎯🎯🎯🎯 目标类型: ClientID | Target=%s", payload.TargetClientID)
+			log.Printf("🎯🎯🎯🎯🎯🎯🎯🎯 目标类型: ClientID | Target=%s", payload.TargetClientID)
 			sendErr = c.sendMessageToClient(payload.TargetClientID, clientMsg)
 			result = fmt.Sprintf("消息已发送到ClientID: %s", payload.TargetClientID)
 		} else if payload.ReceiverUserID != "" {
-			log.Printf("🎯🎯🎯🎯 目标类型: UserID | ReceiverUser=%s", payload.ReceiverUserID)
+			log.Printf("🎯🎯🎯🎯🎯🎯🎯🎯 目标类型: UserID | ReceiverUser=%s", payload.ReceiverUserID)
 			sendErr = c.sendMessageToUser(payload.ReceiverUserID, clientMsg)
 			result = fmt.Sprintf("消息已发送到UserID: %s", payload.ReceiverUserID)
 		} else {
@@ -333,7 +477,7 @@ func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) {
 	ctx.JSON(http.StatusBadRequest, gin.H{"error": "不支持的request_type"})
 }

-// BindHandler 用户绑定处理器
+// 用户绑定处理器
 func (c *WebSocketController) BindHandler(ctx *gin.Context) {
 	start := time.Now()

@@ -350,10 +494,10 @@ func (c *WebSocketController) BindHandler(ctx *gin.Context) {
 		return
 	}

-	log.Printf("🔗🔗🔗🔗 处理用户绑定请求: \n  UserID=%s \n  ClientID=%s", req.UserID, req.ClientID)
+	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 处理用户绑定请求: \n  UserID=%s \n  ClientID=%s", req.UserID, req.ClientID)

 	if err := c.bindClientToUser(req.ClientID, req.UserID); err != nil {
-		log.Printf("❌❌❌❌ 绑定失败: %v", err)
+		log.Printf("❌❌❌❌❌❌❌❌ 绑定失败: %v", err)
 		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
@@ -367,7 +511,7 @@ func (c *WebSocketController) BindHandler(ctx *gin.Context) {
 		req.UserID, req.ClientID, time.Since(start))
 }

-// SendToUserHandler 通过用户ID发送消息处理器
+// SendToUserHandler 通过用户ID发送消息处理器（支持通话信令）
 func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) {
 	start := time.Now()

@@ -385,9 +529,10 @@ func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) {
 		return
 	}

-	log.Printf("📤📤📤📤 处理API用户发送请求: \n  SenderUser=%s → ReceiverUser=%s \n  MsgType=%d",
+	log.Printf("📤📤📤📤📤📤📤📤 处理API用户发送请求: \n  SenderUser=%s → ReceiverUser=%s \n  MsgType=%d",
 		payload.SenderUserID, payload.ReceiverUserID, payload.MessageType)

+	// 创建基础消息结构
 	clientMsg := models.ClientReceivedMessage{
 		SenderID:       "system",
 		ReceiverID:     "",
@@ -398,6 +543,16 @@ func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) {
 		SendTime:       time.Now().Format(time.RFC3339),
 	}

+	// 如果是通话信令消息，添加通话相关字段
+	if payload.MessageType == 6 || payload.MessageType == 7 {
+		// 直接使用内容作为通话信令数据
+		clientMsg.CallStatus = "invite"           // 默认设置为invite，API调用通常是发起通话
+		clientMsg.CallID = utils.GenerateCallID() // 生成唯一的通话ID
+
+		log.Printf("📞📞 处理API通话信令: \n  CallID=%s \n  CallType=%d",
+			clientMsg.CallID, payload.MessageType)
+	}
+
 	if err := c.sendMessageToUser(payload.ReceiverUserID, clientMsg); err != nil {
 		ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
@@ -407,6 +562,7 @@ func (c *WebSocketController) SendToUserHandler(ctx *gin.Context) {
 		"code":    0,
 		"status":  "success",
 		"message": fmt.Sprintf("消息已发送至用户 %s", payload.ReceiverUserID),
+		"call_id": clientMsg.CallID, // 返回生成的call_id
 	})

 	log.Printf("✅ API请求完成: Duration=%s", time.Since(start))
@@ -428,14 +584,14 @@ func (c *WebSocketController) bindClientToUser(clientID, userID string) error {
 		log.Printf("⚠️ 设置过期时间失败: %v", err)
 	}

-	log.Printf("🔗🔗🔗🔗 用户绑定成功: \n  ClientID=%s → UserID=%s", clientID, userID)
+	log.Printf("🔗🔗🔗🔗🔗🔗🔗🔗 用户绑定成功: \n  ClientID=%s → UserID=%s", clientID, userID)
 	return nil
 }

 // 通过用户ID发送消息
 func (c *WebSocketController) sendMessageToUser(userID string, msg models.ClientReceivedMessage) error {
 	start := time.Now()
-	log.Printf("👤👤👤👤 通过用户ID发送消息: UserID=%s", userID)
+	log.Printf("👤👤👤👤👤👤👤👤 通过用户ID发送消息: UserID=%s", userID)

 	userClientKey := fmt.Sprintf("%s:%s", models.UserClientKey, userID)
 	clientIDs, err := c.RedisCli.SMembers(c.RedisCtx, userClientKey).Result()
@@ -447,7 +603,7 @@ func (c *WebSocketController) sendMessageToUser(userID string, msg models.Client
 		return fmt.Errorf("用户未绑定任何客户端")
 	}

-	log.Printf("📡📡📡📡 找到 %d 个关联的客户端: UserID=%s", len(clientIDs), userID)
+	log.Printf("📡📡📡📡📡📡📡 找到 %d 个关联的客户端: UserID=%s", len(clientIDs), userID)

 	var successCount, failCount int
 	var lastError error
@@ -457,7 +613,7 @@ func (c *WebSocketController) sendMessageToUser(userID string, msg models.Client
 		targetMsg.ReceiverID = clientID

 		if err := c.sendMessageToClient(clientID, targetMsg); err != nil {
-			log.Printf("❌❌❌❌ 发送消息到客户端失败: \n  ClientID=%s \n  %v", clientID, err)
+			log.Printf("❌❌❌❌❌❌❌❌ 发送消息到客户端失败: \n  ClientID=%s \n  %v", clientID, err)
 			failCount++
 			lastError = err
 		} else {
@@ -465,7 +621,7 @@ func (c *WebSocketController) sendMessageToUser(userID string, msg models.Client
 		}
 	}

-	log.Printf("📬📬📬📬 消息发送完成: \n  成功 %d \n  失败 %d \n  UserID=%s \n  Duration=%s",
+	log.Printf("📬📬📬📬📬📬📬📬 消息发送完成: \n  成功 %d \n  失败 %d \n  UserID=%s \n  Duration=%s",
 		successCount, failCount, userID, time.Since(start))

 	if failCount > 0 {
@@ -474,22 +630,26 @@ func (c *WebSocketController) sendMessageToUser(userID string, msg models.Client
 	return nil
 }

-// 发送消息给客户端
+// 发送消息给客户端（添加写锁保护）
 func (c *WebSocketController) sendMessageToClient(clientID string, msg models.ClientReceivedMessage) error {
 	start := time.Now()

 	msgJSON, err := json.Marshal(msg)
 	if err != nil {
-		log.Printf("❌❌❌❌ 消息序列化失败: %v", err)
+		log.Printf("❌❌❌❌❌❌❌❌ 消息序列化失败: %v", err)
 		return fmt.Errorf("内部错误")
 	}

 	if conn := c.getClient(clientID); conn != nil {
-		log.Printf("📤📤📤📤 向本地客户端发送消息: \n  ClientID=%s \n  MsgType=%d \n  Size=%d bytes",
+		log.Printf("📤📤📤📤📤📤📤📤 向本地客户端发送消息: \n  ClientID=%s \n  MsgType=%d \n  Size=%d bytes",
 			clientID, msg.MessageType, len(msgJSON))

+		// 加写锁保护
+		c.WriteMutex.Lock()
+		defer c.WriteMutex.Unlock()
+
 		if err := conn.WriteMessage(websocket.TextMessage, msgJSON); err != nil {
-			log.Printf("❌❌❌❌ 发送消息失败: %v | ClientID=%s", err, clientID)
+			log.Printf("❌❌❌极简❌❌❌❌ 发送消息失败: %v | ClientID=%s", err, clientID)
 			c.removeClient(clientID)
 			return fmt.Errorf("发送消息失败")
 		}
@@ -499,7 +659,7 @@ func (c *WebSocketController) sendMessageToClient(clientID string, msg models.Cl
 		return nil
 	}

-	log.Printf("🌐🌐🌐🌐 向远程节点转发消息: ClientID=%s", clientID)
+	log.Printf("🌐🌐🌐🌐🌐🌐🌐🌐 向远程节点转发消息: ClientID=%s", clientID)

 	redisMsg := models.RedisMessage{
 		SenderNodeID: c.NodeID,
@@ -509,16 +669,16 @@ func (c *WebSocketController) sendMessageToClient(clientID string, msg models.Cl

 	redisMsgJSON, err := json.Marshal(redisMsg)
 	if err != nil {
-		log.Printf("❌❌❌❌ Redis消息序列化失败: %v", err)
+		log.Printf("❌❌❌❌❌❌❌❌ Redis消息序列化失败: %v", err)
 		return fmt.Errorf("内部错误")
 	}

 	if err := c.RedisCli.Publish(c.RedisCtx, "ws_messages", redisMsgJSON).Err(); err != nil {
-		log.Printf("❌❌❌❌ 发布消息到Redis失败: %v", err)
+		log.Printf("❌❌❌❌❌❌❌❌ 发布消息到Redis失败: %v", err)
 		return fmt.Errorf("无法转发消息")
 	}

-	log.Printf("📡📡📡📡 消息已转发到Redis: \n  ClientID=%s \n  Size=%d bytes \n  Duration=%s",
+	log.Printf("📡📡📡📡📡📡📡📡 消息已转发到Redis: \n  ClientID=%s \n  Size=%d bytes \n  Duration=%s",
 		clientID, len(redisMsgJSON), time.Since(start))
 	return nil
 }
@@ -529,28 +689,33 @@ func (c *WebSocketController) SubscribeToRedis() {
 	defer pubsub.Close()

 	ch := pubsub.Channel()
-	log.Println("🔔🔔🔔🔔 开始监听Redis消息...")
+	log.Println("🔔🔔🔔🔔🔔🔔🔔🔔 开始监听Redis消息...")

 	for msg := range ch {
 		start := time.Now()
 		var redisMsg models.RedisMessage

 		if err := json.Unmarshal([]byte(msg.Payload), &redisMsg); err != nil {
-			log.Printf("❌❌❌❌ 解析Redis消息失败: %v", err)
+			log.Printf("❌❌❌❌❌❌❌❌ 解析Redis消息失败: %v", err)
 			continue
 		}

-		log.Printf("📥📥📥📥 收到Redis消息: \n  Sender=%s \n  ClientID=%s \n  Size=%d bytes",
+		log.Printf("📥📥📥📥📥📥📥📥 收到Redis消息: \n  Sender=%s \n  ClientID=%s \n  Size=%d bytes",
 			redisMsg.SenderNodeID, redisMsg.ClientID, len(msg.Payload))

 		if redisMsg.SenderNodeID == c.NodeID {
-			log.Println("ℹℹℹℹ️ 忽略本节点转发的消息")
+			log.Println("ℹℹℹℹℹℹℹℹ️ 忽略本节点转发的消息")
 			continue
 		}

 		if conn := c.getClient(redisMsg.ClientID); conn != nil {
-			if err := conn.WriteMessage(websocket.TextMessage, []byte(redisMsg.Message)); err != nil {
-				log.Printf("❌❌❌❌ 处理Redis消息失败: %v | ClientID=%s", err, redisMsg.ClientID)
+			// 加写锁保护
+			c.WriteMutex.Lock()
+			err := conn.WriteMessage(websocket.TextMessage, []byte(redisMsg.Message))
+			c.WriteMutex.Unlock()
+
+			if err != nil {
+				log.Printf("❌❌❌❌❌❌❌❌ 处理Redis消息失败: %v | ClientID=%s", err, redisMsg.ClientID)
 				c.removeClient(redisMsg.ClientID)
 				continue
 			}
@@ -570,7 +735,7 @@ func (c *WebSocketController) addClient(clientID string, conn *websocket.Conn) {
 	c.ClientsMux.Lock()
 	defer c.ClientsMux.Unlock()
 	c.Clients[clientID] = conn
-	log.Printf("📌📌 添加客户端到连接池: \n  ClientID=%s \n  当前连接数=%d",
+	log.Printf("📌📌📌📌 添加客户端到连接池: \n  ClientID=%s \n  当前连接数=%d",
 		clientID, len(c.Clients))
 }

@@ -584,7 +749,7 @@ func (c *WebSocketController) removeClient(clientID string) {
 	}

 	delete(c.Clients, clientID)
-	log.Printf("🗑🗑️ 从连接池移除客户端: \n  ClientID=%s \n  当前连接数=%d",
+	log.Printf("🗑🗑🗑🗑️ 从连接池移除客户端: \n  ClientID=%s \n  当前连接数=%d",
 		clientID, len(c.Clients))

 	if err := c.RedisCli.HDel(c.RedisCtx, models.ClientUserKey, clientID).Err(); err != nil {
@@ -598,3 +763,14 @@ func (c *WebSocketController) getClient(clientID string) *websocket.Conn {
 	defer c.ClientsMux.RUnlock()
 	return c.Clients[clientID]
 }
+
+// 通过客户端ID获取用户ID
+func (c *WebSocketController) getUserIDByClientID(clientID string) (string, error) {
+	userID, err := c.RedisCli.HGet(c.RedisCtx, models.ClientUserKey, clientID).Result()
+	if err == redis.Nil {
+		return "", fmt.Errorf("客户端未绑定用户")
+	} else if err != nil {
+		return "", fmt.Errorf("获取用户ID失败: %w", err)
+	}
+	return userID, nil
+}