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