diff --git a/.env b/.env index 23a439e..d3ad9ed 100644 --- a/.env +++ b/.env @@ -1,5 +1,9 @@ -[Redis] +# [Redis] REDIS_ADDR=127.0.0.1:6379 +REDIS_PASSWORD= -[App] -PORT=12080 \ No newline at end of file +# [App] +PORT=12080 + +# [Node] 用于指定节点 +NODE_ID=ecs1 \ No newline at end of file diff --git a/controller/websocket_controller.go b/controller/websocket_controller.go index da07349..0e4af5a 100644 --- a/controller/websocket_controller.go +++ b/controller/websocket_controller.go @@ -12,9 +12,9 @@ import ( "github.com/gin-gonic/gin" "github.com/go-redis/redis/v8" - "github.com/google/uuid" "github.com/gorilla/websocket" "xk-websocket/models" // 替换为实际路径 + "xk-websocket/utils" // 导入新的utils包 ) var upgrader = websocket.Upgrader{ @@ -25,7 +25,7 @@ var upgrader = websocket.Upgrader{ }, } -// WebSocket控制器结构体 +// WebSocketController WebSocket控制器结构体 type WebSocketController struct { Clients map[string]*websocket.Conn ClientsMux sync.RWMutex @@ -43,10 +43,10 @@ func NewWebSocketController() *WebSocketController { } } -// 配置系统参数 +// ConfigureSystem 配置系统参数 func (c *WebSocketController) ConfigureSystem() { - c.NodeID = c.getEnv("NODE_ID", "local") - c.Port = c.getEnv("PORT", "12080") + c.NodeID = utils.GetEnv("NODE_ID", "local") + c.Port = utils.GetEnv("PORT", "12080") log.SetPrefix(fmt.Sprintf("[Node:%s] ", c.NodeID)) log.SetFlags(log.LstdFlags | log.Lmicroseconds) c.configureLogger() @@ -54,12 +54,12 @@ func (c *WebSocketController) ConfigureSystem() { // 配置日志记录器 func (c *WebSocketController) configureLogger() { - logDir := c.getEnv("LOG_DIR", "./logs") + logDir := utils.GetEnv("LOG_DIR", "./logs") if err := os.MkdirAll(logDir, 0755); err != nil { - log.Fatalf("❌ 创建日志目录失败: %v", err) + log.Fatalf("❌❌ 创建日志目录失败: %v", err) } - logFreq := c.getEnv("LOG_ROTATE_FREQ", "daily") + logFreq := utils.GetEnv("LOG_ROTATE_FREQ", "daily") if c.Logger == nil { c.Logger = log.New(os.Stdout, "", log.LstdFlags|log.Lmicroseconds) } @@ -69,19 +69,14 @@ func (c *WebSocketController) configureLogger() { } // 获取日志文件写入器 -func (c *WebSocketController) getLogWriter(logDir, freq string) *dailyFileWriter { - return &dailyFileWriter{ - logDir: logDir, - frequency: freq, - logger: c.Logger, - nodeID: c.NodeID, - } +func (c *WebSocketController) getLogWriter(logDir, freq string) *utils.DailyFileWriter { + return utils.NewDailyFileWriter(logDir, freq, c.Logger, c.NodeID) } -// 初始化Redis客户端 +// InitRedisClient 初始化Redis客户端 func (c *WebSocketController) InitRedisClient() { - redisAddr := c.getEnv("REDIS_ADDR", "localhost:6379") - redisPassword := c.getEnv("REDIS_PASSWORD", "") + redisAddr := utils.GetEnv("REDIS_ADDR", "localhost:6379") + redisPassword := utils.GetEnv("REDIS_PASSWORD", "") c.RedisCli = redis.NewClient(&redis.Options{ Addr: redisAddr, Password: redisPassword, @@ -89,7 +84,7 @@ func (c *WebSocketController) InitRedisClient() { }) if err := c.checkRedisConnection(); err != nil { - log.Fatalf("❌ Redis连接失败: %v", err) + log.Fatalf("❌❌ Redis连接失败: %v", err) } } @@ -99,21 +94,21 @@ func (c *WebSocketController) checkRedisConnection() error { return err } -// 打印启动信息 +// PrintStartupInfo 打印启动信息 func (c *WebSocketController) PrintStartupInfo() { hostname, _ := os.Hostname() - log.Printf("🚀 WebSocket服务启动: NodeID=%s", c.NodeID) - log.Printf("🌐 监听端口: %s", c.Port) - log.Printf("📡 Redis地址: %s", c.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.Println("🔗🔗 等待客户端连接...") } -// 健康检查处理器 +// HealthHandler 健康检查处理器 func (c *WebSocketController) HealthHandler(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{ "status": "ok", @@ -122,11 +117,11 @@ func (c *WebSocketController) HealthHandler(ctx *gin.Context) { }) } -// WebSocket连接处理器 +// HandleWebSocket WebSocket连接处理器 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 { @@ -136,7 +131,7 @@ func (c *WebSocketController) HandleWebSocket(ctx *gin.Context) { } defer conn.Close() - clientID := c.generateClientID() + clientID := utils.GenerateClientID(c.NodeID) log.Printf("✅ 客户端已连接: \n ClientID=%s \n IP=%s \n Duration=%s", clientID, clientIP, time.Since(start)) @@ -157,7 +152,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) } @@ -172,16 +167,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) } @@ -190,7 +185,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{ @@ -208,7 +203,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{ @@ -229,7 +224,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 @@ -240,7 +235,7 @@ func (c *WebSocketController) handleClientMessage(senderID string, message []byt log.Printf("⚠️ 无法识别的消息格式: \n Size=%d bytes \n Message=%s", len(message), string(message)) } -// API消息发送处理器 +// SendMessageHandler API消息发送处理器 func (c *WebSocketController) SendMessageHandler(ctx *gin.Context) { start := time.Now() @@ -260,7 +255,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{ @@ -277,11 +272,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 { @@ -319,10 +314,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 } @@ -354,7 +349,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{ @@ -397,14 +392,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() @@ -416,7 +411,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 @@ -426,7 +421,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 { @@ -434,7 +429,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 { @@ -449,16 +444,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("发送消息失败") } @@ -468,7 +463,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, @@ -478,48 +473,48 @@ 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 } -// 订阅Redis消息 +// SubscribeToRedis 订阅Redis消息 func (c *WebSocketController) SubscribeToRedis() { pubsub := c.RedisCli.Subscribe(c.RedisCtx, "ws_messages") 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 } @@ -534,17 +529,12 @@ func (c *WebSocketController) SubscribeToRedis() { // ========================== 辅助方法 ========================== // -// 生成客户端ID -func (c *WebSocketController) generateClientID() string { - return fmt.Sprintf("%s-%s", c.NodeID, uuid.New().String()) -} - // 添加客户端到映射 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)) } @@ -558,7 +548,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 { @@ -572,88 +562,3 @@ func (c *WebSocketController) getClient(clientID string) *websocket.Conn { defer c.ClientsMux.RUnlock() return c.Clients[clientID] } - -// 获取环境变量 -func (c *WebSocketController) getEnv(key, defaultValue string) string { - if value := os.Getenv(key); value != "" { - return value - } - return defaultValue -} - -// ========================== 日志文件处理 ========================== // - -// 每日日志文件写入器 -type dailyFileWriter struct { - currentFile *os.File - mu sync.Mutex - logDir string - frequency string // "daily" 或 "10min" - currentDate string - currentSlot int // 10分钟时间段序号 - logger *log.Logger - nodeID string -} - -// 实现Write接口 -func (w *dailyFileWriter) Write(p []byte) (n int, err error) { - w.mu.Lock() - defer w.mu.Unlock() - - now := time.Now() - if w.needRotate(now) { - if err := w.rotateFile(now); err != nil { - w.logger.Printf("❌ 日志文件切换失败: %v", err) - return 0, err - } - } - - if w.currentFile == nil { - if err := w.rotateFile(now); err != nil { - return 0, err - } - } - - return w.currentFile.Write(p) -} - -// 检查是否需要切换文件 -func (w *dailyFileWriter) needRotate(now time.Time) bool { - currentDate := now.Format("2006-01-02") - if w.frequency == "daily" { - return w.currentDate != currentDate - } else if w.frequency == "10min" { - currentSlot := (now.Hour()*60 + now.Minute()) / 10 - return w.currentDate != currentDate || w.currentSlot != currentSlot - } - return false -} - -// 切换日志文件 -func (w *dailyFileWriter) rotateFile(now time.Time) error { - if w.currentFile != nil { - w.currentFile.Close() - } - - dateStr := now.Format("2006-01-02") - var slotStr string - if w.frequency == "10min" { - slot := (now.Hour()*60 + now.Minute()) / 10 - slotStr = fmt.Sprintf("%03d", slot) - } else { - slotStr = "000" - } - - filename := fmt.Sprintf("%s/xk-websocket-%s-%s.log", w.logDir, dateStr, slotStr) - file, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644) - if err != nil { - return fmt.Errorf("打开日志文件失败: %w", err) - } - - w.currentFile = file - w.currentDate = dateStr - if w.frequency == "10min" { - w.currentSlot = (now.Hour()*60 + now.Minute()) / 10 - } - return nil -} diff --git a/go.mod b/go.mod index 46d15ba..33e845d 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-redis/redis/v8 v8.11.5 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 + github.com/joho/godotenv v1.5.1 ) require ( @@ -32,10 +33,10 @@ require ( github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect golang.org/x/arch v0.8.0 // indirect - golang.org/x/crypto v0.23.0 // indirect + golang.org/x/crypto v0.34.1-0.20250224173925-7292932d45d5 // indirect golang.org/x/net v0.25.0 // indirect - golang.org/x/sys v0.20.0 // indirect - golang.org/x/text v0.15.0 // indirect + golang.org/x/sys v0.30.0 // indirect + golang.org/x/text v0.22.0 // indirect google.golang.org/protobuf v1.34.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 0dc3728..aa92e35 100644 --- a/go.sum +++ b/go.sum @@ -40,6 +40,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= @@ -86,14 +88,20 @@ golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc= golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys= golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI= golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= +golang.org/x/crypto v0.34.1-0.20250224173925-7292932d45d5 h1:/JE/WBkjH62TPhYxsJp9TGP+f4OYToEHf29iE5Mw48Q= +golang.org/x/crypto v0.34.1-0.20250224173925-7292932d45d5/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ= golang.org/x/net v0.25.0 h1:d/OCCoBEUq33pjydKrGQhw7IlUPI2Oylr+8qLx49kac= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= +golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk= golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM= +golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.34.1 h1:9ddQBjfCyZPOHPUiPxpYESBLc+T8P3E+Vo4IbKZgFWg= diff --git a/utils/env.go b/utils/env.go new file mode 100644 index 0000000..a9bb410 --- /dev/null +++ b/utils/env.go @@ -0,0 +1,20 @@ +package utils + +import ( + "github.com/joho/godotenv" + "log" + "os" +) + +func GetEnv(key, defaultValue string) string { + + // 加载环境变量 + if err := godotenv.Load(".env"); err != nil { + log.Fatal("Error loading .env file: ", err) + } + log.Printf("--------------------测试【"+key+"】", os.Getenv(key)) + if value := os.Getenv(key); value != "" { + return value + } + return defaultValue +} diff --git a/utils/id_generator.go b/utils/id_generator.go new file mode 100644 index 0000000..ca8e99a --- /dev/null +++ b/utils/id_generator.go @@ -0,0 +1,10 @@ +package utils + +import ( + "fmt" + "github.com/google/uuid" +) + +func GenerateClientID(nodeID string) string { + return fmt.Sprintf("%s-%s", nodeID, uuid.New().String()) +} diff --git a/utils/logger.go b/utils/logger.go new file mode 100644 index 0000000..1043794 --- /dev/null +++ b/utils/logger.go @@ -0,0 +1,93 @@ +package utils + +import ( + "fmt" + "log" + "os" + "sync" + "time" +) + +type DailyFileWriter struct { + currentFile *os.File + mu sync.Mutex + logDir string + frequency string + currentDate string + currentSlot int + logger *log.Logger + nodeID string +} + +func NewDailyFileWriter(logDir, frequency string, logger *log.Logger, nodeID string) *DailyFileWriter { + return &DailyFileWriter{ + logDir: logDir, + frequency: frequency, + logger: logger, + nodeID: nodeID, + } +} + +func (w *DailyFileWriter) Write(p []byte) (n int, err error) { + w.mu.Lock() + defer w.mu.Unlock() + + now := time.Now() + if w.needRotate(now) { + if err := w.rotateFile(now); err != nil { + w.logger.Printf("❌❌ 日志文件切换失败: %v", err) + return 0, err + } + } + + if w.currentFile == nil { + if err := w.rotateFile(now); err != nil { + return 0, err + } + } + + return w.currentFile.Write(p) +} + +func (w *DailyFileWriter) needRotate(now time.Time) bool { + currentDate := now.Format("2006-01-02") + if w.frequency == "daily" { + return w.currentDate != currentDate + } else if w.frequency == "10min" { + currentSlot := (now.Hour()*60 + now.Minute()) / 10 + return w.currentDate != currentDate || w.currentSlot != currentSlot + } + return false +} + +func (w *DailyFileWriter) rotateFile(now time.Time) error { + if w.currentFile != nil { + w.currentFile.Close() + } + + dateStr := now.Format("2006-01-02") + var slotStr string + if w.frequency == "10min" { + slot := (now.Hour()*60 + now.Minute()) / 10 + slotStr = fmt.Sprintf("%03d", slot) + } else { + slotStr = "000" + } + + if err := os.MkdirAll(w.logDir, 0755); err != nil { + return fmt.Errorf("创建日志目录失败: %w", err) + } + + filename := fmt.Sprintf("%s/xk-websocket-%s-%s.log", w.logDir, dateStr, slotStr) + file, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644) + if err != nil { + return fmt.Errorf("打开日志文件失败: %w", err) + } + + w.currentFile = file + w.currentDate = dateStr + if w.frequency == "10min" { + w.currentSlot = (now.Hour()*60 + now.Minute()) / 10 + } + return nil +}