做后端开发的迟早要跟消息队列打交道RocketMQ 消息丢失这个问题几乎每个用过 MQ 的团队都会踩上一轮。我自己就经历过凌晨被值班电话叫起来核对账单不平的情况查到最后问题出在一条消息在发送阶段被静默丢弃了。那之后我把 RocketMQ 从生产到消费的整条链路重新捋了一遍把所有可能导致丢消息的环节都列出来逐个加固。这篇文章就围绕“MQ 消息丢失怎么办”这件事把生产端、Broker 存储端、消费端三个阶段的可靠性机制、参数配置和排查方法完整拆开讲适合正在用 RocketMQ 做订单、支付、库存等核心链路的同学参考。1. 消息丢失的三个核心环节生产、存储、消费1.1 为什么偏偏是这三段RocketMQ 的一条消息从诞生到被业务处理完路径并不复杂生产者把消息发给 BrokerBroker 把消息写到磁盘和内存消费者从 Broker 拉取消息然后执行本地业务逻辑。消息丢失本质上就是在这三段路径的某一处断了。很多人一听到“消息丢失”第一反应就是去看消费逻辑有没有问题但其实生产端丢消息的概率一点不比消费端低。我见过一个典型的案例某团队上线了一个基于异步发送的库存扣减功能发送消息后既不检查发送结果也没注册失败回调上线三天就出了两笔库存扣减异常。原因很简单异步发送时如果 Broker 返回失败、或者网络抖动超时这条消息就直接被丢掉业务完全无感知。把问题拆成“生产、存储、消费”三段还有个好处排查时可以逐段定位。生产端丢的是“还没发出去”存储段丢的是“发出去了但没落盘/没同步”消费端丢的是“消息到了但被消费方错误处理”。三个环节的排查手段、配置参数完全不同搞清楚在哪一段丢才能对症下药。1.2 生产阶段最常见的丢失场景生产端丢消息常见原因有四类。第一类是发送方式选错。RocketMQ 提供同步发送、异步发送、单向发送三种方式。单向发送只负责把消息发出去不管结果丢不丢全看网络心情绝对不能用于核心数据链路。异步发送如果不注册回调失败时你根本不知道这类问题最隐蔽。第二类是超时设置太短。默认发送超时时间通常是 3000 毫秒在高延迟或 Broker 繁忙时很容易超时超时后如果没触发重试或者重试次数不够消息就丢了。很多新手把超时时间调到 1000ms 以为可以提升性能实测下来反而不如默认值可靠。第三类是发送结果SendResult没有被正确判断。同步发送时会有一个返回结果但sendStatus为SEND_OK不代表万无一失。比如发送时 Broker 返回了SLAVE_NOT_AVAILABLE说明主节点已写入但备节点不可写这时候如果主节点随后宕机消息依然可能丢。第四类是重试机制被忽略。生产端的重试不是无限次的默认配置下发送失败会重试两次总共最多尝试三次。如果消息在重试间隔内持续发不出去最终还是会被丢弃。我见过有人把重试次数改成 0理由是“消息丢了可以从业务库重推”这种思路在低频场景也许可行但一旦高频量跑起来补偿机制根本跟不上的时候账面就会对不平。1.3 Broker 存储阶段哪些配置会导致丢消息消息到了 Broker不代表已经安全了。Broker 收到消息后先把消息写到内存 PageCache再由后台线程落盘。这个“先写内存、后落盘”的设计既是提升吞吐的关键也是丢消息的根源之一。如果 Broker 配置的是异步刷盘ASYNC_FLUSH消息写入 PageCache 后就向生产者返回成功磁盘上的数据实际上还没来得及写。这时候如果机器突然断电或者进程被强制杀掉只要 PageCache 还没刷到磁盘消息就丢了。很多测试环境为了性能开了异步刷盘上线的时候忘了改回来这是非常危险的一件事。除了刷盘方式主从复制模式也直接影响消息可靠性。ASYNC_MASTER模式下主节点写入成功了就返回成功消息会异步复制到从节点。一旦主节点在复制完成前宕机这部分没来得及同步的消息就永久丢失了。对应地SYNC_MASTER必须等消息在主节点和从节点都写入成功后才返回成功可靠性更高代价是每条消息都要多等一次跨节点的同步。还有一个容易被忽略的点是自动创建 Topic。如果开启了autoCreateTopicEnable消息发送到一个不存在的主题时会自动创建。这在测试阶段很方便但线上开启后会因为路由信息不同步而出现写入失败也会因为创建出的队列不够导致消息分布极度不均间接造成消费积压和超时丢失。1.4 消费阶段什么情况下会丢消息消费端丢消息往往是“业务代码写出来的”问题而不是 RocketMQ 本身的问题。RocketMQ 的消费模型是这样的消费者从 Broker 拉取一批消息执行本地业务逻辑如果返回CONSUME_SUCCESSBroker 才会把消费位点offset往后推进如果返回RECONSUME_LATER就会触发重试。但很多业务代码里习惯性地把整个处理逻辑包在一个大 try-catch 里捕获了所有异常并且没有重新抛出于是消息处理失败时消费者依然返回消费成功Broker 认为这条消息已经处理完了位点照常前进。等业务侧发现数据不对消息已经被“消费”掉了这是最常见的丢消息方式。另外并发消费模式下消费线程拿到消息后如果业务逻辑还在执行消息实际上已经被当前消费者“取走”了。如果业务处理时间过长触发了消息重新投递就会造成重复消费。RocketMQ 默认提供的是 At-Least-Once 语义也就是说消息保证不丢但可能重复。如果你不做幂等处理重复消费在数据层面造成的脏写某些时候比丢掉消息更可怕。还有一种比较冷门但确实会发生的场景事务消息。事务消息的流程是半消息先到 Broker本地事务执行成功后发送 commitBroker 才会把消息变成用户可见。如果本地事务执行成功了但 commit 请求因为异常没有送达或者事务状态没有通过回查补提交消息就一直停留在半消息状态。对于使用者来说这条消息相当于“没发生”同样属于丢失。2. 生产端防止消息丢失的配置与实操2.1 发送方式的选择核心链路必须用同步发送生产端防止消息丢失第一个决定就是发送方式。RocketMQ 的同步发送会在发消息后等待 Broker 返回SendResult整个过程是阻塞的。阻塞意味着吞吐量会下降但换来的是明确的发送结果。对订单、支付、余额变动这类消息丢一条都是事故必须用同步发送。异步发送则是在回调函数里处理结果适合对吞吐要求极高、且可以有补偿机制的日志类、统计类消息。用异步发送不是不行前提是你必须在回调里检查sendResult并且对发送失败的情况做处理。我这里说的处理包括写失败日志、落库标记、后续定时任务补偿而不是简单地打一条日志。单向发送是彻底的 fire-and-forget只用于可以容忍丢失的监控打点、日志上报场景。凡是核心数据只要用了单向发送不管后面监控多完善消息丢了你都很难查。我自己在代码审查时看到单向发送直接打回除非对方能明确说出这条数据丢了也没关系。// 同步发送示例 SendResult sendResult producer.send(msg, 5000); if (sendResult.getSendStatus() SendStatus.SEND_OK) { // 发送成功继续业务流转 } else { // 发送结果异常按失败处理 log.error(消息发送结果异常: {}, sendResult); }这里额外说一句SEND_OK也可能有水分。如果sendResult.getSendStatus()为SEND_OK但消息实际写入到了主节点、同步到了从节点那才算真的安全如果 Broker 端因为存储异常而走了降级路径SEND_OK也可能是假成功。要彻底确认需要结合 Broker 端监控和存储状态综合判断不过一般业务侧能做的就是看SendResult、开重试、盯监控。2.2 重试机制与超时时间设置同步发送也不是永远一次成功网络抖动随时会发生。RocketMQ 的生产端天然带了重试能力关键参数有两个retryTimesWhenSendFailed和sendMsgTimeout。retryTimesWhenSendFailed默认是 2意味着首次发送失败后还会再尝试 2 次总共 3 次。重试之间会有短暂的退避避免连续轰炸 Broker。对于核心链路我一般会把它调到 3 到 5 次但不要调得太大因为等待重试会占用线程资源太多次重试会拖垮生产者本身的吞吐能力。sendMsgTimeout默认是 3000 毫秒这个值在交易高峰期经常不够用。我见过一个场景某业务中午高峰时 Broker 写完一条消息后磁盘排序等待时间较长发送端 3 秒超时结果大量消息触发重试反而把负载拉得更高。后来把超时调到 5000 毫秒并且配合合理重试次数整体稳定性明显提升。参数调整示例DefaultMQProducer producer new DefaultMQProducer(my_producer_group); producer.setNamesrvAddr(127.0.0.1:9876); producer.setRetryTimesWhenSendFailed(3); producer.setSendMsgTimeout(5000); producer.start();有一点必须重点提醒重试会带来重复发送。如果第一次发送实际已经成功了只是网络回执超时重试会再次发同一条消息到 Broker最终消费者可能收到两条一模一样的消息。所以生产端用了重试之后消费端必须做幂等这是一个打包组合两个缺一个都会出事。2.3 生产端故障排查实录我在排查一次消息丢失时是这样一步步定位到生产端的。当时业务反馈某个积分发放接口偶尔不生效查看数据库发现部分用户没有积分流水。我先在生产者端找对应的发送日志结果发现根本没有发送记录。继续翻代码使用的是异步发送回调里只做了异常日志输出没有真正的失败补偿。进一步在 Broker 端查该 Topic 的写入量发现实际写入量和业务单量对不上。两端数据一核对问题就出在异步发送的成功率上。排查步骤整理如下先确认发送的 Topic、消息体是否正常看生产者日志中有没有SendResult输出。在消费端查看消息消费总数和生产端发送总数比对。如果生产端发送总数小于业务触发次数基本可以确定是发送阶段丢失。检查发送方式是不是异步或单向有没有处理回调结果。检查发送异常时是否走了重试重试次数、超时时间是否合理。检查 Namesrv 地址配置和 Topic 路由信息确认是否选对了队列。那次最终修复方案并不复杂把积分接口的消息改成同步发送加了发送失败重试同时在消费端做了幂等校验。改完后整整一周没有新增异常说明多数生产端消息丢失并不是不可解决只是你没去直面它。3. 存储端可靠性与持久化配置3.1 刷盘方式同步刷盘和异步刷盘怎么选Broker 端的可靠性第一个要看的就是刷盘策略。RocketMQ 的刷盘配置有两种SYNC_FLUSH和ASYNC_FLUSH。SYNC_FLUSH模式下消息写入内存之后必须由刷盘线程把这条消息对应的数据从 PageCache 强制刷到磁盘物理文件返回成功后才算这条消息真正被 Broker 接收。这个模式的核心价值是只要生产者拿到了成功结果就说明消息已经落盘断电、宕机都不怕丢。ASYNC_FLUSH模式下消息写入 PageCache 就返回成功真正的落盘由后台线程定时批量执行。它把刷盘这个耗时的 IO 操作从发送链路上剥离掉性能明显更高但存在一个窗口期如果消息刚写入 PageCache、还没落盘时进程崩溃或者机器断电这部分消息就丢了。怎么选核心交易链路比如支付、订单、积分一律SYNC_FLUSH。你牺牲的可能是部分吞吐但你换回的是“发送成功即安全”这个确定性。大量日志、监控、异步通知类消息可以接受极端情况下丢几条用ASYNC_FLUSH能省不少 IO 资源。实践中有个折中的办法把同步刷盘只用在核心 Topic 所在的 Broker 上而日志类 Topic 可以单独部署一套 Broker 环境用异步刷盘。同一个 RocketMQ 集群里不同 Topic 也可以放在不同的 Broker 组上这样按需分配可靠性成本和性能都能兼顾。3.2 主从复制模式对消息可靠性的影响单独一台 Broker 即使配置了同步刷盘也扛不住机器硬件故障。所以生产环境必须要有主从结构RocketMQ 的主从复制有两种模式SYNC_MASTER和ASYNC_MASTER。SYNC_MASTER模式下主节点成功写入并落盘后还需要把消息同步给从节点从节点写入成功后再回执给生产者。等于说一条消息同时存在两份物理副本之后发送方才拿到成功的结果这时候主节点再怎么出问题从节点上还有一份完整的消息可以接管。ASYNC_MASTER模式则不同主节点写入成功就返回消息复制到从节点是异步进行的。吞吐量比同步模式高但主节点故障的一瞬间尚未同步的一部分消息就找不回来了。我接触过的小团队往往在起步阶段就用单机 Broker或者用了ASYNC_MASTER加ASYNC_FLUSH的组合觉得开发测试都没问题。到了线上一个凌晨的磁盘故障就能把所有未同步消息全部带走这种教训很惨痛。生产环境的推荐组合是SYNC_FLUSHSYNC_MASTER双保险确保发送成功即双副本落地。如果你因为性能瓶颈不能接受同步刷盘至少要在主从层面用SYNC_MASTER兜底主节点没了还有从节点。但请注意如果主节点是突然断电而不是宕机从节点上的数据可能也因为没有落盘而丢失所以可靠性要求高的话不要在这上面抠成本。3.3 关键参数整理与配置示例Broker 端的关键配置集中在broker.conf里可以按下面这份基准配置来校准。# Broker 基础配置 brokerClusterNameDefaultCluster brokerNamebroker-a brokerId0 # 主从同步模式 brokerRoleSYNC_MASTER # 刷盘方式 flushDiskTypeSYNC_FLUSH # 自动创建Topic关闭避免路由问题 autoCreateTopicEnablefalse # 自动创建消费组关闭 autoCreateSubscriptionGroupfalse # 删除文件时间默认72小时 deleteWhen04 fileReservedTime72 # broker主从同步状态检查 slaveReadEnabletrue这份配置里有两个容易被忽略的细节。一个是autoCreateTopicEnablefalse。新手图省事会开自动创建但线上 Topic 必须在控制台或通过 API 预先创建并规划好队列数。自动创建的 Topic 往往只有 4 个队列高并发下容易把压力集中在个别 Broker 机器上。另一个是fileReservedTime。默认 72 小时意思是消息文件保留 72 小时。如果你需要从历史消息中找回数据这个时间要按业务需要调大比如保留 7 天对应的fileReservedTime168。当然保留时间越长磁盘占用越大需要根据容量评估这是一个存储规划和可靠性之间的平衡问题。配置改完记得重启 Broker并且要逐个节点滚动重启避免全部同时重启导致客户端连接全量中断。4. 消费端消息丢失的正确姿势4.1 消费进度更新机制先处理后提交先看一个 RocketMQ 消费的基本流程消费者从 Broker 拉取消息后会执行注册好的MessageListener回调业务代码处理完消息后返回ConsumeConcurrentlyStatus.CONSUME_SUCCESSBroker 收到这个返回值后才会把消费位点往后移动。反过来如果返回的是RECONSUME_LATER这条消息就会进入待重试状态位点不会前进。所以要理解消费端怎么丢消息核心就是一句话位点必须在业务处理成功后才能前进任何“先提交后处理”或者“无论什么异常都返回成功”的行为都是在制造消息丢失。实际业务代码里最常见的坑是下面这种写法boolean result processMsg(message); if (result) { return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } else { return ConsumeConcurrentlyStatus.RECONSUME_LATER; }看起来逻辑正常但如果processMsg内部把所有异常都吞了并且返回 true那问题就大了。比如调用下游接口失败后 catch 住打条日志就返回成功消息就算“处理成功”了可实际业务并没有正确执行等你想补偿的时候已经找不到这条消息。我自己在代码审查里的标准是processMsg不允许吞异常。正常返回表示业务确实执行完成异常必须向上抛出或者在 catch 后返回RECONSUME_LATER。4.2 消费重试与死信队列当消费者返回RECONSUME_LATER消息不会无限次重试。RocketMQ 对并发消费模式下的单条消息有最大重试次数限制默认情况下重试超过一定次数后消息会进入消息死信队列死信队列的命名规则是%DLQ%加消费组名。比如消费组名是consumerA死信队列就是%DLQ%consumerA。消息进了死信队列之后原来的正常队列就不会再消费这条消息了。此时如果没人监控死信队列消息相当于被“丢弃”了虽然它还在磁盘上但已经不在正常处理链路里。处理方案有两个层面。第一个层面是通知和干预对死信队列做监控一旦有消息进入就触发告警由人或者平台工具去重放。第二个层面是重试策略优化分析消息为什么会反复失败是下游服务不稳定、参数错误还是消息体本身有问题。如果是下游瞬时故障可以调大重试间隔如果是消息体数据问题重试再多也没用只能从源头修复。另外顺序消息和并发消息的重试行为不一样。顺序消息如果某条消息处理失败它会阻塞当前队列后续消息的处理重试时会一直卡住直到成功。这种情况下更要重视日志排查因为一条坏消息可能让整个队列堵死。下面是个简单的死信队列消费示例你可以做一个定期扫描任务来处理DefaultMQPushConsumer dlqConsumer new DefaultMQPushConsumer(%DLQ%consumerA); dlqConsumer.setNamesrvAddr(127.0.0.1:9876); dlqConsumer.registerMessageListener((message, context) - { // 把死信消息记录下来或者主动重放到原Topic handleDlqMessage(message); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; }); dlqConsumer.start();不过这里要提醒一点死信队列不是设计用来手动大批量重放的频繁把大量死信重新塞回业务 Topic 可能会导致资源占用。正常的做法是先修复根因再把死信消息一批一批地重放边放边观察。4.3 消费端幂等性设计RocketMQ 保证的是消息不丢但投递语义是 At-Least-Once消息可能会重复。对于支付成功通知、积分发放、库存扣减这类操作不加幂等就是给自己埋雷。幂等设计常见的有三种方式。第一种是在业务表上加唯一键。比如订单号加一个事件类型作为唯一索引重复消费时插入唯一键冲突catch 掉即可。第二种是利用 Redis 的SETNX做短时间去重。处理消息前先以消息唯一 ID 作为 key 执行SETNX拿到锁才继续处理。要注意给 key 设置过期时间避免 key 一直在 Redis 里堆积导致占用内存。第三种是基于状态机的版本号控制。比如处理消息时带上一个业务状态流转的版本号只有当前状态符合预期才执行更新否则直接跳过。我个人的习惯是能通过数据库唯一键解决的就优先用唯一键因为不需要依赖额外组件不会因为 Redis 缓存失效就把幂等抛掉。对于高频削峰场景则用 Redis 做前置去重再配合数据库唯一键做最终兜底两道防线下来重复问题基本可控。5. 消息丢失的排查与监控方案5.1 如何定位丢失发生在哪个环节消息丢了不能靠猜。我们要做的是把链路各环节的“已处理数据量”对账。先看生产端确认业务实际触发了多少次发送生产者日志里SendResult成功了多少次。再看 Broker这个 Topic 在某个时间窗口内的总写入量是多少。最后看消费端这个消费组实际拉取成功了多少条消费位点偏移到了哪里。三次数据一对比问题环节就浮出来了。发送成功数远小于业务触发数问题在生产端检查发送方式、重试、超时。生产端发送成功数正常但 Broker 实际存储数偏少问题在 Broker 存储检查刷盘、主从同步、磁盘状态。Broker 存储数正常但消费数明显滞后或不足问题在消费端或者消费积压检查消费异常、重试、死信队列。RocketMQ 控制台里Topic 状态和消费组状态面板可以用直接看消费位点和堆积数。另外用命令行工具也可以快速查消费进度sh mqadmin consumerStatus -g consumer_group_name sh mqadmin topicStatus -t topic_name这些命令在无控制台环境里非常实用。我记得排查某次故障时就是靠topicStatus发现某个分区的消息数量和预期差了一截然后顺藤摸瓜找到对应 Broker 节点确认它的刷盘模式是异步的最终定位到了问题。5.2 监控指标与告警配置防止消息丢失不只是在代码层面做正确还要让丢失在一开始就暴露出来。监控是关键。生产端需要盯的指标包括发送成功数、发送失败数、发送耗时、重试次数分布。这些数据可以从客户端统计并上报。实践中的做法是在 Producer 发送逻辑里加一层埋点按秒级聚合后在监控大盘上展示任何一个发送失败率升高都能第一时间发现。Broker 端需要盯的指标包括写入 TPS、刷盘耗时、PageCache 命中率、磁盘 IO 等待时间、主从同步延迟、已提交位点。刷盘耗时突然飙高通常意味着磁盘快要撑不住了此时消息落盘可能失败可靠性随之下降。消费端需要盯的指标包括消费 TPS、消费失败率、消费堆积量lag、死信队列数量、重试队列数量。堆积量飙升和死信消息进入是两个最最重要的告警项。建议的告警项至少要有下面这些监控对象告警条件含义生产失败率超过 0.1% 持续 5 分钟发送链路异常发送耗时P99 超过 2000ms写 Broker 太慢消费堆积lag 超过阈值消费能力不足死信队列任意死信队列有消息进入消费失败无法自动恢复主从同步延迟延迟超过 1 分钟容灾能力下降这些告警我在多个团队都实践过。尤其是死信队列告警一旦配置完成很多潜在的“静默丢失”都会被提前拦截住。消息进了死信队列业务可能还没感知但你的告警系统已经知道了。5.3 存量消息找回的几种手段监控只能防未来已经丢掉的消息还是要想办法找回来。根据消息丢失类型可以按优先级尝试不同手段。第一种消息在死信队列里。这是最理想的情况消息没有真正消失只是离开了正常队列。从死信队列里把消息重新投递回原 Topic或者用专门的补偿任务手动处理属于“重放”方案。第二种消息文件还在 Broker 上。如果磁盘文件还在保留周期内可以通过 RocketMQ 的导出工具或脚本按时间范围从 commitlog 文件里把 Broker 存储的消息捞出来再重新投递到一个指定的恢复 Topic。这种方式适合生产端确认发送过、但消费端没消费到的情况。第三种业务侧有发送记录表。这是最可靠的方式。生产者在发送前先把消息内容落一张业务表状态标记为待发送发送成功后再更新状态。丢失发生时只需要扫描这张表里状态异常的数据重新发送一遍即可。很多高可靠系统就是靠这张本地消息表兜底。第四种从下游系统逆向恢复。比如订单状态以数据库为准消息丢了就从数据库重新拉取变更记录生成新的消息。这个方法依赖业务表里有没有可追溯的修改时间或版本号适合订单、支付这类有强业务记录的场景。个人建议在核心系统里从第一步就引入“本地消息表 定时补偿”的机制不要等到出故障再想办法。RocketMQ 本身做得很可靠但任何中间件都经不起业务侧没有兜底。6. 我的一点实操体会关于 RocketMQ 消息丢失我个人的经验总结成一个原则不要寄希望于某一个配置提升可靠性而要把生产、存储、消费当成一个整体去设计。在实际项目里我会这样安排优先级。核心业务消息发送端全部用同步发送开启合适重试超时给足Broker 端配置SYNC_FLUSH和SYNC_MASTER关闭自动创建 Topic消费端处理逻辑不吞异常返回码区分成功与失败同时用数据库唯一键做幂等。这套组合下来消息可靠性已经非常高了换来的一点性能损失完全可接受。还有两个小技巧值得分享。第一个是把死信队列检查做成每日巡检任务哪怕只是看数量有没有变化也能避免死信悄悄积累。第二个是在生产环境的消费组上加一个“按时间回退位点”的能力万一发现批量消息处理异常可以在可控范围内回退消费位点重新消费这个操作要谨慎最好由有权限的人在低峰期执行。消息丢失这种问题一旦出了事故影响往往是资金、账目、用户权益层面。提前做的每一点配置和监控投入都是在帮自己减少一个深夜告警。希望这篇梳理对正在排查 RocketMQ 消息问题的你有用。
