SpringCloud Alibaba無人售貨柜實戰五設備通信協議設計——MQTT/HTTP指令下發與狀態回調讓售貨柜開門它就開門讓它重啟它就重啟——這背后需要一套嚴謹的通信協議。指令丟了怎么辦設備沒響應怎么辦這篇全給你兜住。一、設備通信架構整個通信鏈路是一條完整的指令生命周期服務端下發指令 │ ▼ MQTT Broker → device/{sn}/command Topic │ ▼ 設備端接收 → 執行操作開電磁鎖/重啟等 │ ▼ 設備端上報回調 → device/{sn}/callback Topic │ ▼ 服務端處理回調 → 更新指令狀態 → 觸發后續業務正常情況下這條鏈路在2秒內跑完。但現實世界有網絡抖動、設備死機、MQTT斷連等各種意外所以通信協議必須設計好超時、重試、冪等三道保險。二、通信協議設計原則簡潔字段名短小精悍JSON層級不超過3層減少設備端解析負擔可靠每條指令有唯一ID支持冪等執行和結果追蹤可擴展預留extra字段新增指令類型不改協議結構可追蹤每條指令從下發到回調全鏈路有日志方便排查三、下行指令協議3.1 指令結構定義服務端發給設備的指令格式{commandId:cmd-550e8400-e29b-41d4-a716-446655440000,command:OPEN_DOOR,params:{orderId:202607291234567890,maxDuration:300},timeout:30,timestamp:1753766400000,sign:a1b2c3d4e5f6}字段類型必填說明commandIdString是指令唯一IDUUID生成用于關聯回調commandString是指令類型枚舉paramsObject否指令參數不同指令參數不同timeoutint是超時時間秒默認30timestamplong是下發時間戳設備端可用于防重放signString是簽名MD5(commandId command timestamp secret)3.2 指令類型定義指令類型說明params參數超時建議OPEN_DOOR開柜門orderId(訂單號), maxDuration(最大開門時長秒)10秒CLOSE_DOOR強制關柜門無10秒RESTART重啟設備delay(延遲秒數)60秒SYNC_TIME同步時間serverTime(服務器時間戳)5秒INVENTORY盤點指令無設備端返回當前庫存30秒UPDATE_CONFIG更新配置heartbeatInterval, volume, autoClose…10秒UPLOAD_LOG上傳日志startTime, endTime60秒TAKE_PHOTO拍照cameraId(攝像頭編號)10秒四、上行回調協議設備執行完指令后通過回調Topic上報執行結果{commandId:cmd-550e8400-e29b-41d4-a716-446655440000,status:SUCCESS,data:{doorOpen:true,openDuration:45},errorCode:null,errorMsg:null,timestamp:1753766402000}字段類型必填說明commandIdString是關聯的指令ID和下行指令一一對應statusString是SUCCESS / FAILED / TIMEOUT / UNSUPPORTEDdataObject否執行結果數據不同指令返回不同errorCodeString否失敗時的錯誤碼errorMsgString否失敗時的錯誤描述timestamplong是回調時間戳4.1 各指令的回調data定義指令回調dataOPEN_DOOR{doorOpen: true, openDuration: 45}CLOSE_DOOR{doorClosed: true}RESTART{restartScheduled: true}SYNC_TIME{synced: true, deviceTime: 1753766402000}INVENTORY{items: [{productId: P001, count: 5}, ...]}TAKE_PHOTO{imageUrl: http://minio.xxx/photo/cmd-xxx.jpg}五、指令下發ServiceSlf4jServicepublicclassDeviceCommandService{AutowiredprivateDeviceCommandMappercommandMapper;AutowiredprivateMqttGatewaymqttGateway;AutowiredprivateRedisUtilsredisUtils;privatestaticfinalStringCOMMAND_PENDING_PREFIXcmd:pending:;privatestaticfinalStringDEVICE_TOKEN_PREFIXdevice:token:;/** * 下發指令 */publicDeviceCommandsendCommand(Stringsn,Stringcommand,JSONObjectparams,inttimeout){// 1. 生成指令IDStringcommandIdcmd-UUID.randomUUID().toString();// 2. 簽名StringtokenredisUtils.get(DEVICE_TOKEN_PREFIXsn);StringsignSecureUtil.md5(commandIdcommandSystem.currentTimeMillis()token);// 3. 構建指令消息JSONObjectmessagenewJSONObject();message.put(commandId,commandId);message.put(command,command);message.put(params,params);message.put(timeout,timeout);message.put(timestamp,System.currentTimeMillis());message.put(sign,sign);// 4. 存入數據庫DeviceCommandcmdnewDeviceCommand();cmd.setCommandId(commandId);cmd.setDeviceSn(sn);cmd.setCommand(command);cmd.setParams(params.toJSONString());cmd.setStatus(0);// 待執行cmd.setTimeoutSeconds(timeout);cmd.setSendTime(LocalDateTime.now());commandMapper.insert(cmd);// 5. 通過MQTT下發Stringtopicdevice/sn/command;mqttGateway.sendToMqtt(topic,message.toJSONString());log.info(指令已下發: sn{}, commandId{}, command{},sn,commandId,command);// 6. 存入Redis待回調集合用于超時檢查redisUtils.set(COMMAND_PENDING_PREFIXcommandId,sn,timeout10,TimeUnit.SECONDS);// 7. 更新指令狀態為已下發cmd.setStatus(1);commandMapper.updateById(cmd);returncmd;}/** * 發送開門指令業務封裝 */publicDeviceCommandopenDoor(Stringsn,StringorderId){JSONObjectparamsnewJSONObject();params.put(orderId,orderId);params.put(maxDuration,300);returnsendCommand(sn,OPEN_DOOR,params,10);}}六、回調處理Slf4jServicepublicclassDeviceCallbackService{AutowiredprivateDeviceCommandMappercommandMapper;AutowiredprivateRedisUtilsredisUtils;AutowiredprivateOrderFeignClientorderFeignClient;privatestaticfinalStringCOMMAND_PENDING_PREFIXcmd:pending:;/** * 監聽設備回調 */MqttMessageListener(topicdevice//callback)publicvoidonCallback(MqttMessagemessage){Stringtopicmessage.getTopic();Stringsntopic.split(/)[1];StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);CallbackReqreqJSON.parseObject(payload,CallbackReq.class);log.info(收到設備回調: sn{}, commandId{}, status{},sn,req.getCommandId(),req.getStatus());// 1. 查詢指令DeviceCommandcmdcommandMapper.selectByCommandId(req.getCommandId());if(cmdnull){log.error(回調指令不存在: commandId{},req.getCommandId());return;}// 2. 冪等檢查已經處理過的回調直接忽略if(cmd.getStatus()2||cmd.getStatus()3){log.warn(指令已處理忽略重復回調: commandId{}, status{},req.getCommandId(),cmd.getStatus());return;}// 3. 更新指令狀態if(SUCCESS.equals(req.getStatus())){cmd.setStatus(2);// 成功}else{cmd.setStatus(3);// 失敗}cmd.setResultData(req.getData()!null?req.getData().toJSONString():null);cmd.setCallbackTime(LocalDateTime.now());commandMapper.updateById(cmd);// 4. 清除Redis待回調標記redisUtils.delete(COMMAND_PENDING_PREFIXreq.getCommandId());// 5. 觸發后續業務handleCommandResult(sn,cmd,req);}/** * 根據指令類型觸發后續業務 */privatevoidhandleCommandResult(Stringsn,DeviceCommandcmd,CallbackReqreq){switch(cmd.getCommand()){caseOPEN_DOOR:if(SUCCESS.equals(req.getStatus())){// 開門成功通知訂單服務orderFeignClient.onDoorOpened(cmd.getParamsObject().getString(orderId));}else{// 開門失敗通知訂單服務取消訂單orderFeignClient.onDoorOpenFailed(cmd.getParamsObject().getString(orderId),req.getErrorMsg());}break;caseINVENTORY:// 盤點結果同步到庫存服務break;caseRESTART:log.info(設備重啟指令已確認: sn{},sn);break;}}}七、指令超時處理指令下發后不是萬事大吉——設備可能沒收到、可能收到了但執行卡死了。必須有超時檢查機制。7.1 延遲隊列方案用RocketMQ的延遲消息實現超時檢查Slf4jServicepublicclassCommandTimeoutChecker{AutowiredprivateDeviceCommandMappercommandMapper;AutowiredprivateRocketMQTemplaterocketMQTemplate;AutowiredprivateDeviceCommandServicecommandService;privatestaticfinalStringTIMEOUT_TOPICcommand-timeout-check;privatestaticfinalintMAX_RETRY2;/** * 下發指令時發送延遲消息延遲時間指令超時時間 */publicvoidsendTimeoutCheck(StringcommandId,intdelaySeconds){MessageStringmsgMessageBuilder.withPayload(commandId).build();// RocketMQ延遲級別: 1s1, 5s2, 10s3, 30s4, 1m5...intdelayLeveldelaySeconds5?2:(delaySeconds10?3:4);rocketMQTemplate.asyncSend(TIMEOUT_TOPIC,msg,newSendCallback(){OverridepublicvoidonSuccess(SendResultsendResult){}OverridepublicvoidonException(Throwablee){log.error(超時檢查消息發送失敗: commandId{},commandId,e);}},3000,delayLevel);}/** * 消費超時檢查消息 */RocketMQMessageListener(topicTIMEOUT_TOPIC,consumerGroupcommand-timeout-group)ComponentpublicclassTimeoutConsumerimplementsRocketMQListenerString{OverridepublicvoidonMessage(StringcommandId){DeviceCommandcmdcommandMapper.selectByCommandId(commandId);if(cmdnull)return;// 指令已完成成功或失敗無需處理if(cmd.getStatus()2||cmd.getStatus()3){return;}log.warn(指令超時未回調: commandId{}, command{}, retryCount{},commandId,cmd.getCommand(),cmd.getRetryCount());if(cmd.getRetryCount()MAX_RETRY){// 重試重新下發指令cmd.setRetryCount(cmd.getRetryCount()1);cmd.setStatus(1);commandMapper.updateById(cmd);// 重新通過MQTT下發JSONObjectmessagebuildCommandMessage(cmd);mqttGateway.sendToMqtt(device/cmd.getDeviceSn()/command,message.toJSONString());// 再次發送延遲檢查sendTimeoutCheck(commandId,cmd.getTimeoutSeconds());}else{// 超過最大重試次數標記超時cmd.setStatus(4);// 超時commandMapper.updateById(cmd);log.error(指令最終超時: commandId{},commandId);// 通知業務方處理}}}}八、HTTP備選通道MQTT不可用時Broker掛了或網絡斷了設備通過HTTP輪詢兜底拉取指令。8.1 設備端輪詢邏輯設備端如果MQTT連接失敗自動降級為HTTP輪詢模式每10秒請求: GET /api/device/{sn}/commands/pending 拉取待執行指令 → 執行 → POST /api/device/{sn}/callback 上報結果8.2 服務端輪詢接口RestControllerRequestMapping(/api/device)publicclassDevicePollController{AutowiredprivateDeviceCommandMappercommandMapper;/** * 設備拉取待執行指令 */GetMapping(/{sn}/commands/pending)publicResultListDeviceCommandgetPendingCommands(PathVariableStringsn){// 查詢狀態為已下發且未回調的指令ListDeviceCommandcommandscommandMapper.selectList(newLambdaQueryWrapperDeviceCommand().eq(DeviceCommand::getDeviceSn,sn).eq(DeviceCommand::getStatus,1).orderByAsc(DeviceCommand::getSendTime).last(LIMIT 5));returnResult.success(commands);}/** * 設備HTTP上報回調 */PostMapping(/{sn}/callback)publicResultVoidcallback(PathVariableStringsn,RequestBodyCallbackReqreq){callbackService.onCallback(sn,req);returnResult.success();}}HTTP輪詢是兜底方案不是常態。MQTT恢復后設備自動切回MQTT模式。雙通道設計保證了通信可靠性。九、安全設計9.1 設備Token認證設備連接MQTT時用Token做密碼認證。EMQX配置用戶認證后端對接Redis驗證MQTT連接用戶名: {設備SN} MQTT連接密碼: {Token} EMQX認證邏輯: GET device:token:{sn} → 比對密碼9.2 指令簽名防偽造每條指令帶sign字段設備端驗簽后才執行// 設備端驗簽Android/Java偽代碼publicbooleanverifySign(JSONObjectcommand,Stringtoken){StringcommandIdcommand.getString(commandId);Stringcmdcommand.getString(command);longtimestampcommand.getLong(timestamp);Stringsigncommand.getString(sign);StringexpectedSignMD5Utils.md5(commandIdcmdtimestamptoken);returnexpectedSign.equals(sign);}9.3 防重放攻擊設備端維護一個最近100條commandId的LRU緩存收到重復commandId直接忽略。配合timestamp字段超過5分鐘的指令直接丟棄。十、通信協議完整定義表指令方向params回調data超時重試OPEN_DOOR下行orderId, maxDurationdoorOpen, openDuration10s2次CLOSE_DOOR下行無doorClosed10s1次RESTART下行delayrestartScheduled60s0次SYNC_TIME下行serverTimesynced, deviceTime5s1次INVENTORY下行無items[]30s1次UPDATE_CONFIG下行多個配置項updated10s1次UPLOAD_LOG下行startTime, endTimelogUrl60s0次TAKE_PHOTO下行cameraIdimageUrl10s1次十一、小結設備通信協議設計的核心就四個字可靠、冪等。commandId貫穿整個生命周期從下發到回調到超時檢查全靠它串聯。MQTT是主通道HTTP輪詢是兜底RocketMQ延遲消息做超時檢查三層保障確保指令不丟、不重、不卡。安全層面Token認證指令簽名防重放三管齊下。這套協議跑通了設備端和服務端就能穩定對話后面的業務邏輯就是水到渠成的事。