拆分文件

2025-07-03 09:16:16 +08:00
parent 863adfa7d0
commit fb3918124d
7 changed files with 205 additions and 164 deletions
--- a/.env
+++ b/.env
@@ -1,5 +1,9 @@
-[Redis]
+# [Redis]
 REDIS_ADDR=127.0.0.1:6379
+REDIS_PASSWORD=

-[App]
-PORT=12080
+# [App]
+PORT=12080
+
+# [Node] 用于指定节点
+NODE_ID=ecs1
--- 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
-}
--- 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
 )
--- 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=
--- a/utils/env.go
+++ 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
+}
--- a/utils/id_generator.go
+++ 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())
+}
--- a/utils/logger.go
+++ 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
+}