SpringCloud Alibaba无人售货柜实战（四）：设备接入模块——设备注册、心跳上报、在线状态管理

SpringCloud Alibaba无人售货柜实战四设备接入模块——设备注册、心跳上报、在线状态管理一台售货柜上电后怎么让服务器知道我来了怎么判断它还活着怎么知道它掉线了这篇把设备接入的全流程讲透。一、设备接入流程总览无人售货柜不是普通的服务器它是一台跑在RK3588工控板上的Android设备通过MQTT和服务器通信。设备从上电到在线可用要经过四个阶段设备上电 → 连接MQTT Broker → 注册报到 → 心跳保活 → 在线服务详细步骤工控板上电启动Android App拉起读取本地存储的设备SN码连接EMQX MQTT Broker连接成功后向服务端发送注册请求携带SN码服务端校验SN合法性分配Token更新设备状态为在线设备每30秒发送一次心跳包服务端定时扫描90秒未收到心跳则标记离线二、MQTT Topic设计MQTT通信的核心是Topic设计。本项目采用{前缀}/{设备SN}/{动作}的层级结构Topic方向说明device/{sn}/register设备→服务端设备注册device/{sn}/heartbeat设备→服务端心跳上报device/{sn}/status设备→服务端状态上报开门/关门/故障device/{sn}/command服务端→设备下行指令开柜/重启等device/{sn}/callback设备→服务端指令执行回调每个设备订阅自己的device/{sn}/commandTopic服务端往这个Topic发消息设备就能收到。SN码天然就是路由Key。三、设备注册流程3.1 注册逻辑设备首次上电或Token过期后需要注册设备读取本地SN码出厂烧录不可篡改拼接SN 设备型号 固件版本发送到device/{sn}/registerTopic服务端收到消息查device表确认SN存在且未被禁用生成JWT Token存入Rediskeydevice:token:{sn}有效期7天更新设备状态为在线记录上线时间通过MQTT返回Token给设备3.2 注册Controller设备注册走MQTT消息不走HTTP。服务端用MQTT消息监听器处理Slf4jComponentpublicclassDeviceRegisterListener{AutowiredprivateDeviceServicedeviceService;AutowiredprivateRedisUtilsredisUtils;AutowiredprivateMqttGatewaymqttGateway;privatestaticfinalStringDEVICE_TOKEN_PREFIXdevice:token:;privatestaticfinallongTOKEN_EXPIRE_DAYS7L;/** * 监听设备注册Topic */MqttMessageListener(topicdevice//register)publicvoidonRegister(MqttMessagemessage){Stringtopicmessage.getTopic();Stringsntopic.split(/)[1];StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);log.info(收到设备注册请求: sn{}, payload{},sn,payload);DeviceRegisterReqreqJSON.parseObject(payload,DeviceRegisterReq.class);req.setSn(sn);// 1. 校验设备合法性DevicedevicedeviceService.getBySn(sn);if(devicenull){log.error(设备未登记: sn{},sn);sendRegisterResult(sn,false,设备未登记,null);return;}if(device.getStatus()DeviceStatus.MAINTENANCE.getCode()){sendRegisterResult(sn,false,设备维护中,null);return;}// 2. 生成TokenStringtokenJwtUtils.generateDeviceToken(sn,device.getId());// 3. 缓存Token到RedisredisUtils.set(DEVICE_TOKEN_PREFIXsn,token,TOKEN_EXPIRE_DAYS,TimeUnit.DAYS);// 4. 更新设备状态device.setStatus(DeviceStatus.ONLINE.getCode());device.setLastOnlineTime(LocalDateTime.now());device.setLastHeartbeatTime(LocalDateTime.now());device.setFirmwareVersion(req.getFirmwareVersion());device.setIpAddress(req.getIpAddress());deviceService.updateById(device);// 5. 返回注册结果sendRegisterResult(sn,true,注册成功,token);log.info(设备注册成功: sn{},sn);}privatevoidsendRegisterResult(Stringsn,booleansuccess,Stringmsg,Stringtoken){JSONObjectresultnewJSONObject();result.put(success,success);result.put(message,msg);if(token!null){result.put(token,token);}mqttGateway.sendToMqtt(device/sn/register/resp,result.toJSONString());}}四、心跳上报机制4.1 心跳协议设备每30秒上报一次心跳内容非常精简{sn:VM202601001,timestamp:1753766400000,cpuTemp:52.3,memoryUsage:68,diskUsage:45,signalStrength:-67}心跳不只是我还活着顺便带上设备健康指标。cpuTemp超过80度就该告警了。4.2 心跳处理ServiceSlf4jServicepublicclassDeviceHeartbeatService{AutowiredprivateDeviceServicedeviceService;AutowiredprivateRedisUtilsredisUtils;privatestaticfinalStringDEVICE_ONLINE_PREFIXdevice:online:;privatestaticfinallongONLINE_EXPIRE_SECONDS90L;/** * 处理心跳消息 */MqttMessageListener(topicdevice//heartbeat)publicvoidonHeartbeat(MqttMessagemessage){Stringtopicmessage.getTopic();Stringsntopic.split(/)[1];StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);HeartbeatReqreqJSON.parseObject(payload,HeartbeatReq.class);// 1. 更新Redis在线状态90秒过期StringonlineKeyDEVICE_ONLINE_PREFIXsn;redisUtils.set(onlineKey,1,ONLINE_EXPIRE_SECONDS,TimeUnit.SECONDS);// 2. 更新数据库最后心跳时间异步不阻塞MQTT回调DevicedevicenewDevice();device.setSn(sn);device.setLastHeartbeatTime(LocalDateTime.now());deviceService.updateHeartbeatBySn(sn,LocalDateTime.now());// 3. 异常告警检查if(req.getCpuTemp()!nullreq.getCpuTemp()80){log.warn(设备CPU温度过高: sn{}, temp{}℃,sn,req.getCpuTemp());// 发送告警通知...}log.debug(心跳处理完成: sn{},sn);}}心跳处理一定要快。MQTT消息回调是串行的如果心跳处理慢了后面的消息全部排队。所以DB更新用异步核心逻辑只做Redis更新。五、在线状态判定5.1 双重判定机制设备在线状态用Redis 定时扫描双重保障Redis快速判定device:online:{sn}这个Key存在就算在线过期时间90秒。心跳每30秒续期一旦心跳停止Key自动过期。DB定时扫描兜底机制防止Redis故障导致状态不一致。5.2 定时状态扫描Slf4jComponentpublicclassDeviceStatusScanner{AutowiredprivateDeviceServicedeviceService;AutowiredprivateRedisUtilsredisUtils;privatestaticfinalStringDEVICE_ONLINE_PREFIXdevice:online:;privatestaticfinallongOFFLINE_THRESHOLD_SECONDS90L;/** * 每60秒扫描一次将超时设备标记为离线 */Scheduled(fixedRate60000)publicvoidscanOfflineDevices(){log.info(开始扫描离线设备...);// 查询所有状态为在线的设备ListDeviceonlineDevicesdeviceService.listByStatus(DeviceStatus.ONLINE.getCode());intofflineCount0;LocalDateTimethresholdLocalDateTime.now().minusSeconds(OFFLINE_THRESHOLD_SECONDS);for(Devicedevice:onlineDevices){// 优先检查RedisBooleanisOnlineredisUtils.hasKey(DEVICE_ONLINE_PREFIXdevice.getSn());if(BooleanUtil.isFalse(isOnline)){// Redis已过期再检查DB最后心跳时间兜底if(device.getLastHeartbeatTime()null||device.getLastHeartbeatTime().isBefore(threshold)){// 标记离线device.setStatus(DeviceStatus.OFFLINE.getCode());deviceService.updateById(device);offlineCount;log.warn(设备离线: sn{}, lastHeartbeat{},device.getSn(),device.getLastHeartbeatTime());}}}log.info(离线扫描完成: 在线{}, 新增离线{},onlineDevices.size(),offlineCount);}}5.3 设备状态机注册成功 心跳超时/异常 OFFLINE ────────→ ONLINE ──────────→ OFFLINE │ │ 故障上报 重新注册 ↓ │ FAULT ←──────────────┘ │ 维修开始 ↓ MAINTENANCE │ 维修完成 ↓ OFFLINE → 重新注册 → ONLINE六、设备信息缓存策略设备信息是高频读取数据每次下单都要查设备状态直接查DB扛不住。缓存策略数据缓存KeyTTL说明在线状态device:online:{sn}90秒心跳续期设备信息device:info:{sn}30分钟DB变更时主动刷新认证Tokendevice:token:{sn}7天注册时写入设备锁device:lock:{sn}10秒防止并发操作同一设备publicDevicegetDeviceBySn(Stringsn){Stringkeydevice:info:sn;DevicedeviceredisUtils.get(key,Device.class);if(device!null){returndevice;}devicethis.getBySn(sn);if(device!null){redisUtils.set(key,device,30,TimeUnit.MINUTES);}returndevice;}七、常见问题7.1 设备频繁上下线网络抖动设备网络不稳定一会儿上线一会儿掉线状态频繁切换产生大量日志。解决方案引入离线延迟确认机制。心跳超时后不立即标记离线先进入疑似离线状态再等一个心跳周期30秒如果还没心跳才正式标记离线。// 疑似离线缓存30秒后过期redisUtils.set(device:pending_offline:sn,1,30,TimeUnit.SECONDS);7.2 心跳丢失MQTT QoS设置为1至少一次但极端情况下仍可能丢消息。解决设备端本地记录心跳序号连续3次未收到服务端ACK就重连服务端用Scheduled补偿扫描以DB时间戳为最终判据八、设备管理表补充在第一篇的device表基础上补充设备扩展信息表CREATETABLEdevice_config(idbigintNOTNULL,device_snvarchar(64)NOTNULLCOMMENT设备SN,heartbeat_intervalintNOTNULLDEFAULT30COMMENT心跳间隔(秒),offline_thresholdintNOTNULLDEFAULT90COMMENT离线判定阈值(秒),door_timeoutintNOTNULLDEFAULT120COMMENT开门超时(秒),max_open_durationintNOTNULLDEFAULT300COMMENT最大开门时长(秒),auto_close_enabledtinyintNOTNULLDEFAULT1COMMENT是否自动关门,volumeintNOTNULLDEFAULT50COMMENT音量(0-100),create_timedatetimeNOTNULLDEFAULTCURRENT_TIMESTAMP,update_timedatetimeNOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,PRIMARYKEY(id),UNIQUEKEYuk_device_sn(device_sn))ENGINEInnoDBDEFAULTCHARSETutf8mb4COMMENT设备配置表;九、小结设备接入是IoT项目的第一道关卡。核心就三件事注册让设备拿到身份、心跳让服务端知道设备活着、状态扫描兜底处理异常掉线。MQTT Topic设计要提前规划好层级结构后期改Topic成本极高。Redis做在线状态缓存是标配——DB扛不住每30秒一次的心跳写入。