“消息不丢”是一条链，不是某个开关
很多人以为把 RocketMQ 的刷盘方式改成同步、把重试次数调大，消息就不会丢了。实际上消息从生产到消费要经过三个环节，任何一环没有兜底都会丢：生产端发送失败、Broker 还没落盘就宕机、消费端处理到一半进程被杀。可靠性是这三段分别设计出来的，不是某个参数一键开启的。
三个环节分别怎么保证
| 环节 | 典型失败 | 兜底手段 |
|---|---|---|
| 生产 | 发送超时、Broker 拒收 | 同步发送 + 校验 SendStatus + 本地消息表/事务消息 |
| 存储 | 刷盘前宕机、主从异步复制 | 同步刷盘、主从同步复制、DLedger |
| 消费 | 处理中进程退出、业务异常 | 手动返回状态、重试队列、死信队列 + 幂等 |
生产端最容易被忽略的是“发送成功”的判断。send 返回的 SendStatus 不是 SEND_OK 时，消息可能只写到了从节点或者根本没落盘，必须当成失败处理。异步发送加回调虽然吞吐高，但回调里抛出的异常没有人接，等于把失败静默掉了。
消费端要区分“业务失败”和“系统异常”。参数错误、订单不存在这类重试也没用的错误，应该记录后返回成功（或直接进死信），让消息离开重试队列；数据库超时、下游 503 这类临时故障才返回重试。全部返回重试会让一条坏消息卡住整个队列。
事务消息：让“本地事务 + 发消息”变成一件事
RocketMQ 的事务消息把本地事务和消息发送绑定：先发半消息（对消费者不可见），执行本地事务，再根据结果提交或回滚；如果执行到一半进程挂了，Broker 会回查本地事务状态。
@RocketMQTransactionListener
public class OrderTransactionListener implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
try {
orderService.createOrder((CreateOrderCommand) arg);
return RocketMQLocalTransactionState.COMMIT;
} catch (BizException e) {
log.warn("本地事务失败，回滚半消息", e);
return RocketMQLocalTransactionState.ROLLBACK;
} catch (Exception e) {
log.error("本地事务结果未知，等待 Broker 回查", e);
return RocketMQLocalTransactionState.UNKNOW;
}
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
String orderNo = String.valueOf(msg.getHeaders().get("orderNo"));
return orderMapper.existsByOrderNo(orderNo)
? RocketMQLocalTransactionState.COMMIT
: RocketMQLocalTransactionState.ROLLBACK;
}
}
发送端用 RocketMQTemplate.sendMessageInTransaction，把业务参数通过 arg 传进去，同时把业务唯一键放进消息头，供回查使用。回查逻辑必须能根据这个消息头判断本地事务到底成没成，所以“订单号”这类业务键一定要在发送前就确定好。
事务消息解决的是“生产端不丢”，不解决“消费端不重复”。RocketMQ 保证的是至少一次投递，重复消费是正常现象。
重复消费：幂等键不能用 msgId
消费端最常见的错误是拿 msgId 做幂等键。RocketMQ 在重试时会生成新的消息 ID，同一个业务消息重投后 msgId 变了，幂等判断直接失效。正确的幂等键来自消息体里的业务唯一键，比如订单号、支付流水号。
落库时用唯一索引兜底，而不是“先查后插”：
INSERT INTO consume_record (consumer_group, biz_key, created_at)
VALUES (?, ?, NOW(3));
插入冲突就说明处理过，直接返回成功。业务处理和这条记录放在同一个本地事务里，失败一起回滚。Redis 的去重只能当第一层拦截，不能当最终防线。
常见坑与排查
坑一：消费者吞异常。 try-catch 住所有异常，最后返回 CONSUME_SUCCESS，消息看起来消费成功，实际业务没执行。捕获后要么返回重试，要么写失败表并告警，不能静默。
坑二：事务消息回查读到未提交的数据。 本地事务还没提交，回查就查不到订单，于是回滚了半消息，但事务随后提交成功，数据就丢了。回查必须能读到已提交状态，必要时给本地事务加状态字段并等提交完成。
坑三：@RocketMQTransactionListener 没生效。 监听器的 rocketMQTemplateBeanName 要和发送方使用的 template 对应，生产组名不匹配时监听器不会被调用，半消息永远不提交。
坑四：只配了同步刷盘，没配同步复制。 主节点刷盘成功后宕机，从节点还没同步到，切换后消息就没了。对可靠性要求高的 topic 要同时开启同步刷盘和同步复制。
坑五：重试消息把消费打满。 一条持续失败的消息会不断进入重试队列，占用消费线程。设置最大重试次数，超过后进入死信队列，并给死信配告警和人工处理入口。
排查时按“生产有没有成功 → Broker 有没有落盘 → 消费有没有 ack”三段查：先在生产者日志里找发送结果和 msgId，再用 mqadmin topicStatus、consumerProgress 看堆积和位点，然后按业务键查消费记录表和死信队列，最后对照消息轨迹确认消息到底走到了哪一步。
评论