1. 从一次订单超卖事故说起为什么消息中间件能治分布式事务的“痛”分布式事务这个话题几乎每个做微服务的人都绕不开。我印象特别深早些年参与过一个电商下单链路的改造当时订单服务扣完库存、写完订单库再去调用支付服务结果支付服务超时了。订单库回滚了可库存服务那边因为网络抖动没收到回滚指令库存白白少了一件。这种“一半成功一半失败”的场景就是分布式事务最典型的痛点。传统单体应用靠数据库的本地事务就能搞定一致性一个Transactional注解基本解决大部分问题。可一旦拆成多个服务、多个数据库本地事务就鞭长莫及了。这时候业界有几条主流路线两阶段提交2PC、TCC、Saga、本地消息表、事务消息等。而 RabbitMQ 在这套体系里扮演的角色主要是通过可靠消息最终一致性来解决问题——它不追求强一致而是保证“消息一定发出去、一定被消费、消费失败能重试”最终让各个服务的数据达成一致。这篇是 SpringBoot 整合 RabbitMQ 系列的收尾篇前面几篇我们聊了基础收发、交换机类型、死信队列、延迟队列、消息确认机制这些内容。到了最后一篇我想把前面所有零散的知识点串起来落到一个真实场景上用 RabbitMQ 的可靠消息机制来解决分布式事务。如果你正在做微服务拆分或者被跨服务数据一致性折磨过这篇内容应该能给你一条可落地的思路。整篇我会从原理讲到代码再讲我踩过的坑尽量让你看完就能动手试。需要先说明一点RabbitMQ 解决分布式事务不是银弹它适合的是“允许短暂不一致、最终必须一致”的业务场景比如订单创建后发优惠券、支付成功后更新积分、下单后扣减库存这类。如果你要求的是银行转账那种强一致那还是得看 TCC 或者 Seata 这类方案。搞清楚适用边界比盲目上工具重要得多。2. 可靠消息最终一致性的核心原理拆解2.1 本地消息表与消息中间件的分工要理解 RabbitMQ 怎么解决分布式事务得先搞清楚“可靠消息最终一致性”这套方案的骨架。它的核心思想其实很朴素把“业务操作”和“发消息”这两件事通过本地事务绑在一起保证它们要么都成功要么都失败。然后再靠消息中间件的重试和确认机制保证消息最终被消费方处理。具体来说生产方这边通常有两种做法。一种是本地消息表在业务库里建一张消息表业务操作和往消息表插一条记录放在同一个本地事务里。事务提交后再有一个后台任务或者定时任务去扫描消息表把待发送的消息投递到 RabbitMQ投递成功后更新消息状态。另一种是利用 RabbitMQ 的发送方确认机制Publisher Confirm配合手动事务或者数据库状态标记来实现。消费方这边则要保证幂等消费。因为消息可能重复投递消费方必须能识别“这条消息我处理过了”否则就会出现重复扣款、重复发券这种事故。常见的幂等方案有用业务唯一 ID 做去重表、用 Redis 记录已处理的消息 ID、或者利用数据库唯一索引。提示可靠消息最终一致性的“最终”两个字很关键。它不保证实时一致中间可能有一个短暂的不一致窗口业务上要能容忍这个窗口。2.2 三个必须守住的环节发、存、消我把这套方案拆成三个必须守住的环节任何一个环节掉链子整个一致性就崩了。第一是“发得出去”。生产方把消息投递到 RabbitMQ 的 Broker这一步要保证消息不丢。RabbitMQ 提供了 Publisher Confirm 机制Broker 收到消息后会回一个 ack生产方收到 ack 才能确认发送成功。如果没收到 ack就要重发。这里要注意Confirm 是异步的得用回调或者CorrelationData来关联。第二是“存得住”。消息到了 Broker 之后默认是存在内存里的Broker 重启就没了。所以要开启持久化交换机持久化、队列持久化、消息持久化三件套。但光持久化还不够如果队列满了或者磁盘故障消息还是可能丢。生产环境一般会配合镜像队列或者仲裁队列来做高可用。第三是“消费得掉”。消费方拿到消息后业务处理可能失败。这时候不能直接 ack要让消息重新入队或者进死信队列配合重试机制。但重试次数不能无限否则会一直卡住。通常设置一个最大重试次数超过之后进死信队列由人工或者补偿任务处理。这三个环节对应到代码里就是发送方确认、消息持久化、消费方手动 ack 加幂等。下面我会逐个展开。2.3 为什么不用 XA 或者 2PC有朋友可能会问RabbitMQ 本身支持 AMQP 协议AMQP 里也有事务txSelect、txCommit为什么不直接用原因很简单RabbitMQ 的事务性能太差它会把整个通道变成同步阻塞模式吞吐量断崖式下跌。而且 AMQP 事务只能保证“消息发送到 Broker”这一段的原子性管不了消费方那边。至于 XA 和 2PC它们要求所有参与方都支持 XA 协议数据库还好说但很多中间件和第三方服务根本不支持。而且 2PC 有同步阻塞、协调者单点、数据不一致等固有问题在互联网高并发场景下基本被淘汰了。所以可靠消息最终一致性才成了主流选择——它用“异步 重试 幂等”换来了性能和可用性。3. 生产端落地把消息可靠地送进 Broker3.1 环境准备与依赖配置先把环境搭起来。我用的是 SpringBoot 2.7 加 RabbitMQ 3.11依赖就两个核心的dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-jdbc/artifactId /dependency配置文件里要开启发送方确认和返回机制spring: rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest publisher-confirm-type: correlated publisher-returns: true template: mandatory: true这里publisher-confirm-type设成correlated是关键它允许我们在发送时带一个CorrelationData回调里能拿到具体是哪条消息被确认了。publisher-returns配合mandatory: true用来处理“消息到了交换机但路由不到队列”的情况。3.2 本地消息表的设计与写入时机本地消息表我一般设计成这几个字段消息 ID唯一、业务类型、消息内容JSON、状态待发送/已发送/已消费、重试次数、创建时间、更新时间。建表语句大概这样CREATE TABLE t_local_message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, msg_id VARCHAR(64) NOT NULL UNIQUE, biz_type VARCHAR(32) NOT NULL, payload TEXT NOT NULL, status TINYINT NOT NULL DEFAULT 0, retry_count INT NOT NULL DEFAULT 0, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL );写入时机非常讲究必须和业务操作在同一个本地事务里。比如下单场景扣库存、写订单、插消息表这三步要放在一个Transactional方法里。这样只要事务提交消息记录就一定在库里事务回滚消息记录也不会存在。这就解决了“业务成功但消息没发”的问题。Transactional(rollbackFor Exception.class) public void createOrder(OrderDTO dto) { orderMapper.insert(dto.toOrder()); stockMapper.deduct(dto.getSkuId(), dto.getCount()); LocalMessage msg new LocalMessage(); msg.setMsgId(UUID.randomUUID().toString()); msg.setBizType(ORDER_CREATED); msg.setPayload(JSON.toJSONString(dto)); msg.setStatus(0); msg.setCreateTime(new Date()); msg.setUpdateTime(new Date()); localMessageMapper.insert(msg); }3.3 定时投递与发送方确认的配合消息写进本地表之后需要一个投递器把它送到 RabbitMQ。我一般用一个定时任务每隔几秒扫一批状态为“待发送”的记录逐条投递。投递时带上CorrelationData在 Confirm 回调里更新消息状态。Scheduled(fixedDelay 3000) public void deliverMessages() { ListLocalMessage list localMessageMapper.selectPending(100); for (LocalMessage msg : list) { CorrelationData cd new CorrelationData(msg.getMsgId()); rabbitTemplate.convertAndSend(order.exchange, order.created, msg.getPayload(), cd); } }Confirm 回调这样写rabbitTemplate.setConfirmCallback((correlationData, ack, cause) - { if (ack correlationData ! null) { localMessageMapper.updateStatus(correlationData.getId(), 1); } else { localMessageMapper.increaseRetry(correlationData.getId()); } });这里有个细节投递和更新状态之间不是原子的。可能出现消息投递成功但更新状态失败的情况导致下次扫描又投一遍。所以消费方必须做幂等这一点后面会讲。另外重试次数要设上限超过上限的消息标记为异常转人工处理避免无限重试拖垮系统。注意定时任务的扫描频率和批量大小要结合业务量调。频率太高会给数据库压力太低会导致消息延迟。我一般从 3 秒扫 100 条起步根据积压情况再调。4. 消费端落地幂等、重试与死信兜底4.1 手动 ack 与消费失败的处理策略消费端第一件事是把自动 ack 关掉改成手动 ack。自动 ack 的问题是消息一投递给消费者就立刻确认如果消费者处理到一半挂了消息就丢了。手动 ack 则是在业务处理成功后才确认失败就不确认让消息重新入队。spring: rabbitmq: listener: simple: acknowledge-mode: manual prefetch: 10消费代码RabbitListener(queues order.queue) public void onMessage(Message message, Channel channel) throws IOException { String msgId message.getMessageProperties().getMessageId(); long tag message.getMessageProperties().getDeliveryTag(); try { if (idempotentService.isProcessed(msgId)) { channel.basicAck(tag, false); return; } orderService.handle(JSON.parseObject(new String(message.getBody()))); idempotentService.markProcessed(msgId); channel.basicAck(tag, false); } catch (Exception e) { channel.basicNack(tag, false, false); } }这里basicNack的第三个参数设成false表示不重新入队直接进死信队列。为什么不设true因为如果设成true消息会立刻重新入队然后又被这个消费者拿到又失败形成死循环。正确做法是配合重试次数超过次数就进死信。4.2 幂等设计的三种常见方案对比幂等是消费端的生命线。我总结下来有三种常用方案各有适用场景方案实现方式优点缺点适用场景去重表数据库建唯一索引表处理前插入强一致可靠有数据库写入开销金融、订单类Redis 标记用 SETNX 记录消息 ID性能高Redis 故障可能丢标记高并发、可容忍少量重复业务唯一键利用业务本身的唯一约束无额外开销依赖业务设计有天然唯一键的场景我一般首选去重表因为最稳。去重表就两个字段消息 ID 和创建时间消息 ID 上加唯一索引。处理前先insert如果抛唯一键冲突异常说明处理过了直接 ack。public boolean isProcessed(String msgId) { try { idempotentMapper.insert(new Idempotent(msgId, new Date())); return false; } catch (DuplicateKeyException e) { return true; } }4.3 死信队列作为最后一道防线死信队列DLX是这套方案的最后兜底。当消息重试超过上限、或者被basicNack拒绝、或者队列过期时消息会进入死信交换机再路由到死信队列。我们可以在死信队列上挂一个消费者专门处理这些“疑难杂症”。配置死信队列Bean public Queue orderQueue() { return QueueBuilder.durable(order.queue) .withArgument(x-dead-letter-exchange, order.dlx) .withArgument(x-dead-letter-routing-key, order.dead) .build(); } Bean public Queue orderDeadQueue() { return QueueBuilder.durable(order.dead.queue).build(); }死信消费者里可以做告警、落库、人工补偿。我一般会把死信消息存到一张异常表里然后发告警通知让运维或者开发介入。这样即使自动流程全挂了消息也不会凭空消失。提示死信队列一定要配监控和告警否则消息进了死信没人知道等于没兜底。5. 完整链路串讲一次下单场景的全流程复盘5.1 从用户点击下单到库存扣减的时序把前面的碎片拼起来走一遍完整流程。用户点击下单订单服务做三件事写订单、扣库存、插本地消息表这三步在一个本地事务里。事务提交后定时任务扫描到这条待发送消息投递到order.exchange路由到order.queue。RabbitMQ 收到消息后回 Confirm生产方更新消息状态为“已发送”。消费方比如积分服务从队列拿到消息先查去重表没处理过就执行业务逻辑然后写去重表、手动 ack。如果处理失败basicNack让消息进死信死信消费者记录异常并告警。整个链路里任何一个环节失败都有对应的补偿生产方投递失败会重试消费方处理失败会进死信Broker 宕机有持久化和镜像队列兜底。这就是“最终一致”的完整闭环。5.2 消息状态流转与对账补偿光靠自动流程还不够生产环境一定要有对账。我一般会做一个定时对账任务比如每小时跑一次比对本地消息表里“已发送但未消费”的记录和消费方的处理记录。如果发现某条消息发出去了但消费方一直没处理就触发补偿。对账的粒度可以按业务类型分比如订单类消息对账频率高一点通知类消息可以低一点。对账发现的不一致要么重发消息要么人工介入。这一步是很多团队容易忽略的但恰恰是保证长期一致性的关键。5.3 性能与可靠性的权衡取舍这套方案不是没有代价的。本地消息表多了一次数据库写入定时任务有扫描延迟消费方幂等判断也有开销。所以要在性能和可靠性之间做取舍。我的经验是核心链路订单、支付用完整的本地消息表方案非核心链路通知、日志可以简化比如直接用 Confirm 机制不做本地消息表。另外批量投递、异步 Confirm、合理设置 prefetch 这些都能提升吞吐。别为了追求极致可靠把所有链路都搞成重装步兵那样系统会又慢又复杂。6. 那些年我踩过的坑与实战经验6.1 消息重复消费的隐蔽来源很多人以为重复消费只来自重试其实不然。我遇到过几次诡异的重复消费一次是生产方投递成功但更新状态失败定时任务又投了一遍一次是消费方 ack 时网络抖动Broker 没收到 ack消息重新入队还有一次是镜像队列切换消息被重复投递。这些场景说明重复消费是常态不是异常。所以幂等设计不能有侥幸心理必须每个消费方都做。而且幂等判断要放在业务逻辑之前不能等业务执行完再判断否则重复执行已经发生了。6.2 Confirm 回调丢失与状态不一致Confirm 回调是异步的如果应用在回调执行前重启回调就丢了消息状态永远停在“待发送”。这时候定时任务会重新投递导致重复。解决办法是投递前先更新状态为“发送中”回调里再更新为“已发送”并且对“发送中”超过一定时间的记录做补偿扫描。另外 Confirm 回调里不要做太重的操作比如查库、调远程服务否则会阻塞回调线程。我一般只做状态更新复杂的补偿逻辑放到独立线程池里。6.3 死信队列积压的排查思路死信队列积压是常见问题。排查思路是先看死信消息的内容和异常堆栈判断是业务 bug 还是数据问题再看积压速度是突增还是缓慢增长最后看消费方日志确认是消费逻辑失败还是根本没消费。我遇到过一次死信暴增最后发现是消费方依赖的一个下游服务挂了导致所有消息处理超时。这种时候光重试没用得先恢复下游服务再把死信消息重新投递回原队列。所以死信队列最好配一个“重新投递”的管理接口方便批量处理。6.4 几个容易被忽略的配置细节最后分享几个配置上的坑。第一prefetch不要设太大设太大会导致消息都堆在一个消费者那里其他消费者闲着一般设 10 到 50 之间。第二队列的x-message-ttl和x-max-length要谨慎设置设不好会导致消息被提前丢弃。第三生产环境一定要开publisher-confirm和publisher-returns别嫌麻烦这是消息不丢的基础。还有一点SpringBoot 的RabbitListener默认是单线程消费高并发场景要配concurrency参数比如concurrency 5-10让多个消费者并行处理。但并发数也不是越大越好要结合数据库连接池和下游服务承受能力来定。这套方案我在几个项目里落地过整体是稳的但前提是每个环节都做扎实。分布式事务没有一劳永逸的解法可靠消息最终一致性只是其中一条相对成熟的路。真正上线前压测、对账、告警这三样一个都不能少。