1. 分布式事务的老大难问题到底难在哪做过微服务拆分的兄弟大概率都经历过这种场景用户下单订单服务写本地库成功了库存服务扣减库存的时候网络抖了一下超时了结果订单创建了、库存没扣两边数据对不上。更恶心的是这种问题往往不是必现的测试环境跑一百遍都没事生产环境高峰期偶尔冒出来一两条脏数据排查起来能把人逼疯。这就是分布式事务的经典困境。单体应用时代一个Transactional注解就能搞定的事情拆成微服务之后变成了跨进程、跨数据库的协调问题。核心矛盾在于每个服务只能保证自己本地事务的ACID但没有任何一个服务能单独保证跨服务操作的原子性。分布式事务的解决方案其实不少两阶段提交2PC、TCC、Saga、本地消息表、最大努力通知每种方案都有自己的适用场景和代价。但今天我要聊的是其中落地成本最低、对业务侵入最小的一种思路基于RabbitMQ消息中间件来实现最终一致性。这套方案在很多互联网公司的交易链路里跑了多年稳定可靠而且不需要引入额外的TC协调器组件。这篇文章适合谁看如果你正在做微服务拆分遇到了跨服务数据一致性问题又不想引入Seata这种重量级框架那RabbitMQ这套方案值得你花时间研究。如果你对RabbitMQ还停留在发消息、收消息的认知层面这篇文章也会帮你打开新世界的大门——原来消息队列还能这么用。注意本文讨论的是最终一致性方案不是强一致性。如果你的业务场景要求毫秒级强一致比如金融核心账务那这套方案不适合你老老实实上TCC或者2PC。2. 为什么选RabbitMQ而不是其他方案2.1 几种主流分布式事务方案的横向对比在决定用RabbitMQ之前我先把常见的几种方案拉出来遛一遛看看各自的优劣。方案一致性强度业务侵入性能开销落地难度适用场景2PC/XA强一致低高中传统金融、数据库层面TCC强一致极高中高资金交易、库存扣减Saga最终一致高低中长流程业务编排本地消息表最终一致中低低异步解耦场景RabbitMQ消息最终一致低低低异步解耦、事件驱动从表格能看出来RabbitMQ方案在业务侵入和落地难度上都有明显优势。TCC需要每个参与方实现try-confirm-cancel三个接口开发量直接翻三倍Saga需要写补偿逻辑回滚路径复杂本地消息表虽然简单但需要额外建表、写定时任务扫表而且消息表和业务表在同一个库数据库压力大的时候会互相影响。RabbitMQ方案的核心思路是把消息的发送和本地业务操作绑定在同一个本地事务里利用RabbitMQ的确认机制保证消息可靠投递消费方通过幂等处理保证消息重复消费不出问题。这样就把分布式事务拆解成了两个本地事务加一个可靠消息传递复杂度大幅降低。2.2 RabbitMQ的可靠投递机制是这套方案的基石很多人用RabbitMQ就是简单的basicPublish发出去就不管了消息丢了都不知道。但要拿它来解决分布式事务必须把RabbitMQ的可靠投递机制吃透。RabbitMQ保证消息不丢需要三个环节都做到位生产者到Broker开启publisher confirm机制。消息发送后Broker会异步回一个ack或者nack告诉你消息到底有没有落盘。没收到confirm的消息需要重发。这里有个坑很多人在confirmCallback里做重发但没考虑重发次数上限和幂等导致消息重复或者死循环。Broker内部交换机、队列、消息都要持久化。durabletrue的队列加上deliveryMode2的消息再配合镜像队列或者Quorum队列才能保证Broker宕机后消息不丢。单节点RabbitMQ没有高可用可言生产环境必须上集群。Broker到消费者关闭自动ack改成手动ack。消费者处理完业务逻辑后再basicAck处理失败就basicNack让消息重新入队或者进死信队列。自动ack模式下消费者拿到消息就ack了如果这时候进程挂了消息就丢了。这三个环节缺一不可。我见过太多团队只做了持久化没开confirm结果生产者发消息的时候网络闪断消息根本没到Broker业务方还以为发出去了。2.3 最终一致性的核心本地事务与消息发送的原子性这套方案最精妙的地方在于它把发消息这个动作和写业务数据这个动作放进了同一个本地事务。具体怎么做有两种主流做法做法一本地消息表。在业务库里建一张消息表业务操作和插入消息记录在同一个本地事务里。然后有一个独立的线程或者定时任务去扫这张表把消息投递到RabbitMQ投递成功后更新消息状态。这种做法最可靠但需要额外维护一张表和扫描逻辑。做法二RabbitMQ事务。RabbitMQ本身支持txSelect、txCommit、txRollback可以把消息发送和数据库操作放在一个分布式事务里。但说实话RabbitMQ的事务性能很差吞吐量会掉一个数量级生产环境基本没人用。做法三Confirm机制补偿。先发消息再执行本地事务如果本地事务失败就发一条补偿消息或者删除消息。这种做法实现简单但极端情况下可能消息发出去了、本地事务失败了需要消费方做额外的校验。实际落地中做法一和做法三结合使用比较多。核心业务用本地消息表保证绝对可靠非核心业务用Confirm补偿简化实现。3. 基于RabbitMQ的分布式事务完整实现方案3.1 整体架构设计与消息流转路径先把这个方案的架构图在脑子里画出来。假设我们有一个订单服务和库存服务用户下单需要同时创建订单和扣减库存。整个流程是这样的订单服务收到下单请求开启本地数据库事务在本地事务中插入订单记录同时插入一条消息记录到本地消息表状态为待发送提交本地事务一个独立的投递线程扫描本地消息表把待发送的消息投递到RabbitMQRabbitMQ把消息路由到库存服务的队列库存服务消费消息执行库存扣减扣减成功后手动ack扣减失败则nack并记录日志投递线程收到Broker的confirm后更新本地消息表状态为已发送这个流程里订单创建和消息记录写入是原子的消息投递和状态更新是最终一致的库存扣减通过手动ack保证至少消费一次。整个链路的最终结果是要么订单创建成功且库存扣减成功要么订单创建失败且库存不会扣减。提示库存服务消费消息时必须做幂等处理。因为网络抖动、消费者重启等原因同一条消息可能被投递多次。幂等方案可以用消息IDRedis去重或者数据库唯一索引。3.2 本地消息表的设计与关键字段说明本地消息表的设计直接决定了这套方案的可靠性。我一般会设计这么几个核心字段CREATE TABLE t_local_message ( id BIGINT PRIMARY KEY AUTO_INCREMENT, message_id VARCHAR(64) NOT NULL COMMENT 全局唯一消息ID, exchange_name VARCHAR(128) NOT NULL COMMENT 交换机名称, routing_key VARCHAR(128) NOT NULL COMMENT 路由键, message_body TEXT NOT NULL COMMENT 消息体JSON, status TINYINT NOT NULL DEFAULT 0 COMMENT 0-待发送 1-已发送 2-发送失败, retry_count INT NOT NULL DEFAULT 0 COMMENT 重试次数, next_retry_time DATETIME COMMENT 下次重试时间, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_message_id (message_id), KEY idx_status_next_retry (status, next_retry_time) );几个关键点解释一下message_id是全局唯一的用UUID或者雪花算法生成。这个ID会随消息一起发到RabbitMQ消费方用它来做幂等判断。status字段标记消息状态。0表示待发送投递线程只扫这个状态的数据。1表示已发送Broker已经confirm。2表示发送失败需要人工介入或者定时重试。retry_count和next_retry_time配合使用实现指数退避重试。第一次失败后1分钟重试第二次5分钟第三次30分钟超过最大重试次数就标记为失败并告警。idx_status_next_retry这个联合索引很重要。投递线程的查询条件是status0 AND next_retry_time NOW()没有这个索引的话消息表数据量大了之后扫描会非常慢。3.3 消息投递线程的实现细节与重试策略投递线程是整个方案的心脏。它的职责很单一不停地扫描本地消息表把待发送的消息投递到RabbitMQ根据投递结果更新消息状态。Component public class MessageDispatchTask { Autowired private LocalMessageMapper messageMapper; Autowired private RabbitTemplate rabbitTemplate; Scheduled(fixedDelay 1000) public void dispatch() { ListLocalMessage messages messageMapper.selectPendingMessages( LocalDateTime.now(), 100); for (LocalMessage msg : messages) { try { rabbitTemplate.convertAndSend( msg.getExchangeName(), msg.getRoutingKey(), msg.getMessageBody(), new MessagePostProcessor() { Override public Message postProcessMessage(Message message) { message.getMessageProperties() .setMessageId(msg.getMessageId()); message.getMessageProperties() .setDeliveryMode(MessageDeliveryMode.PERSISTENT); return message; } } ); messageMapper.updateStatus(msg.getId(), 1); } catch (Exception e) { int newRetryCount msg.getRetryCount() 1; if (newRetryCount MAX_RETRY) { messageMapper.updateStatus(msg.getId(), 2); alertService.sendAlert(消息投递失败, msg); } else { long delay calculateDelay(newRetryCount); messageMapper.updateRetryInfo(msg.getId(), newRetryCount, LocalDateTime.now().plusSeconds(delay)); } } } } }这段代码有几个细节值得展开说批量拉取每次拉100条避免一次拉太多导致内存溢出也避免一次拉太少导致吞吐量不够。这个数字可以根据消息量和处理速度调整。confirm机制上面代码用的是同步confirmconvertAndSend会阻塞直到Broker返回confirm。如果追求更高吞吐量可以用异步confirm在ConfirmCallback里更新消息状态。但异步confirm的代码复杂度会高不少需要维护一个未确认消息的映射表。重试延迟计算calculateDelay方法实现指数退避比如delay baseDelay * 2^(retryCount-1)baseDelay设为60秒。这样第一次重试等1分钟第二次2分钟第三次4分钟避免短时间内大量重试把Broker打挂。消息ID透传通过MessagePostProcessor把本地消息表的message_id设置到RabbitMQ消息的messageId属性上消费方可以直接从消息属性里拿到这个ID做幂等。实操心得投递线程最好独立部署不要和业务服务混在一起。业务服务重启的时候投递线程也会停消息就积压了。独立部署的话投递线程可以一直稳定运行不受业务发布影响。3.4 消费端的幂等处理与手动确认机制消费端要做两件事保证消息不丢手动ack和保证重复消费不出问题幂等。先看消费端的配置spring: rabbitmq: listener: simple: acknowledge-mode: manual prefetch: 10 retry: enabled: falseacknowledge-mode: manual开启手动ackprefetch: 10表示每个消费者最多预取10条消息避免消息分配不均。retry.enabled: false关掉Spring AMQP的自动重试我们自己控制重试逻辑。消费端的代码骨架RabbitListener(queues inventory.deduct.queue) public void handleInventoryDeduct(Message message, Channel channel) throws IOException { String messageId message.getMessageProperties().getMessageId(); String body new String(message.getBody(), StandardCharsets.UTF_8); try { // 幂等检查 if (idempotentService.isProcessed(messageId)) { channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); return; } // 执行业务逻辑 InventoryDeductDTO dto JSON.parseObject(body, InventoryDeductDTO.class); inventoryService.deduct(dto); // 标记已处理 idempotentService.markProcessed(messageId); // 手动ack channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (BusinessException e) { // 业务异常记录日志ack掉消息避免无限重试 log.error(库存扣减业务异常, messageId{}, messageId, e); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); } catch (Exception e) { // 系统异常nack并重新入队 log.error(库存扣减系统异常, messageId{}, messageId, e); channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); } }这里有几个关键决策点业务异常和系统异常要分开处理。业务异常比如库存不足重试多少次都没用直接ack掉记录日志人工处理。系统异常比如数据库连接超时可以重试nack后重新入队。幂等检查要在业务逻辑之前。先查Redis或者数据库如果这个messageId已经处理过了直接ack返回。注意幂等标记和业务操作最好在同一个本地事务里避免业务操作成功了但幂等标记没写进去。nack的requeue参数要慎重。basicNack(tag, false, true)会把消息重新放回队列头部如果这条消息一直处理失败会无限循环。更好的做法是nack到死信队列由死信队列的消费者做延迟重试。3.5 死信队列与延迟重试的配合使用死信队列DLX是RabbitMQ处理失败消息的利器。当消息被nack且requeuefalse或者消息过期或者队列满了消息就会进入死信交换机路由到死信队列。我们的重试策略可以这样设计// 声明业务队列时绑定死信交换机 Bean public Queue inventoryDeductQueue() { MapString, Object args new HashMap(); args.put(x-dead-letter-exchange, dlx.exchange); args.put(x-dead-letter-routing-key, inventory.deduct.dlq); args.put(x-message-ttl, 30000); // 消息30秒未消费进入死信 return new Queue(inventory.deduct.queue, true, false, false, args); } Bean public Queue inventoryDeductDlq() { return new Queue(inventory.deduct.dlq, true); }消费失败的消息进入死信队列后有一个专门的死信消费者来处理。死信消费者可以等一段时间再重新投递到业务队列实现延迟重试。或者直接记录到数据库由定时任务补偿。注意死信队列的消息也要设置过期时间或者最大长度否则死信消息越积越多最后把Broker磁盘撑爆。我见过一个线上事故就是因为死信队列没设上限跑了一个月积了几千万条消息。4. 生产环境踩过的坑与排查技巧4.1 消息重复消费的三种典型场景消息重复消费是这套方案最常见的线上问题。RabbitMQ保证的是at least once不是exactly once所以重复消费是必然的只能靠幂等兜底。场景一消费者ack超时。消费者处理完业务逻辑发送ack的时候网络抖动Broker没收到ack以为消费者挂了把消息重新投递给其他消费者。这时候业务逻辑已经执行了一次第二次消费就是重复。场景二消费者进程崩溃。消费者拿到消息业务逻辑执行到一半进程OOM挂了。Broker检测到连接断开把未ack的消息重新入队。重启后消费者再次消费这条消息。场景三投递线程重复投递。投递线程更新消息状态为已发送后数据库主从延迟导致下次扫描又扫到了这条消息重复投递到Broker。这三种场景都指向同一个结论幂等处理是消费端的必修课没有幂等就没有最终一致性。幂等方案的选择也有讲究Redis去重性能好但有丢失风险Redis挂了数据库唯一索引最可靠但性能差一些。核心业务建议用数据库唯一索引非核心业务可以用Redis。4.2 消息积压的紧急处理与预防措施消息积压是另一个高频问题。某天早上打开监控发现队列里积了几百万条消息消费速度远远跟不上生产速度。紧急处理三板斧第一扩容消费者。如果是消费者处理能力不足直接加机器。但要注意RabbitMQ的队列是独占的一个队列只能被一个消费者消费其实是轮询分发加消费者不一定能线性提升消费速度。这时候需要增加队列数量用一致性哈希或者sharding的方式把消息分散到多个队列。第二优化消费逻辑。看看消费端有没有慢查询、有没有不必要的远程调用、有没有可以批量处理的地方。我遇到过一个案例消费端每条消息都去查一次用户信息改成批量查询后消费速度提升了10倍。第三临时降级。如果积压太严重可以先ack掉消息记录到数据库等高峰期过了再慢慢处理。这种做法会丢失实时性但能快速恢复队列健康。预防措施方面监控告警是必须的。队列长度超过阈值就告警消费延迟超过阈值也告警。另外生产端的限流也很重要大促期间订单量暴涨如果生产端不限流消息瞬间涌入消费端肯定扛不住。4.3 消息丢失的排查思路与恢复方案消息丢失比重复消费更可怕因为重复消费只是多处理一次消息丢失意味着业务数据不一致。排查消息丢失按照消息流转路径逐段检查生产端有没有开启confirmconfirmCallback有没有正确处理nack本地消息表的记录有没有成功写入投递线程有没有正常运行Broker端队列和消息有没有持久化有没有配置镜像队列Broker有没有发生过重启或者磁盘满消费端有没有开启手动ack有没有在业务逻辑执行前就ack了有没有异常被吞掉导致消息没处理但ack了恢复方案取决于丢失的消息是否还能找回。如果本地消息表还有记录状态为待发送重新投递即可。如果本地消息表也没有记录那就只能通过业务对账来发现不一致然后人工补偿。实操心得建议每天跑一次对账任务比对订单表和库存表的数据。发现不一致的记录自动生成补偿消息重新投递。对账是最后一道防线不能省。4.4 常见问题速查表问题现象可能原因排查方法解决方案消息重复消费ack超时、消费者崩溃、投递重复查消费日志、查幂等记录加强幂等、用数据库唯一索引消息积压消费能力不足、消费逻辑慢查队列长度、查消费耗时扩容消费者、优化消费逻辑消息丢失confirm未开、未持久化、自动ack查confirm日志、查队列属性开启confirm、持久化、手动ack投递线程不工作线程挂了、数据库连接池满查线程状态、查数据库连接独立部署、加监控告警死信队列爆满没有设置TTL或最大长度查死信队列长度设置TTL、定期清理死信5. 几个容易忽略的细节与优化建议5.1 消息表的分库分表与归档策略本地消息表的数据量增长很快一个日订单10万的服务消息表一年就是3650万条数据。如果不做处理查询会越来越慢。分表策略上可以按月份分表每个月一张表。投递线程查询的时候只查当前月份的表历史表只用于归档和排查。归档策略上已发送状态的消息保留7天就可以删除了。删除的时候要批量删每次删1000条避免大事务。删除操作放在凌晨低峰期执行。-- 批量删除已发送的历史消息 DELETE FROM t_local_message WHERE status 1 AND update_time DATE_SUB(NOW(), INTERVAL 7 DAY) LIMIT 1000;5.2 消息体的序列化选择与版本兼容消息体的序列化格式直接影响兼容性和性能。JSON最通用但体积大、解析慢。Protobuf体积小、解析快但需要维护schema版本升级麻烦。我的建议是内部服务之间用JSON跨团队或者跨语言用Protobuf。JSON的可读性好排查问题的时候直接看消息内容就行。Protobuf虽然性能好但排查问题的时候需要反序列化多了一道工序。版本兼容方面消息体里最好带一个version字段。消费方根据version做不同的处理逻辑。新增字段要保证向后兼容不能删除已有字段只能标记为deprecated。5.3 监控告警体系的最小化搭建监控告警不需要搞得太复杂但几个核心指标必须有队列长度超过阈值告警。这个指标直接反映消费能力是否跟得上。消息投递延迟从消息写入本地消息表到投递到Broker的时间差。延迟过大说明投递线程有问题。消费失败率消费失败的消息数除以总消费消息数。失败率突增说明消费端有bug或者下游服务有问题。死信队列长度死信队列持续增长说明有大量消息处理失败需要人工介入。这些指标可以用Prometheus采集Grafana展示Alertmanager告警。如果团队没有这套基础设施用RabbitMQ自带的HTTP API也能拿到这些数据写个定时任务定期检查就行。5.4 什么场景不适合这套方案最后说一下这套方案的适用边界。以下几种场景不建议用RabbitMQ做分布式事务强一致性要求的场景。比如金融核心账务要求毫秒级强一致这套最终一致性方案满足不了。同步返回结果的场景。用户下单后要立即知道库存扣减结果这套异步方案做不到。需要配合前端的轮询或者WebSocket推送。超长事务链路。如果一笔业务涉及十几个服务的协同消息会经过多次转发链路越长出问题的概率越大排查也越困难。消息量极大的场景。RabbitMQ单队列的吞吐量有限如果每秒几十万条消息需要考虑Kafka或者Pulsar。我在实际项目中的体会是这套方案最适合的是交易链路中的异步解耦场景比如下单后异步扣库存、支付成功后异步发通知、订单完成后异步算积分。这些场景对实时性要求不高但对数据一致性有要求正好是RabbitMQ方案的甜区。还有一个小技巧分享本地消息表的投递线程和业务服务的数据库连接池要分开配置。投递线程如果和业务服务共用连接池业务高峰期连接池被占满投递线程拿不到连接消息就积压了。独立的数据源配置虽然麻烦一点但能避免互相影响。