线拔了，状态还亮着：SagooIoT 设备在线判定的过期键、两级缓存与上下线链路-育师

把一台网关的网线拔掉，然后盯着 SagooIoT 的设备列表。

它不会立刻变灰。它会先亮着，亮满两分钟，然后才翻成"离线"——在翻之前那几秒里，日志里甚至可能还写着它"刚上线"。

这不是延迟，也不是缺陷。设备"在线"这件事在物联网平台里从来不是被"知道"的，而是被推断出来的：平台没有握着设备的手，它只能靠最后一个信号有多新来猜。猜的方式决定了三件事——掉线要多久被发现、几万台设备同时心跳时后端扛不扛得住、以及一次误判会不会同时引爆告警、场景联动和设备升级。

这篇文章拆的就是这套推断机制：一条带 TTL 的键、一层进程内的本地缓存、一个 Redis 过期事件的订阅，以及上下线之后被串起来的那四件下游动作。代码来自本地专业版源码的pkg/dcache与pkg/redisDb。

一、为什么不用一张心跳表

最直觉的做法是建一张device_heartbeat表，每次收到报文UPDATE ... SET last_seen = now()，要判断在线就WHERE last_seen > now() - interval '120 seconds'。

这个方案在小规模下完全能用，但有两个绕不过去的成本。

第一是写入放大。一台设备 30 秒一次心跳，一万台就是每秒 333 次 UPDATE，而且每一行都被反复改写，索引维护、WAL、MVCC 垃圾回收全都压在这一张表上。心跳本身不含任何业务价值，它只是为了回答一个是/否的问题。

第二是判定成本。列表页要展示在线状态，就得对每台设备做一次时间比较；统计在线数量，就得全表扫一遍时间窗口。设备规模越大，这个查询越贵，而它被调用的频率恰恰和用户刷页面的频率正相关。

SagooIoT 换了个思路：不存"最后活跃时间"，而是存"这条记录还能活多久"。状态不是算出来的，是由过期这件事本身表达出来的。

方案	写入形态	判定成本	掉线发现时延
心跳表 + 时间窗口	每次心跳 UPDATE	O(设备数) 聚合查询	取决于查询时点，最多一个心跳周期
带 TTL 的状态键	每条报文一次 EXPIRE	O(1) EXISTS	由 TTL 到期事件驱动，秒级

二、判定在线，其实是一条会过期的键

状态键的命名规则很朴素：

// internal/consts/dcache.go const ( // 设备在线状态超时，120秒 DeviceOnlineTimeOut = 120 // 设备状态缓存KEY前缀 DeviceStatusPrefix = "deviceStatus:" ) const ( DeviceStatueOffline = 1 DeviceStatueOnline = 2 )

deviceStatus:<deviceKey>存在就是在线，不存在就是离线。没有第三个状态，也没有"未知"。

读状态的方法短到可以整段贴出来：

// GetDeviceStatus 获取指定的设备状态 func GetDeviceStatus(ctx context.Context, deviceKey string) (res int) { // 从本地缓存中获取数据 if cachedData, found := localCache.Get(consts.DeviceStatusPrefix + deviceKey); found { if cachedData != nil { return 2 } return 1 } // 本地缓存中不存在数据，从 Redis 中获取 data, err := cache.Instance().Get(ctx, consts.DeviceStatusPrefix+deviceKey) if err != nil || data == nil { return 1 } if data.Val() != nil { // 将数据写入本地缓存，有效期为10秒。 localCache.Set(consts.DeviceStatusPrefix+deviceKey, data.Val(), 10*time.Second) return 2 } return 1 }

localCache是一个patrickmn/go-cache实例，全包只有一个：

// 创建一个全局的本地缓存，用于缓解高并发访问redis数据库 var localCache = myCache.New(5*time.Minute, 10*time.Minute)

注意这里有个反直觉的地方：在线的键值（time.Now()）本身毫无意义，值是什么根本不重要。它唯一的职责是让键存在。所以这里真正的信息只有"有没有"，data.Val()一旦非 nil 就直接返回 2，不做任何解析。

状态写入有两个分支，分水岭是GetDeviceStatus的返回值：

func UpdateStatus(ctx context.Context, device *model.DeviceOutput) { if device == nil { return } timeout, err := GetTimeout(ctx, device.OnlineTimeout) if err != nil { g.Log().Debug(ctx, "获取设备超时配置失败", err) return } if timeout <= 0 { timeout = 60 } deviceStatus := dcache.GetDeviceStatus(ctx, device.Key) if deviceStatus == consts.DeviceStatueOffline { // 原来不在线，写一条全新的键，并处理"上线" err := redisDb.DB().WithLazyConn(ctx, func(conn *redis.Conn) error { return conn.SetEx(ctx, consts.DeviceStatusPrefix+device.Key, time.Now(), time.Duration(timeout)*time.Second).Err() }) if err != nil { g.Log().Debug(ctx, "更新设备状态失败", err) return } if err := Online(ctx, device); err != nil { g.Log().Debug(ctx, device.Key, "设备上线处理失败") } return } // 已经在线，只把过期时间往后推 err = redisDb.DB().WithLazyConn(ctx, func(conn *redis.Conn) error { _, err := conn.Expire(ctx, consts.DeviceStatusPrefix+device.Key, time.Duration(timeout)*time.Second).Result() return err }) if err != nil { g.Log().Debugf(ctx, "更新设备状态过期时间失败:%v", err) } }

这是整套机制的核心：高频路径只做EXPIRE，不做SET。

SETEX会覆盖值、重置 TTL、在 Redis 的 AOF/RDB 里留下一次写；EXPIRE只改元数据，且对已存在的键几乎无副作用。更重要的是，EXPIRE不会触发过期事件，而SETEX到期的旧键会——如果每条报文都SETEX，等于反复重建这个键，过期事件的语义就被搅乱了。区分 SET 与 EXPIRE，是为了让"键过期"这件事只在真正掉线时发生一次。

超时时间本身有三层来源：

// GetTimeout 获取设备超时时间 func GetTimeout(ctx context.Context, onlineTimeout int) (int, error) { if onlineTimeout > 0 { return onlineTimeout, nil } // 获取系统配置的默认超时时间 defaultTimeout, err := service.ConfigData().GetConfigByKey(ctx, consts.DeviceDefaultTimeoutTime) if err != nil || defaultTimeout == nil { return 120, nil } else { timeout := gconv.Int(defaultTimeout.ConfigValue) if timeout <= 0 { timeout = 120 } return timeout, nil } }

优先级顺序是：设备自身的OnlineTimeout→ 系统参数device.default.timeout.time→ 硬编码 120 秒。这个顺序是对的——不同协议的心跳周期不一样，NB-IoT 的设备两小时才醒一次，用统一的 120 秒会把它们全部判死；而掉线告警又必须有一个兜底默认值，否则配置一空，判定就失效。

三、读状态要读两次：5 分钟的本地缓存与 10 秒的在线快照

localCache的构造函数有两个时间参数：New(5*time.Minute, 10*time.Minute)——默认过期时间 5 分钟，清理间隔 10 分钟。但状态键在写入时覆盖了这个默认值：

localCache.Set(consts.DeviceStatusPrefix+deviceKey, data.Val(), 10*time.Second)

10 秒。设备详情缓存则用满 5 分钟：

localCache.Set(consts.DeviceDetailInfoPrefix+deviceKey, data, 5*time.Minute)

同一个本地缓存，两种截然不同的有效期，背后是两类数据的性质差异：

数据	键前缀	本地有效期	Redis 侧 TTL	变化频率
设备在线状态	`deviceStatus:`	10 秒	由设备超时决定，默认 120 秒	秒级
设备详情（含物模型）	`DeviceDetailInfo:`	5 分钟	永久（不设 TTL）	用户改配置时
产品详情	`ProductDetailInfo:`	无本地缓存	永久	极少
系统参数	`systemConfig:`	无本地缓存	永久	极少
设备告警规则	`deviceAlarmRule:`	无本地缓存	永久	偶尔

在线状态之所以只用 10 秒，是因为它是唯一一个"平台自己会改"的缓存项：设备可能在第 0 秒还在线，第 11 秒掉线。如果本地缓存也留 5 分钟，那么掉线事件处理完之后，接口层还可能在 5 分钟内继续返回"在线"——告警已经发了，界面还绿着，这种自相矛盾比单纯的延迟更难排查。

设备详情则相反：它是用户配置出来的，改一次动一次，而且结构复杂（含 TSL 物模型），每次从 Redis 反序列化都不便宜。5 分钟的本地副本换来的收益远大于风险。

不过要注意，这里有两份同名函数：

// pkg/dcache/deviceCache/device.go —— 直连 Redis，不碰本地缓存 func GetDeviceDetailInfo(deviceKey string) (out *model.DeviceOutput, err error) { data, err := cache.Instance().Get(context.Background(), consts.DeviceDetailInfoPrefix+deviceKey) if err != nil || data.Val() == nil { return } if err = gconv.Scan(data.Val(), &out); err != nil { return } return }

pkg/dcache下的那份带本地缓存与重试，pkg/dcache/deviceCache下的这份只读 Redis。后者是给"设备刚连上来、必须以此刻的配置为准"的场景准备的。

四、设备详情缓存：三次重试、指数退避与一个 3 秒的边界

设备详情缓存是整条链路上被调用最频繁的东西——每解析一条上报报文都要读一次。所以它的读取路径写得更保守：

func GetDeviceDetailInfo(deviceKey string) (out *model.DeviceOutput, err error) { converter := general.NewConverter() if cachedData, found := localCache.Get(consts.DeviceDetailInfoPrefix + deviceKey); found { out, err = converter.ConvertToDeviceOutput(cachedData) return } // 定义重试次数和退避策略，根据Redis配置调整 maxRetries := 3 baseBackoff := 50 * time.Millisecond // 设置超时时间为3秒，避免长时间阻塞 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() for i := 0; i < maxRetries; i++ { if ctx.Err() != nil { return nil, ctx.Err() } redisValue, err := cache.Instance().Get(ctx, consts.DeviceDetailInfoPrefix+deviceKey) if err == nil && redisValue != nil && redisValue.Val() != nil { data = redisValue.Val() break } ... } }

重试只对连接类错误生效，判断方式是白名单式地看错误文本：

// 检查是否是连接错误 connErr := err != nil && (strings.Contains(err.Error(), "connect: connection") || strings.Contains(err.Error(), "i/o timeout") || strings.Contains(err.Error(), "connection reset") || strings.Contains(err.Error(), "EOF") || strings.Contains(err.Error(), "broken pipe")) // 如果不是连接错误，不需要重试 if !connErr && err != nil { return nil, err }

这点很关键。如果是"键不存在"，重试三次毫无意义，只是白白拖慢一次请求；如果是网络抖动导致的connection reset，重试才可能救回来。退避是标准的移位运算：

backoff := baseBackoff * time.Duration(1<<uint(i))

50ms、100ms、200ms，三次总计不到 400ms 的等待，全部压在 3 秒的上下文里。这个 3 秒是整个设计里最硬的一个数字：调用方通常就在消息处理协程里，一个卡住的 Redis 会顺着调用链把它挂住，所以宁可提前返回错误，也不能无界等待。

五、真正的离线信号，是 Redis 自己发出的

到现在为止，机制还缺最后一块：谁在什么时候去检查键有没有过期？

答案是不检查。Redis 的键空间通知会在键到期时主动把事件推出来。deviceStatus包在init()里就起了订阅：

func init() { go SubscribeExpiryEvents() } // SubscribeExpiryEvents 订阅设备离线事件 func SubscribeExpiryEvents() { ctx := context.Background() redisDbName, _ := g.Cfg().Get(ctx, "redis.default.db", 0) channels := fmt.Sprintf("__keyevent@%d__:expired", redisDbName.Int()) ... }

订阅的是__keyevent@<db>__:expired，用的是模式订阅（PSubscribe）而不是精确订阅——因为键名里带着设备标识，不可能枚举。

这里有个容易踩的坑：通道名里的<db>必须和实际使用的 Redis 库号一致，而且 Redis 服务端必须开启notify-keyspace-events，至少要包含Ex。这两个条件任何一个不满足，订阅会静默地成功，但一个事件都收不到——现象是"在线的能上、离线的永远不离"。排查这类问题，第一步永远是手动CONFIG GET notify-keyspace-events。

收到事件后的处理逻辑很轻：

keys := strings.Split(message.Payload, ":") if len(keys) == 2 { if keys[0] == "deviceStatus" { deviceID := keys[1] // 使用独立协程处理离线逻辑，避免阻塞PubSub消息接收 go func(deviceID string) { processCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() device, err := dcache.GetDeviceDetailInfo(deviceID) if err != nil { g.Log().Debugf(processCtx, "获取设备%s信息失败: %v", deviceID, err) return } if err := Offline(processCtx, device); err != nil { g.Log().Debugf(processCtx, "设备%s离线处理失败: %v", deviceID, err) return } }(deviceID) } }

deviceStatus:<key>按冒号切开刚好两段。这个"刚好"是有代价的：如果设备 key 本身包含冒号，解析就会错位。所以设备命名规则里不允许出现冒号，这是一条隐含约束，写在契约里比写在注释里更可靠。

每个受影响的设备单独起一个协程、配一个 10 秒的超时，而不是在 PubSub 的接收循环里同步处理。原因很实际：PubSub 的接收循环一旦被阻塞，后面所有过期事件都会在缓冲区里堆积，而缓冲区是有界队列，满了就是静默丢消息。用协程换吞吐，代价是需要在Offline内部自己保证幂等——后面会看到，它确实是靠状态口径来兜底的。

六、一个订阅能不能活过一夜：健康检查、静默检测与 4 小时重建

PubSub 这种东西，写起来三行，跑起来能撑多久是另一回事。长连接会遇到 NAT 超时、中间设备静默断链、Redis 主从切换——而且 TCP 层往往不会立刻报错，表现为"连接看着还在，消息再也不来"。

socketeer/gpool那套东西解决的是并发上限问题，而订阅这里要解决的是活性检测问题。这段代码在SubscribeExpiryEvents里占了很大篇幅，值得单独看一下它的三套保险：

第一套是健康检查协程。每 15 秒给 PubSub 发一次Ping，连续失败 3 次就主动退出主循环：

go func() { ticker := time.NewTicker(15 * time.Second) defer ticker.Stop() var failCount int const maxFailCount = 3 for { select { case <-done: return case <-ticker.C: receiveCtx, cancel := context.WithTimeout(ctx, 3*time.Second) var pingErr error func() { defer func() { if r := recover(); r != nil { g.Log().Debugf(ctx, "PubSub健康检查发生panic: %v", r) failCount = maxFailCount // 立即触发重连 } }() // Redis PubSub不提供IsAlive方法，通过Ping尝试给服务器发消息来检查连接状态 pingErr = pubSub.Ping(receiveCtx, "") }() cancel() if pingErr != nil { failCount++ } else { failCount = 0 } if failCount >= maxFailCount { healthCheckFailed = true close(done) return } } } }()

第二套是静默检测。这个更隐蔽也更实用：即使Ping一切正常，只要连续 2 分钟没有收到任何消息，也认为链路已经不可信：

// 添加连续消息接收超时检测 lastMsgTime := time.Now() timeoutChecker := time.NewTicker(30 * time.Second) for { select { case <-timeoutChecker.C: // 检查是否长时间没有接收到消息 if time.Since(lastMsgTime) > 2*time.Minute { g.Log().Debug(ctx, "PubSub长时间未收到消息，准备重新订阅") close(done) return } case msg, ok := <-ch: ... lastMsgTime = time.Now() } }

这套检测针对的是"半开连接"：TCP 还没断，Ping 还能收到 PONG，但订阅状态已经丢了。在设备量大的系统里这种情况反而不容易察觉，因为总有别的消息在来；而在设备量小的部署里，它恰好是最容易踩到的。

第三套是 4 小时的定期重建：

// 定期重建订阅以避免长时间连接问题 reconnectTicker := time.NewTicker(4 * time.Hour) defer reconnectTicker.Stop()

主动、无理由地定期把连接换掉，看起来是浪费，实际上是把"不可预期的失效"转化为"可预期的重建窗口"。很多长期运行的订阅类服务最后都会走到这一步。

退出时的重启也是自恢复的：

restartSubscription := func() { g.Log().Warning(ctx, "设备状态订阅通道已关闭，5秒后尝试重新订阅") time.Sleep(5 * time.Second) go SubscribeExpiryEvents() } defer func() { if r := recover(); r != nil { g.Log().Errorf(ctx, "设备状态订阅异常: %v", r) restartSubscription() } }()

整个循环外面套了recover，任何一处 panic 都不会让订阅永久死亡——这是"守护进程不能自己退出"这个朴素原则的直接落地。

七、上下线之后被串起来的四件事

Online和Offline是这条链路的终点，但它们做的事远不止改状态。以Online为例：

func Online(ctx context.Context, device *model.DeviceOutput) (err error) { // 插入设备上线日志 deviceCacheData.InertDeviceLog(ctx, consts.MsgTypeOnline, device.Key, iotModel.DeviceOnlineMessage{ Timestamp: time.Now().UnixMilli(), Desc: "", }, true) pushDeviceStatus(device.Key, consts.DeviceStatueOnline) // 上线告警提醒 data := iotModel.ReportStatusData{ Status: "online", CreateTime: gtime.Now().Unix(), } err = service.AlarmRule().Check(ctx, device.Product.Key, device.Key, consts.AlarmTriggerTypeOnline, data) // 场景规则处理 go func() { reportData, _ := json.Marshal(iotModel.ReportData{ ProductKey: device.ProductKey, DeviceKey: device.Key, DataType: consts.SceneTriggerOnLine, Data: data, }) err = queues.ScheduledSceneRule.Push(ctx, consts.QueueSceneRuleTopic, guid.S(), reportData, 10) // 北向上线消息 north.WriteMessage(ctx, north.DeviceOnlineMessageTopic, nil, device.Product.Key, device.Key, iotModel.DeviceOnlineMessage{ Timestamp: time.Now().UnixMilli(), Desc: "", }) }() go func() { if upgradeCommandErr := ota.UpgradeCommand(ctx, device); upgradeCommandErr != nil { g.Log().Debugf(ctx, "设备上线后通知设备升级失败，失败原因：%s 设备信息:%+#v", upgradeCommandErr.Error(), device) } }() return }

拆开看是四类动作，而且它们的可靠性要求完全不同：

动作	载体	同步/异步	失败后果
设备日志	Redis List（批处理）	同步调用、异步落盘	日志缺失，不影响业务
状态推送	本地任务队列	同步入队	前端状态更新滞后
告警检查	直接调用告警服务	同步	上线告警不触发
场景规则	分布式任务队列	独立协程	场景联动漏一次
北向消息	MQTT 发布	独立协程	上游订阅方遗漏
OTA 升级通知	独立协程	独立协程	升级提示延后

同一个"上线"，被拆成了同步的、必须成功的（状态、告警）和异步的、尽力而为的（场景、北向、OTA）。这个划分不是随手写的：告警漏了是事故，OTA 通知晚 200 毫秒没人会发现。

Offline的结构基本对称，但有一处不同值得注意——它的AlarmRule().Check是直接return err的，而Online里是吞掉错误只打日志：

// Online 里 err = service.AlarmRule().Check(ctx, device.Product.Key, device.Key, consts.AlarmTriggerTypeOnline, data) if err != nil { g.Log().Debugf(ctx, "告警检测失败: %s", err.Error()) } // Offline 里 err = service.AlarmRule().Check(ctx, device.ProductKey, device.Key, consts.AlarmTriggerTypeOffline, data) return

离线告警的语义是"这件事必须有人知道"，所以错误要往上抛；上线告警更像一条通知，抛出去反而会让调用方误判整条上线链路失败。

八、MQTT 设备走的是另一条路：1e9 秒的上线与 1 秒的离线

上面那套是给主动上报型设备用的：设备不知道自己被平台看着，平台靠 TTL 推断它是否还在。

但 MQTT、TCP 这类长连接设备不一样——平台真的知道它连上了还是断了。broker 会给出明确的 connect/disconnect 事件，不需要等 120 秒去猜。所以在 MQTT 监控器里，挂的是另一对方法：

// RunMqttMonit 启动mqtt 设备上线离线监控 monitor.SetConnectHandler(func(info mqttMonit.DeviceInfo) { if deviceCache.VerifyDevice(info.ClientID) { g.Log().Debugf(ctx, "设备已连接: %s", info.ClientID) deviceStatus.UpdateDeviceOnlineStatusInRealTime(info.ClientID) } }) monitor.SetDisconnectHandler(func(info mqttMonit.DeviceInfo) { if deviceCache.VerifyDevice(info.ClientID) { g.Log().Debugf(ctx, "设备已断开连接: %s", info.ClientID) deviceStatus.UpdateDeviceOfflineStatusInRealTime(info.ClientID) } })

注意这里用的是deviceCache.VerifyDevice——直连 Redis 的那份，刻意绕开了 5 分钟的本地缓存。原因很清楚：设备刚连上来那一刻，需要的是此刻的配置，而不是五分钟前的快照。如果设备刚被禁用，缓存里的旧配置会让它"通过校验"然后被放行。

这两个方法的实现很短，但每一行都有原因：

// UpdateDeviceOnlineStatusInRealTime 即时更新设备上线状态 func UpdateDeviceOnlineStatusInRealTime(deviceKey string) { device, err := deviceCache.GetDeviceDetailInfo(deviceKey) if err != nil { g.Log().Errorf(context.Background(), "get device info error: %v", err) return } if device == nil { return } ctx := context.Background() timeout := time.Duration(1000000000) * time.Second err = redisDb.DB().GetClient().SetEx(ctx, consts.DeviceStatusPrefix+device.Key, time.Now(), timeout).Err() if err != nil { g.Log().Debug(ctx, "即时更新设备设备状态失败", err) return } if err := Online(ctx, device); err != nil { g.Log().Debug(ctx, device.Key, "设备上线处理失败") } return }

time.Duration(1000000000) * time.Second——十亿秒，约 31.7 年。这不是笔误，而是一个刻意的"永不过期"：MQTT 设备既然有明确的断开事件，就不需要用 TTL 来推断。把这个键的过期时间推到远得没有任何实际意义，等于告诉系统"这个键只由断线事件来撤销"。

离线那一半更巧妙：

// UpdateDeviceOfflineStatusInRealTime 即时更新设备离线状态 func UpdateDeviceOfflineStatusInRealTime(deviceKey string) { ctx := context.Background() err := redisDb.DB().GetClient().Expire(ctx, consts.DeviceStatusPrefix+deviceKey, time.Second*1).Err() if err != nil { g.Log().Debug(ctx, "更新设备交线状态失败", err) return } }

它没有直接调用Offline，而是把 TTL 改成 1 秒，剩下的事交给上一节那个订阅去做。1 秒后键过期 → 触发expired事件 → 订阅里走Offline分支。

这么写的好处是只有一个地方会执行离线逻辑。如果这里直接调Offline，那么 TTL 到期的路径也必须屏蔽掉，否则同一台设备会离线两次——一次日志、两条告警、两轮场景联动。把"撤销在线状态"和"处理离线业务"绑在同一个触发点上，就从根本上消除了重复。代价是延迟从 0 变成了约 1 秒，对一个本来就要等 120 秒的系统来说可以忽略。

九、状态之外：设备数据到底落在哪儿

状态键回答的是"在不在"，上报的数据本身则进另一个键空间，前缀是deviceCacheData:。写入走的是LPush + LTrim的固定窗口列表：

fullKey := DeviceDataCachePrefix + key // 使用pipeline执行操作，采用长连接模式 return r.execPipelineWithLazy(timeoutCtx, func(pipe redis.Pipeliner) error { pipe.LPush(timeoutCtx, fullKey, serializedValue) pipe.LTrim(timeoutCtx, fullKey, 0, r.recordLimit-1) pipe.TTL(timeoutCtx, fullKey) return nil }, fullKey, buffer)

LPush让最新的数据永远在索引 0，LTrim把列表长度卡在recordLimit（默认从配置system.deviceCacheData.recordLimit读，缺省 1000）。这就是"设备详情页最近 1000 条数据"的来源，而且读的时候天然是倒序：

func GetDeviceDetailDataByLatest(ctx context.Context, deviceKey string) (res iotModel.ReportPropertyData) { data, err := redisDb.DB().GetDataByLatest(context.Background(), deviceKey) if err != nil || data == "" { return } var value = iotModel.DeviceLog{} if err := json.Unmarshal([]byte(data), &value); err != nil { return } // 基于物模型解析数据 device, err := dcache.GetDeviceDetailInfo(deviceKey) ... res, err = service.DevTSLParse().ParseData(ctx, device, []byte(value.Content)) return }

注意GetDeviceDetailInfo在这里又被调了一次——落库是原始 JSON，读出来才按物模型解析。这么做让写入路径极轻（一个json.Marshal加一次 pipeline），解析成本被推迟到真正有人看的时候。代价是每条数据在网络里传了两遍、解析了两遍（一次写、一次读），以及物模型变更后老数据的解析口径会自动变新——这是取舍，不是疏漏。

大批量写入时还有个批处理器兜底。InertDeviceLog的第三个参数buffer决定走哪条路：

if buffer { err := redisDb.GetDeviceBatcher().Add(ctx, deviceKey, deviceLog) if err != nil { // 批处理失败时回退到直接写入 g.Log().Warningf(ctx, "批处理器添加数据失败，尝试直接写入: %v", err) if insertErr := redisDb.DB().InsertData(ctx, deviceKey, deviceLog, false); insertErr != nil { g.Log().Errorf(ctx, "设备日志数据写入失败: %v", insertErr) } } } else { if err := redisDb.DB().InsertData(ctx, deviceKey, deviceLog, false); err != nil { g.Log().Errorf(ctx, "设备日志数据直接写入失败: %v", err) } }

DeviceDataBatcher是按设备分组的缓冲区，默认每 20 条或每 100 毫秒刷一次。上线/下线日志传的是true，也就是允许攒批——这类日志量小、可以忍受百毫秒延迟，攒起来能明显减少 round trip。

十、几处值得留神的地方

读源码比夸源码有用。挑四处我认为需要留意的说说。

第一，本地缓存会制造 10 秒的状态不一致窗口。设备掉线后，Offline()会把 Redis 里的键删掉（准确地说是等着它自己过期），但localCache里那份"在线"副本还有最多 10 秒寿命。也就是说，在这 10 秒里，GetDeviceStatus依然返回 2。对大多数展示场景无所谓，但如果某个联动逻辑用GetDeviceStatus做互斥判断，就要接受这 10 秒的重叠。要么接受，要么在Offline里顺手把本地缓存也删掉——源码没有做，这是有意的宽松。

第二，SearchKey是一次全量KEYS扫描。

func SearchKey(keyword string) (keys []string, err error) { data, err := cache.Instance().Keys(context.Background()) if err != nil { return nil, err } for _, item := range data { if strings.Contains(item.(string), keyword) { keys = append(keys, item.(string)) } } return }

GetOnlineDeviceList、CountDeviceOnlineNum、GetAllDeviceList、GetRuleChainData全都建在它之上。单实例、几千个键时完全没问题；一旦 Redis 里有了几十万个键，这条命令会阻塞整个实例——KEYS是 O(N) 且不释放事件循环。真到那个规模，得换成SCAN游标，或者干脆为在线设备维护一个 Set，用SADD/SREM同步维护。

第三，GetDeviceStatus里本地缓存命中时那个return 1分支实际上是死代码。

if cachedData, found := localCache.Get(consts.DeviceStatusPrefix + deviceKey); found { if cachedData != nil { return 2 } return 1 }

只有写缓存时传了nil，found才是true而cachedData是nil；但唯一的写入点是localCache.Set(..., data.Val(), ...)，而那里已经保证data.Val()非 nil。所以return 1走不到。留着无害，但如果哪天有人改成"用 nil 表示离线"来省内存，这段逻辑的语义会突然变成正确的——这种"暂时无用但语义自洽"的代码，恰恰是最容易在重构中被误删的。

第四，PSubscribe的断线重连是"整体重启"而不是"局部恢复"。任何一处失败都会close(done)然后走restartSubscription，重新走一遍PSubscribe、Receive初始化检查、创建 channel 的完整流程。整套流程里 pubSub 的Close()挂在defer上，理论上不会泄漏连接；但 5 秒的固定退避意味着如果 Redis 整体不可用，就会以每 5 秒一次的频率反复建连——量不大，但日志会很吵。生产环境上更常见的是把固定退避换成带抖动的指数退避。

把这四处摆出来，不是要否定这套设计。相反：用一条 TTL 键替代心跳表、用过期事件替代轮询、用两套超时策略区分主动上报型和长连接型设备，这三步已经把"设备是否在线"这个最基础也最容易腐化的问题收得很紧。它没打算给出一个数学上精确的答案，而是给出一个在规模下依然成立、且失效方式可解释的答案。