feat: 优化WebSocket设备状态订阅，使用Redis Pub/Sub替代MQTT，简化架构并提升性能

Author: hezhuozhuo

docs/demand-community/2025.10.16-WebSocket设备状态订阅优化方案.md

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strconv"
 	"sync"
 	"time"
@@ -457,52 +458,99 @@ func (*TelemetryDataApi) ServeDeviceStatusByWS(c *gin.Context) {
 
 	logrus.Infof("WebSocket连接已建立 - 设备ID: %s, 用户ID: %s, 租户ID: %s", deviceID, claims.ID, claims.TenantID)
 
-	// 订阅设备在线状态
-	var mu sync.Mutex
-	logrus.Info("User SubscribeOnlineOffline")
-	var mqttClient ws_subscribe.WsMqttClient
-	if err := mqttClient.SubscribeOnlineOffline(deviceID, conn, msgType, &mu); err != nil {
-		logrus.Error("订阅设备状态失败:", err)
-		conn.WriteMessage(msgType, []byte("Failed to subscribe to device status"))
+	// 查询设备当前状态并立即推送 (保持原有格式: is_online 为整数 0/1)
+	currentStatusMap, err := service.GroupApp.Device.GetDeviceOnlineStatus(deviceID)
+	if err != nil {
+		logrus.WithError(err).Error("查询设备当前状态失败")
+		conn.WriteMessage(msgType, []byte("Failed to query device status"))
 		return
 	}
-	defer mqttClient.Close()
 
-	// 处理心跳
+	// 提取在线状态 (整数格式)
+	isOnline := 0
+	if status, ok := currentStatusMap["is_online"]; ok {
+		isOnline = status
+	}
+
+	// 发送当前状态 (简化格式,与原有接口保持一致)
+	initialMsg := map[string]interface{}{
+		"is_online": isOnline,
+	}
+	if data, err := json.Marshal(initialMsg); err == nil {
+		conn.WriteMessage(msgType, data)
+	}
+
+	// 订阅 Redis Pub/Sub 通道
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	channel := fmt.Sprintf("device:%s:status", deviceID)
+	pubsub := global.REDIS.Subscribe(ctx, channel)
+	defer pubsub.Close()
+
+	logrus.Infof("WebSocket已订阅Redis通道: %s", channel)
+
+	// 使用 sync.Mutex 保护 WebSocket 写操作
+	var mu sync.Mutex
+
+	// Goroutine 1: Redis消息转发到WebSocket
+	go func() {
+		ch := pubsub.Channel()
+		for {
+			select {
+			case <-ctx.Done():
+				logrus.Debug("Redis订阅上下文取消")
+				return
+			case redisMsg, ok := <-ch:
+				if !ok {
+					logrus.Warn("Redis通道已关闭")
+					return
+				}
+
+				// 转发到WebSocket
+				mu.Lock()
+				err := conn.WriteMessage(msgType, []byte(redisMsg.Payload))
+				mu.Unlock()
+
+				if err != nil {
+					logrus.WithError(err).Error("WebSocket写入失败")
+					cancel() // 通知主goroutine退出
+					return
+				}
+
+				logrus.WithField("device_id", deviceID).Debug("状态更新已推送到WebSocket")
+			}
+		}
+	}()
+
+	// Goroutine 2 (主): WebSocket心跳处理
 	for {
-		_, msg, err := conn.ReadMessage()
+		_, wsMsg, err := conn.ReadMessage()
 		if err != nil {
-			// 记录错误日志
-			logrus.Error("WebSocket读取错误:", err)
+			logrus.WithError(err).Info("WebSocket连接关闭")
 
-			// 尝试发送错误消息给客户端
-			closeMsg := []byte("connection closed due to error")
-			// 使用 WriteControl 发送关闭消息，设置1秒超时
+			// 发送关闭消息
+			closeMsg := []byte("connection closed")
 			deadline := time.Now().Add(time.Second)
 			conn.WriteControl(websocket.CloseMessage,
-				websocket.FormatCloseMessage(websocket.CloseInternalServerErr, string(closeMsg)),
+				websocket.FormatCloseMessage(websocket.CloseNormalClosure, string(closeMsg)),
 				deadline)
 
-			// 现在可以安全退出了
+			cancel() // 通知Redis订阅goroutine退出
 			return
 		}
 
 		// 处理心跳消息
-		if string(msg) == "ping" {
+		if string(wsMsg) == "ping" {
 			mu.Lock()
-			if err := conn.WriteMessage(msgType, []byte("pong")); err != nil {
-				logrus.Error("发送pong消息失败:", err)
-
-				// 尝试发送错误消息
-				deadline := time.Now().Add(time.Second)
-				conn.WriteControl(websocket.CloseMessage,
-					websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "failed to send pong"),
-					deadline)
+			err := conn.WriteMessage(msgType, []byte("pong"))
+			mu.Unlock()
 
-				mu.Unlock()
+			if err != nil {
+				logrus.WithError(err).Error("发送pong失败")
+				cancel()
 				return
 			}
-			mu.Unlock()
 		}
 	}
 }

internal/api/telemetry_data.go

@@ -162,13 +162,16 @@ func (f *StatusFlow) processMessage(msg *DeviceMessage) {
 	// 5. 清理设备缓存
 	initialize.DelDeviceCache(device.ID)
 
-	// 6. SSE 通知客户端
+	// 6. 发布到 Redis Pub/Sub (供 WebSocket 订阅)
+	go f.publishToRedis(device, status, msg.Metadata)
+
+	// 7. SSE 通知客户端
 	go f.notifyClients(device, status)
 
-	// 7. 触发自动化
+	// 8. 触发自动化
 	go f.triggerAutomation(device, status)
 
-	// 8. 预期数据发送(上线时)
+	// 9. 预期数据发送(上线时)
 	if status == 1 {
 		go f.sendExpectedData(device)
 	}
@@ -275,3 +278,51 @@ func (f *StatusFlow) sendExpectedData(device *model.Device) {
 		f.logger.WithField("device_id", device.ID).Debug("Expected data sent")
 	}
 }
+
+// publishToRedis 发布设备状态到 Redis Pub/Sub (供 WebSocket 订阅)
+func (f *StatusFlow) publishToRedis(device *model.Device, status int16, metadata map[string]interface{}) {
+	// 构造设备名称
+	var deviceName string
+	if device.Name != nil {
+		deviceName = *device.Name
+	} else {
+		deviceName = device.DeviceNumber
+	}
+
+	// 获取来源信息
+	source, _ := metadata["source"].(string)
+	if source == "" {
+		source = "unknown"
+	}
+
+	// 构造消息 (保持原有 WebSocket 接口格式: is_online 为整数 0/1)
+	messageData := map[string]interface{}{
+		"is_online": int(status), // 0 或 1
+	}
+
+	jsonBytes, err := json.Marshal(messageData)
+	if err != nil {
+		f.logger.WithError(err).Error("Failed to marshal Redis message")
+		return
+	}
+
+	// 发布到设备专属通道: device:{device_id}:status
+	channel := fmt.Sprintf("device:%s:status", device.ID)
+	if err := global.REDIS.Publish(f.ctx, channel, string(jsonBytes)).Err(); err != nil {
+		// 记录错误但不阻塞主流程
+		f.logger.WithError(err).WithFields(logrus.Fields{
+			"device_id": device.ID,
+			"channel":   channel,
+			"status":    status,
+		}).Error("Failed to publish to Redis")
+		return
+	}
+
+	f.logger.WithFields(logrus.Fields{
+		"device_id":   device.ID,
+		"device_name": deviceName,
+		"channel":     channel,
+		"status":      status,
+		"source":      source,
+	}).Debug("Status published to Redis")
+}

internal/app/manager.go

@@ -21,6 +21,7 @@ import (
 // @name                        x-token
 func main() {
 	// 解析命令行参数
+	// go run . -config ./configs/config.yaml
 	configPath := flag.String("config", "", "配置文件路径...")
 	flag.Parse()