lq/nl-im-websocket-demo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

打包、拆分

Browse Source
This commit is contained in:
年糕崽崽
2025-07-03 12:49:49 +08:00
parent fb3918124d
commit cb79e11f9b
7 changed files with 439 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
controller/websocket_controller.go
View File

@@ -13,8 +13,8 @@ import (
"github.com/gin-gonic/gin"
"github.com/go-redis/redis/v8"
"github.com/gorilla/websocket"
"xk-websocket/models" // 替换为实际路径
"xk-websocket/utils" // 导入新的utils包
"xk-websocket/models"
"xk-websocket/utils"
)
var upgrader = websocket.Upgrader{
@@ -46,17 +46,51 @@ func NewWebSocketController() *WebSocketController {
// ConfigureSystem 配置系统参数
func (c *WebSocketController) ConfigureSystem() {
c.NodeID = utils.GetEnv("NODE_ID", "local")
c.Port = utils.GetEnv("PORT", "12080")
// 只有当端口未设置时才从环境变量获取
if c.Port == "" {
c.Port = utils.GetEnv("PORT", "12080")
log.Printf("📡📡📡📡📡📡📡📡 使用环境变量设置端口: %s", c.Port)
}
log.SetPrefix(fmt.Sprintf("[Node:%s] ", c.NodeID))
log.SetFlags(log.LstdFlags | log.Lmicroseconds)
c.configureLogger()
// 必须先初始化 Redis 客户端
c.InitRedisClient()
// 注册节点到Redis
go c.registerNode() // 现在 RedisCli 已初始化
}
// 注册节点到Redis
func (c *WebSocketController) registerNode() {
// 添加空值检查
if c.RedisCli == nil {
log.Println("⚠️ Redis客户端未初始化，节点注册已取消")
return
}
for {
key := fmt.Sprintf("websocket:nodes:%s", c.NodeID)
value := fmt.Sprintf("%s:%s", utils.GetOutboundIP(), c.Port)
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)
}
time.Sleep(20 * time.Second)
}
}
// 配置日志记录器
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")
@@ -84,7 +118,7 @@ func (c *WebSocketController) InitRedisClient() {
})
if err := c.checkRedisConnection(); err != nil {
log.Fatalf("❌❌ Redis连接失败: %v", err)
log.Fatalf("❌❌❌❌ Redis连接失败: %v", err)
}
}
@@ -97,15 +131,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.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:视频通话")
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 健康检查处理器
@@ -113,6 +148,7 @@ func (c *WebSocketController) HealthHandler(ctx *gin.Context) {
ctx.JSON(http.StatusOK, gin.H{
"status": "ok",
"node": c.NodeID,
"port": c.Port,
"time": time.Now().Format(time.RFC3339),
})
}
@@ -121,7 +157,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 {
@@ -152,7 +188,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)
}
@@ -167,16 +203,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)
}
@@ -185,7 +221,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{
@@ -203,7 +239,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{
@@ -224,7 +260,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
@@ -255,7 +291,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{
@@ -272,11 +308,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 {
@@ -314,10 +350,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
}
@@ -349,7 +385,7 @@ 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{
@@ -392,14 +428,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()
@@ -411,7 +447,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
@@ -421,7 +457,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 {
@@ -429,7 +465,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 {
@@ -444,16 +480,16 @@ func (c *WebSocketController) sendMessageToClient(clientID string, msg models.Cl
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))
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("发送消息失败")
}
@@ -463,7 +499,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,
@@ -473,16 +509,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
}
@@ -493,28 +529,28 @@ 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)
log.Printf("❌❌❌❌ 处理Redis消息失败: %v | ClientID=%s", err, redisMsg.ClientID)
c.removeClient(redisMsg.ClientID)
continue
}