MQTT消息容错与幂等处理指令去重、超时重试、离线消息补发机制大家好我是黒漂技术佬。MQTT专题第二篇咱们聊一个业务层面的核心问题——消息可靠性。上一篇讲了连接层面的配置这一篇讲消息层面的容错。你有没有遇到过开门指令发了两次门开了两次或者指令发出去了不知道设备收到没又或者设备断网期间的指令上线后重复执行…这些问题都属于消息容错和幂等的范畴直接关系到交易安全今天一次性讲透。一、为什么需要消息容错和幂等先讲两个真实的生产事故事故一网络抖动云端的开门指令重发了一次设备收到两条一模一样的指令连续开了两次门。第一次用户正常取货第二次门开了没人管货丢了好几瓶。事故二设备断网了云端连续发了三条重启指令。设备上线后一次性收到三条连续重启了三次用户半天用不了。这两个问题的根源都是网络不可靠导致消息可能重复、可能延迟、可能丢失但业务逻辑假设消息是恰好一次的。MQTT QoS1的本质QoS1保证至少一次送达但不保证只送达一次。也就是说消息不会丢大概率但可能重复所以用QoS1就必须做幂等处理这是铁律。二、指令去重机制核心思路消息ID 已处理记录每条指令带一个唯一的消息ID设备端记录已经处理过的消息ID重复的直接忽略。消息格式设计每条MQTT指令都要有统一的格式{msg_id:cmd_20260708_153022_abc123,cmd_type:open_door,timestamp:1720423822000,data:{order_id:O20260708123456,door_no:1}}msg_id 生成规则前缀 时间戳 随机串保证全局唯一。去重实现publicclassMessageDeduplicator{// 用LRU缓存存已处理的消息ID最多存1000条防止内存爆privateLruCacheString,LongprocessedMessagesnewLruCache(1000);// 同时持久化到数据库重启了也能去重privateMessageLogDbdb;/** * 检查消息是否重复 * return true重复false新消息 */publicbooleanisDuplicate(StringmsgId){// 先查内存if(processedMessages.get(msgId)!null){returntrue;}// 再查数据库if(db.exists(msgId)){// 查到了放回内存缓存processedMessages.put(msgId,System.currentTimeMillis());returntrue;}returnfalse;}/** * 标记消息已处理 */publicvoidmarkProcessed(StringmsgId){processedMessages.put(msgId,System.currentTimeMillis());db.insert(msgId,System.currentTimeMillis());}/** * 定期清理过期记录比如只保留7天 */publicvoidcleanExpired(){longexpireTimeSystem.currentTimeMillis()-7*24*3600*1000;db.deleteExpired(expireTime);}}使用方式publicvoidhandleCommand(Stringpayload){CommandMessagecmdparseCommand(payload);StringmsgIdcmd.getMsgId();// 第一步去重检查if(deduplicator.isDuplicate(msgId)){Log.w(MQTT,重复指令忽略: msgId);return;}// 第二步执行业务booleansuccessexecuteCommand(cmd);// 第三步标记已处理deduplicator.markProcessed(msgId);// 第四步上报结果reportCommandResult(msgId,success);}三、指令超时与重试机制发出去的指令多久没响应算超时超时了怎么办超时判定publicclassCommandTimeoutManager{privatestaticfinallongDEFAULT_TIMEOUT30000;// 默认30秒超时privateMapString,CommandWaiterwaitingCommandsnewConcurrentHashMap();/** * 发送指令并等待响应 */publicCommandResultsendCommandWithTimeout(CommandMessagecmd){StringmsgIdcmd.getMsgId();CommandWaiterwaiternewCommandWaiter(msgId);waitingCommands.put(msgId,waiter);// 发送消息MqttManager.publish(cmd.getTopic(),cmd.getPayload());// 等待响应带超时try{booleansuccesswaiter.await(DEFAULT_TIMEOUT,TimeUnit.MILLISECONDS);if(success){returnwaiter.getResult();}else{// 超时了returnCommandResult.timeout();}}finally{waitingCommands.remove(msgId);}}/** * 收到指令结果时调用 */publicvoidonCommandResult(StringmsgId,CommandResultresult){CommandWaiterwaiterwaitingCommands.get(msgId);if(waiter!null){waiter.setResult(result);waiter.countDown();}}}重试策略指令超时了要不要重试分情况指令类型是否重试重试次数原因开门指令不重试0重复开门风险太大宁可不成功也不能重复开状态查询可重试2次幂等操作重试没关系配置更新可重试2次幂等重复下发不影响重启指令不重试0重复重启体验差核心原则有副作用的、不能重复执行的指令不要自动重试让上层业务决定。四、离线消息补发处理设备断网期间云端下发的指令会存在broker里设备上线后一股脑全发过来。离线消息的问题过期指令断了一天网上线后收到昨天的开门指令还有意义吗重复指令云端可能重发了好几次离线消息里有重复的顺序问题多条指令的执行顺序有没有要求处理策略publicclassOfflineMessageHandler{privatestaticfinallongMSG_EXPIRE_TIME5*60*1000;// 消息5分钟过期/** * 处理收到的消息包括离线的 */publicvoidhandleMessage(MqttMessagemsg){CommandMessagecmdparseCommand(msg.getPayload());// 1. 先去重if(deduplicator.isDuplicate(cmd.getMsgId())){Log.i(Offline,重复消息忽略);return;}// 2. 检查是否过期if(isExpired(cmd)){Log.i(Offline,消息已过期忽略: cmd.getMsgId());deduplicator.markProcessed(cmd.getMsgId());// 也要标记避免每次都判断reportExpiredMessage(cmd);return;}// 3. 检查指令类型是否需要延迟执行if(needDelayWhenBatch(cmd)){// 批量离线消息错开执行避免一下子全执行scheduleDelayedExecution(cmd);}else{// 立即执行executeCommand(cmd);}deduplicator.markProcessed(cmd.getMsgId());}privatebooleanisExpired(CommandMessagecmd){longageSystem.currentTimeMillis()-cmd.getTimestamp();returnageMSG_EXPIRE_TIME;}}不同类型指令的过期时间指令类型过期时间原因开门指令2分钟用户等不了太久过期了就别开了配置更新24小时配置更新不着急但最终要生效重启指令1小时太久了就没必要重启了OTA升级7天升级指令可以等五、消息日志与溯源出了问题要能查所以每条消息的处理都要有日志。消息日志表设计CREATETABLEmessage_log(idINTEGERPRIMARYKEYAUTOINCREMENT,msg_idVARCHAR(64)UNIQUE,-- 消息IDdirectionINTEGER,-- 1接收 2发送topicVARCHAR(128),-- 主题cmd_typeVARCHAR(32),-- 指令类型payloadTEXT,-- 消息内容receive_timeBIGINT,-- 收到时间process_timeBIGINT,-- 处理完成时间statusINTEGER,-- 处理状态0待处理 1成功 2失败 3过期 4重复error_msgVARCHAR(256)-- 失败原因);日志的作用问题排查用户说我扫码没开门查一下收到指令没、什么时候处理的统计分析指令成功率、平均处理时长、重复率、过期率对账开门次数和交易订单对不对得上六、关键指令的幂等设计去重是第一层幂等是第二层。就算去重失效了重复执行也不能出问题。什么是幂等同一个指令执行一次和执行N次结果是一样的。各类指令的幂等实现1. 开门指令开门是最危险的必须保证幂等。publicclassDoorCommandExecutor{/** * 幂等开门同一个订单号只开一次门 */publicbooleanopenDoor(StringorderId,intdoorNo){// 用订单号做幂等键if(hasOpenedForOrder(orderId)){Log.w(Door,订单 orderId 已经开过门了跳过);returntrue;// 也算成功}// 执行开门booleansuccessDoorController.open(doorNo);if(success){markOrderOpened(orderId);}returnsuccess;}}2. 配置更新配置更新天然幂等同一个配置设多少次都一样但要注意版本号publicvoidupdateConfig(ConfigUpdatecmd){// 只更新版本号更高的配置if(cmd.getVersion()ConfigManager.getCurrentVersion()){Log.i(Config,配置版本过低忽略);return;}ConfigManager.applyConfig(cmd.getConfig(),cmd.getVersion());}3. 重启指令重启没法完全幂等但可以加防抖publicvoidreboot(Stringreason){// 5分钟内已经重启过一次了就别再重启了if(System.currentTimeMillis()-lastRebootTime5*60*1000){Log.w(Reboot,短时间内重复重启请求忽略);return;}lastRebootTimeSystem.currentTimeMillis();doReboot();}七、消息优先级与队列设备端收到的消息有急有缓不能按收到顺序挨个处理。优先级划分优先级消息类型处理策略高开门指令、紧急告警立即处理插队中配置更新、重启指令正常排队低日志查询、调试指令闲时处理优先级队列实现publicclassPriorityMessageQueue{privatePriorityQueueQueuedMessagequeuenewPriorityQueue((a,b)-Integer.compare(b.getPriority(),a.getPriority()));publicvoidenqueue(MqttMessagemsg,intpriority){queue.offer(newQueuedMessage(msg,priority));}publicQueuedMessagedequeue(){returnqueue.poll();}}八、踩坑经验坑1消息ID用时间戳并发下重复现象高并发下两条指令msgId一样去重把正常的也过滤了原因时间戳精度不够毫秒级同一毫秒可能产生多个解决时间戳 随机数 序列号组合确保唯一坑2去重记录存内存重启就失效现象设备刚重启的时候重复指令拦不住原因只在内存里存了已处理记录重启就没了解决必须持久化到数据库内存只是缓存加速坑3离线消息太多设备上线卡死现象断网好几天上线后几百条离线消息涌进来APP直接ANR解决消息入队异步处理不要在MQTT回调线程里直接执行业务控制处理速率坑4时钟不同步导致过期判断错现象设备时间不准明明是新消息被判断成过期的原因用设备本地时间判断消息时间戳解决过期时间用相对时间比如broker的时间或者设备定期同步时间九、总结MQTT消息容错的核心不丢、不重、不乱序。消息去重每条消息带唯一ID记录已处理的重复的直接忽略超时重试区分指令类型有副作用的不自动重试离线补发过期判断优先级限流别让离线消息冲垮设备幂等设计就算去重失效了重复执行也不能出问题消息日志每条消息全链路留痕出问题能溯源把这些做好了MQTT层面的消息可靠性就基本达标了业务才能跑得稳。我是黒漂技术佬专注无人零售物联网实战。MQTT专题两篇到这里结束下一章进入硬件控制专题聊聊怎么通过串口/继电器控制门锁。有MQTT消息可靠性相关的坑欢迎私信交流~