面试官问到我这个问题时我第一反应是他真正想听的并不是“我会用同步发送”这种单点答案而是想看我能否把“消息不丢失”这件事拆成端到端的整体设计。RocketMQ作为消息中间件可靠性从来不是某一个开关或者某一段代码能保证的而是生产端、Broker存储端、消费端三个环节共同作用的结果。今天我就用自己的实际运维经验把这条链路上的每一个风险点怎么堵、每个参数怎么调、每个坑怎么踩一次性讲透。不管你是准备面试还是线上真的需要配置高可靠的消息服务这份梳理都可以直接拿去参考。先明确一件事我们说的“不丢失”在分布式系统里并不等于“一条都不能少”的绝对保证通常指的是“至少投递一次”。RocketMQ默认就是这种语义也就是每条消息在极端情况下可能重复但绝不允许丢失。理解了这一点后面所有做法都围绕一个核心原则——生产端确认Broker收到、Broker确认落盘并复制、消费端确认处理完再提交偏移量。1. 消息不丢失的定义与三大环节拆解1.1 什么才是“消息不丢失”很多初学者一上来就问“怎么保证不丢失”但没想过业务对消息的容忍度。比如订单状态变更、支付回调这类场景丢了就是事故必须做到一条都不能少而像日志上报、用户行为采集这种偶尔丢一两条可能影响不大。RocketMQ官方提供的能力是“至少一次”也就是保证每条消息都会送达但重复消费是可能的。所以你在设计时首先要跟业务约定清楚如果真的追求“恰好一次”那需要在消费侧做幂等这是另一个话题。我见过不少团队盲目开启“所有队列都同步刷盘”以为自己绝对安全了结果吞吐掉了一大截最后又不得不改回来。这就是没搞清楚“不丢失”的真正瓶颈在哪里。事实上只要把生产、存储、消费三段的丢失风险逐一堵上就能达到实际业务可接受的高可靠标准。1.2 消息在RocketMQ里走一趟哪些地方会丢一条消息从生产者发出去到消费者真正处理中间要经过三个物理阶段每个阶段都有独立的失控可能生产端生产者把消息发到Broker时如果网络抖动、Broker宕机或者发送超时没收到确认这条消息就丢了。存储端Broker收到消息后先写入内存再决定是否刷到磁盘。如果写入内存但来不及刷盘Broker进程突然崩溃内存数据就没了。消费端消费者拉取消息处理完后需要向Broker提交偏移量Offset表示“我已经处理完了”。如果没提交就宕机重启后会重新消费旧消息这不丢但可能重复反过来如果自动提交了偏移量但业务逻辑没真正执行完那这条消息等于永久丢失。所以所谓“保证不丢失”本质上是在这三个环节分别选择“可靠的确认方式”。下面我按顺序讲每一环怎么操作最后给你一套可以抄作业的参数组合。2. 生产端不丢消息的第一道防线2.1 别用异步发送除非你确认能接受丢失RocketMQ发送消息有三种方式同步发送、异步发送、单向发送。其中单向发送根本不管Broker是否收到只要发完就算结束这种肯定有丢的风险异步发送虽然有个回调可以感知发送结果但很多人写完回调就打个日志后续没有重试逻辑跟单向几乎没区别。真正可靠的生产方式只有同步发送它会阻塞等待Broker返回一个写盘成功的响应。我建议所有核心业务消息一律使用同步发送。上线前最好在代码里做一次测试把Broker故意停掉同步发送的客户端会立刻得到异常跑不了下一行而异步发送可能看起来“成功了”其实消息早丢了。下面是一段常见可靠发送配置可以用Java客户端实现DefaultMQProducer producer new DefaultMQProducer(order_producer); producer.setNamesrvAddr(127.0.0.1:9876); producer.setRetryTimesWhenSendFailed(5); // 失败自动重试 producer.setSendMsgTimeout(3000); // 发送超时时间 producer.start(); // 同步发送并检查发送结果 SendResult result producer.send(new Message(order_topic, tag, body)); if (result.getSendStatus() ! SendStatus.SEND_OK) { // 这里要记录到本地日志或数据库后续补发 }注意setRetryTimesWhenSendFailed虽然可以自动重试但它默认的重试策略在同一Broker上重发如果Broker挂了重试也是徒劳。所以我在实际项目中还会在外面包一层业务级重试机制把发送失败的消息先落到本地一张pending表里每隔一段时间扫描重发。这听起来多一套表但真能救急尤其是网络抖动超过默认重试时长的时候。2.2 事务消息本地事务和发消息的原子性还有一种比同步发送更稳妥的写法就是事务消息。它的核心价值是解决“本地数据库写成功了但消息没发出去”的经典问题。比如你下单流程里要插一条订单数据同时发一条消息给下游如果先插库再发消息发消息失败就会留下脏数据如果先发消息再插库又可能消息已经出去了订单却回滚没建成。RocketMQ的事务消息做法是先发一个半消息Half MessageBroker先把消息存起来但不对消费者可见然后本地执行事务逻辑最后向Broker提交Commit或回滚Rollback。如果本地事务卡住没给结果Broker会反向询问事务状态这样无论如何事务都会在本地真正结束之后才决定发送。我在电商项目里用事务消息处理过类似库存扣减和优惠券发放的一致性问题。当时踩过一次细节事务消息所在的普通topic消费端必须配一个TransactionListener实现来检查本地事务状态具体如下producer.setTransactionListener(new TransactionListener() { Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { // 这里执行本地数据库事务成功返回COMMIT_MESSAGE失败返回ROLLBACK_MESSAGE return orderService.createOrder((String) arg) ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE; } Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { // 回调检查本地表是否有记录 return orderService.hasOrder(msg.getTransactionId()) ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.UNKNOW; } });注意事务消息并不能保证消息永远不丢它只是保证“要么事务都成功要么都不成功”。一旦你提交了后续还是要靠Broker存储层来兜底。3. 存储端Broker的持久化保障3.1 同步刷盘和异步刷盘选错就要付代价Broker收到消息后默认先写进内存的PageCache然后由系统异步刷到物理磁盘。这种异步刷盘方式性能极好但不是宕机安全的因为内存数据丢失后无法恢复。想要把丢失风险压到接近零就得把刷盘策略改成同步刷盘也就是每写完一条消息必须强制flush到磁盘后才返回成功。在RocketMQ的broker.conf里这样配置关键参数# 刷盘策略 flushDiskTypeSYNC_FLUSH # 也可以利用默认的异步复制但注意一致性短板 # 同步刷新单个文件延迟单位毫秒 syncFlushTimeout5同步刷盘的代价是吞吐量显著下降大概会损失三到五成的写入性能。我的实际建议是核心交易topic用同步刷盘普通日志topic用异步刷盘。我在某二线电商平台部署时就是按照“支付同步刷、曝光日志异步刷”来分流既保证了核心链路安全又没牺牲整体吞吐。还有一个细节坑同步刷盘虽然保证了单机写入但如果你用异步刷盘却被外界包装成“高可靠”那等于白搭运维文档一定要写明当前topic到底是什么刷盘级别。3.2 主从复制模式Master挂了Slave要能顶上来刷盘解决的是单机重启丢数据。但如果遇到物理机器断电、磁盘损坏那磁盘上的数据也保不住。RocketMQ的应对方案是多副本也就是Master和Slave之间做复制。复制模式同步复制的意思是Master写入成功之后要等Slave也追加成功才给生产者返回成功异步复制则是Master写入完成就立即返回Slave的复制过程允许稍微延迟。生产环境我强烈建议把核心集群配置成SYNC_MASTER。配置在broker.conf里brokerRoleSYNC_MASTER # 磁盘不作伪SLAVE常规配置 # 从节点配置 brokerRoleSLAVE这里有个容易被忽略的点主从同步复制虽然能避免“Master写成功但Slave没复制”的情况但如果Master宕机且Slave自动切换未同步到Slave的消息还是会丢。所以最稳妥的组合是“同步刷盘 同步复制”也就是说我上一节说的SYNC_FLUSH和这里的SYNC_MASTER要搭配使用缺一个都达不到真正的高可靠。我自己在实际压测中发现双同步模式下的单机QPS大概能到两万左右已经能满足绝大多数中等规模业务的峰值了。另外RocketMQ 4.5之后支持用Raft协议在多副本上实现自动选主也就是所谓的DLedger模式。这个模式比传统的Master/Slave更智能不需要人工手动切换节点我建议对可用性要求高的新项目直接走DLedger配置一个节点的DLedger集群配合官方文档写好dLgerPeers可以省去大量手工运维成本。4. 消费端确认机制与偏移量管理4.1 手动提交偏移量别开自动提交消费端最容易丢消息的地方是“偏移量提交与业务处理不同步”。RocketMQ默认消费者是自动提交偏移量的也就是拉取到消息后过一段时间自动提交当前消费位置。如果业务处理逻辑抛出异常但偏移量已经被提交掉那这条消息就永远从Loader的角度消失了。这也是新手最容易踩的坑之一。解决方案只有一条手动提交偏移量并且保证先处理完业务逻辑再提交。代码示例Java版DefaultMQPushConsumer consumer new DefaultMQPushConsumer(order_consumer); consumer.setNamesrvAddr(127.0.0.1:9876); consumer.subscribe(order_topic, *); consumer.setMessageListener(new MessageListenerConcurrently() { Override public ConsumeConcurrentlyStatus consumeMessage(ListMessageExt msgs, ConsumeConcurrentlyContext context) { try { for (MessageExt msg : msgs) { // 处理业务写库、调接口等等 orderService.handleMessage(new String(msg.getBody())); } // 全部处理成功手动返回CONSUME_SUCCESS等价于提交偏移量 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } catch (Exception e) { // 处理失败返回RECONSUME_LATERRocketMQ会稍后重试 return ConsumeConcurrentlyStatus.RECONSUME_LATER; } } });这里有个细节如果你想彻底避免“处理失败却被提交”就不要依赖API的返回值而是自己显式调用consumer.updateOffset(msgQueue, offset, true)同时将autoCommit关掉。不过这样会引入不少额外复杂度一般优先使用返回状态的方式。注意RECONSUME_LATER不会无限重试下去重试超过16次后会进入死信队列因此你还得给死信队列加一个专门的监听实现人工治理。4.2 消费重试与死信队列的兜底RocketMQ默认消费失败后会重试重试次数可以配置。DefaultMQPushConsumer可以设定重试次数consumer.setMaxReconsumeTimes(3); // 最多重试3次超过进死信重试不是立即的它有个退避策略比如第1次延迟10秒第2次延迟30秒……这个设计很人性化但也会让消息看起来“卡住”了。我碰到过一个场景业务方等不及重试误以为消息丢了直接重启消费进程结果消息又被处理一遍。我当时给他解释如果看到消息积压且一直在重试说明消费端有脏数据导致处理失败要去看异常日志而不是盲目重启。死信队列DLQ保存的是重试次数用完但依然处理失败的消息。默认topic名称是%DLQ%consumerGroup你可以专门启动一个消费进程去消费这个队列把死信消息捞出来做补偿或人工处理。注意死信队列里的消息并不会自动被删我这里常用的巡检策略是每天扫一次死信队列消息量一旦出现积压就第一时间跟进。5. 端到端最佳配置与防丢实战5.1 一套稳妥的参数组合直接抄作业聊完三段各自的做法我最后总结一套在线上运行两年多的配置模板。生产端一律同步发送 重试3次 外部业务表兜底Broker端每个核心topic的Broker节点设置flushDiskTypeSYNC_FLUSH和brokerRoleSYNC_MASTER消费端全链路手动提交 统一幂等设计 死信队列监控。这样的组合理论上可以做到任意一点故障都不会丢业务消息只要你的网络没有两层脑裂这种极端情况。配置落在一个核心服务上大概长这样# broker.conf 核心配置 brokerClusterNameDefaultCluster brokerNamebroker-a namesrvAddr192.168.1.10:9876;192.168.1.11:9876 flushDiskTypeSYNC_FLUSH brokerRoleSYNC_MASTER syncFlushTimeout5 # DLedger模式如有需要配置消费者客户端初始化时还要把consumeFromWhere设置为CONSUME_FROM_LAST_OFFSET还是CONSUME_FROM_FIRST_OFFSET这个取决于你重启时要不要从头补数据。如果业务不能接受旧数据重复我建议在消费端自己做幂等表而不要天真地用“从头开始”去补救。5.2 我踩过的丢消息坑每条都是血泪第一是生产端使用异步发送后回调里只打了日志没有重发结果Broker短暂宕机一批订单消息静默丢失。后来我强制规定所有业务必须用同步发送只有埋点类才允许异步。第二是Broker配置了异步复制主库宕机后从库切上来数据发现少了最近20秒的攒单。后来我把副本模式改为同步复制这个隐患直接消除。第三是消费端自动提交偏移量导致某次异常大批量丢消息。当时业务方反馈“消息明明处理失败却显示成功”排查半天发现是默认自动提交在作怪。从那以后我要求所有核心消费组必须手动提交。另外还有一个经典误区Kafka那套acksall、min.insync.replicas的经验直接套到RocketMQ上。RocketMQ没有acks配置对应的是生产端同步发送加Broker同步复制。两者模型不一样别绕晕了。6. 常见问题与排查技巧实录6.1 如何判断“消息确实丢了”排查消息丢失最直接的手段是看监控指标。RocketMQ社区版自带的Dashboard会给出三个关键数字生产TPS、消费TPS、消费积压量。如果某个topic的生产TPS突然升高而消费TPS却没有跟进或者积压迟迟不降就可能存在消费丢失。再进一步可以通过生产端的show lastMsgID和消费端的show lastConsumerOffset比对链路向前推找到差距在哪一段。我建议你至少在生产端和消费端都打上唯一消息ID的业务日志然后做到“每个关键消息从发送到处理都有一条贯通日志”。这样丢失时你能快速定位是生产确认没收到、Broker排推失败还是消费端偏移量跳过了。实测中这种日志方式比任何监控面板都更能救命。6.2 重试与阻塞不要用无限重试掩盖代码缺陷有些同学为了不丢消息把所有下游接口都包上无限重试逻辑结果下游一旦有问题线程被死死卡住积压越来越多最终消息队列被拖垮。我的想法是重试次数必须设上限它只是解决瞬时故障不能替代业务监控和补偿。超过上限就进死信队列然后由运维人员介入分析根本原因。往往代码Bug修完了重试次数根本用不完。另外消费组里的实例数不要随便扩容。如果逻辑上不幂等扩了实例虽然吞吐上去了但同一条消息可能被多个消费者同时拿到引发重复处理。所以我在做消费幂等时常用的方案是给每条消息加一个全局唯一ID消费前先查一遍数据库里是否已有这个ID的操作记录。这是最稳妥的兜底没有之一。6.3 事务消息的半消息处理失败事务消息有个容易忽略的小坑如果客户端在executeLocalTransaction里抛了异常事务状态会变成UNKNOWBroker会反复调用checkLocalTransaction直到到达一个最大次数。如果这个回调一直没有明确结果消息就会一直挂在半消息状态。我见过半夜线上告警提示大量半消息积压排查后才发现是本地查询方法用了旧的主键字段根本查不到。所以事务消息的本地表一定要设计好唯一标识和状态字段查询方法必须稳定可用。把这个点最后再写一遍RocketMQ不是靠某个神秘开关防止丢失它是一套需要你根据业务特性和系统规模去拼装的可靠性策略。我在生产环境跑了小两年最终的体会是三个环节里最容易“意外丢失”的大多不是Broker硬件故障而是配置不一致、客户端使用不当、运维监控缺失。只要这三件事做到位——生产同步发送加兜底、Broker同步刷盘并同步复制、消费端手动提交加幂等——消息不丢失是很稳的。
