企业官网建设流程全解析

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条，防止内存爆privateLruCache<String,Long>processedMessages=newLruCache<>(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(){longexpireTime=System.currentTimeMillis()-7*24*3600*1000;db.deleteExpired(expireTime);}}

使用方式

publicvoidhandleCommand(Stringpayload){CommandMessagecmd=parseCommand(payload);StringmsgId=cmd.getMsgId();// 第一步：去重检查if(deduplicator.isDuplicate(msgId)){Log.w("MQTT","重复指令，忽略: "+msgId);return;}// 第二步：执行业务booleansuccess=executeCommand(cmd);// 第三步：标记已处理deduplicator.markProcessed(msgId);// 第四步：上报结果reportCommandResult(msgId,success);}

三、指令超时与重试机制

发出去的指令，多久没响应算超时？超时了怎么办？

超时判定

publicclassCommandTimeoutManager{privatestaticfinallongDEFAULT_TIMEOUT=30000;// 默认30秒超时privateMap<String,CommandWaiter>waitingCommands=newConcurrentHashMap<>();/** * 发送指令并等待响应 */publicCommandResultsendCommandWithTimeout(CommandMessagecmd){StringmsgId=cmd.getMsgId();CommandWaiterwaiter=newCommandWaiter(msgId);waitingCommands.put(msgId,waiter);// 发送消息MqttManager.publish(cmd.getTopic(),cmd.getPayload());// 等待响应，带超时try{booleansuccess=waiter.await(DEFAULT_TIMEOUT,TimeUnit.MILLISECONDS);if(success){returnwaiter.getResult();}else{// 超时了returnCommandResult.timeout();}}finally{waitingCommands.remove(msgId);}}/** * 收到指令结果时调用 */publicvoidonCommandResult(StringmsgId,CommandResultresult){CommandWaiterwaiter=waitingCommands.get(msgId);if(waiter!=null){waiter.setResult(result);waiter.countDown();}}}

重试策略

指令超时了要不要重试？分情况：

核心原则：有副作用的、不能重复执行的指令，不要自动重试，让上层业务决定。

四、离线消息补发处理

设备断网期间，云端下发的指令会存在broker里，设备上线后一股脑全发过来。

离线消息的问题

1. 过期指令：断了一天网，上线后收到昨天的开门指令，还有意义吗？
2. 重复指令：云端可能重发了好几次，离线消息里有重复的
3. 顺序问题：多条指令的执行顺序有没有要求？

处理策略

publicclassOfflineMessageHandler{privatestaticfinallongMSG_EXPIRE_TIME=5*60*1000;// 消息5分钟过期/** * 处理收到的消息（包括离线的） */publicvoidhandleMessage(MqttMessagemsg){CommandMessagecmd=parseCommand(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){longage=System.currentTimeMillis()-cmd.getTimestamp();returnage>MSG_EXPIRE_TIME;}}

不同类型指令的过期时间

五、消息日志与溯源

出了问题要能查，所以每条消息的处理都要有日志。

消息日志表设计

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)-- 失败原因);

日志的作用

1. 问题排查：用户说"我扫码没开门"，查一下收到指令没、什么时候处理的
2. 统计分析：指令成功率、平均处理时长、重复率、过期率
3. 对账：开门次数和交易订单对不对得上

六、关键指令的幂等设计

去重是第一层，幂等是第二层。就算去重失效了，重复执行也不能出问题。

什么是幂等？

同一个指令执行一次和执行N次，结果是一样的。

各类指令的幂等实现

1. 开门指令

开门是最危险的，必须保证幂等。

publicclassDoorCommandExecutor{/** * 幂等开门：同一个订单号只开一次门 */publicbooleanopenDoor(StringorderId,intdoorNo){// 用订单号做幂等键if(hasOpenedForOrder(orderId)){Log.w("Door","订单 "+orderId+" 已经开过门了，跳过");returntrue;// 也算成功}// 执行开门booleansuccess=DoorController.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()-lastRebootTime<5*60*1000){Log.w("Reboot","短时间内重复重启请求，忽略");return;}lastRebootTime=System.currentTimeMillis();doReboot();}

七、消息优先级与队列

设备端收到的消息有急有缓，不能按收到顺序挨个处理。

优先级划分

优先级队列实现

publicclassPriorityMessageQueue{privatePriorityQueue<QueuedMessage>queue=newPriorityQueue<>((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消息可靠性相关的坑，欢迎私信交流~