RabbitMQ 分布式事务：基于消息表的柔性事务实践

💡 message_id 用于保证消息幂等性，建议使用 UUID 或雪花 ID。

2. Order Service 代码实现

Maven 依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
</dependencies>

配置文件 application.yml

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/order_db?useSSL=false&serverTimezone=UTC
    username: root
    password: root
  jpa:
    hibernate:
      ddl-auto: update
      show-sql: true
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: correlated
    publisher-returns: true
    template:
      mandatory: true

实体类

// Order.java
@Entity
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private Long userId;
    private Long productId;
    private Integer quantity;
    private String status = "CREATED";
    // getters & setters
}

// MessageOutbox.java
@Entity
@Table(name = "message_outbox")
public class MessageOutbox {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String messageId; // 全局唯一
    private String eventType;
    private String payload; // JSON 字符串
    private String status = "PENDING";
    private LocalDateTime createdAt = LocalDateTime.now();
    private LocalDateTime sentAt;
    // getters & setters
}

服务层：创建订单并记录消息

@Service
@Transactional
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;
    @Autowired
    private MessageOutboxRepository messageOutboxRepository;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void createOrder(Long userId, Long productId, Integer quantity) {
        // 1. 创建订单
        Order order = new Order();
        order.setUserId(userId);
        order.setProductId(productId);
        order.setQuantity(quantity);
        order.setStatus("CREATED");
        orderRepository.save(order);

        // 2. 构造消息内容
        Map<String, Object> event = new HashMap<>();
        event.put("orderId", order.getId());
        event.put("productId", productId);
        event.put("quantity", quantity);
        String payload = new ObjectMapper().writeValueAsString(event);

        // 3. 保存消息到 outbox 表（与订单在同一事务）
        MessageOutbox message = new MessageOutbox();
        message.setMessageId(UUID.randomUUID().toString());
        message.setEventType("ORDER_CREATED");
        message.setPayload(payload);
        messageOutboxRepository.save(message);
    }
}

✅ 关键：@Transactional 确保订单和消息表写入原子性。

3. 异步投递消息：MessagePublisher

我们需要一个后台任务，定期扫描 message_outbox 表，将 PENDING 状态的消息发送到 RabbitMQ。

@Component
public class MessagePublisher {
    private static final Logger log = LoggerFactory.getLogger(MessagePublisher.class);
    
    @Autowired
    private MessageOutboxRepository messageOutboxRepository;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    // 每 10 秒执行一次
    @Scheduled(fixedDelay = 10_000)
    @Transactional
    public void publishPendingMessages() {
        List<MessageOutbox> pendingMessages = messageOutboxRepository.findByStatus("PENDING");
        for (MessageOutbox msg : pendingMessages) {
            try {
                // 发送消息到 RabbitMQ
                rabbitTemplate.convertAndSend("order.exchange", "order.created", msg.getPayload());
                // 标记为已发送
                msg.setStatus("SENT");
                msg.setSentAt(LocalDateTime.now());
                messageOutboxRepository.save(msg);
                log.info("Message published: {}", msg.getMessageId());
            } catch (Exception e) {
                log.error("Failed to publish message: {}", msg.getMessageId(), e);
                // 不更新状态，下次重试
            }
        }
    }
}

⚠️ 注意：即使 RabbitMQ 发送失败，也不更新消息状态，确保下次重试。

4. Inventory Service：消费消息并扣库存

消费者代码

@Component
public class InventoryConsumer {
    private static final Logger log = LoggerFactory.getLogger(InventoryConsumer.class);
    
    @Autowired
    private InventoryService inventoryService;

    @RabbitListener(queues = "inventory.queue")
    public void handleOrderCreated(String payload, Channel channel,
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            // 1. 解析消息
            ObjectMapper mapper = new ObjectMapper();
            JsonNode event = mapper.readTree(payload);
            Long orderId = event.get("orderId").asLong();
            Long productId = event.get("productId").asLong();
            Integer quantity = event.get("quantity").asInt();

            // 2. 幂等性检查（可选：记录已处理的消息 ID）
            if (inventoryService.isMessageProcessed(event.get("messageId").asText())) {
                log.info("Duplicate message ignored: {}", event.get("messageId"));
                channel.basicAck(deliveryTag, false);
                return;
            }

            // 3. 执行扣库存
            inventoryService.deductInventory(productId, quantity);

            // 4. 记录消息已处理（用于幂等）
            inventoryService.markMessageAsProcessed(event.get("messageId").asText());

            // 5. 手动 ACK
            channel.basicAck(deliveryTag, false);
            log.info("Inventory deducted for order: {}", orderId);
        } catch (Exception e) {
            log.error("Failed to process message", e);
            try {
                // 拒绝消息，requeue=false 表示不重回队列（避免无限循环）
                // 可配置死信队列进行后续处理
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ioEx) {
                log.error("Failed to nack message", ioEx);
            }
        }
    }
}

幂等性处理（InventoryService）

@Service
public class InventoryService {
    @Autowired
    private ProcessedMessageRepository processedMessageRepository;
    @Autowired
    private InventoryRepository inventoryRepository;

    @Transactional
    public void deductInventory(Long productId, Integer quantity) {
        Inventory inventory = inventoryRepository.findByProductId(productId);
        if (inventory.getStock() < quantity) {
            throw new RuntimeException("Insufficient stock");
        }
        inventory.setStock(inventory.getStock() - quantity);
        inventoryRepository.save(inventory);
    }

    public boolean isMessageProcessed(String messageId) {
        return processedMessageRepository.existsByMessageId(messageId);
    }

    @Transactional
    public void markMessageAsProcessed(String messageId) {
        ProcessedMessage record = new ProcessedMessage();
        record.setMessageId(messageId);
        processedMessageRepository.save(record);
    }
}

✅ 幂等性表 processed_message 防止重复消费导致多次扣库存。

五、RabbitMQ 队列与交换机配置

为了实现可靠投递和死信处理，我们需要合理配置 RabbitMQ 的 Exchange、Queue 和 DLX。

声明 Bean（Order Service）

@Configuration
public class RabbitMQConfig {
    // 交换机
    @Bean
    public TopicExchange orderExchange() {
        return new TopicExchange("order.exchange");
    }

    // 死信交换机
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dlx.exchange");
    }

    // 死信队列
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable("dlq.inventory").build();
    }

    // 绑定 DLQ 到 DLX
    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("inventory.dlq");
    }

    // 库存队列（带死信配置）
    @Bean
    public Queue inventoryQueue() {
        return QueueBuilder.durable("inventory.queue")
                .withArgument("x-dead-letter-exchange", "dlx.exchange")
                .withArgument("x-dead-letter-routing-key", "inventory.dlq")
                .withArgument("x-message-ttl", 30000) // 30 秒 TTL，用于延迟重试（可选）
                .build();
    }

    // 绑定库存队列到主交换机
    @Bean
    public Binding inventoryBinding() {
        return BindingBuilder.bind(inventoryQueue()).to(orderExchange()).with("order.created");
    }
}

🔁 通过 TTL + DLX 可实现延迟重试：消息消费失败后进入 DLQ，再通过另一个消费者将消息重新投递回原队列（需额外逻辑）。

六、流程图：可靠消息最终一致性

下图展示了从订单创建到库存扣减的完整链路，以及失败处理机制：

sequenceDiagram
    participant User
    participant OrderService
    participant MySQL
    participant RabbitMQ
    participant InventoryService

    User->>OrderService: 下单请求
    OrderService->>MySQL: 开启事务
    OrderService->>MySQL: 插入订单
    OrderService->>MySQL: 插入消息表 (PENDING)
    MySQL-->>OrderService: 事务提交
    OrderService->>RabbitMQ: 异步投递消息
    RabbitMQ->>InventoryService: 消费消息
    InventoryService->>InventoryService: 幂等性检查
    InventoryService->>InventoryService: 扣减库存
    InventoryService->>RabbitMQ: 手动 ACK

该图清晰展示了从订单创建到库存扣减的完整链路，以及失败处理机制。

七、柔性事务的其他模式

除了'可靠消息最终一致性'，还有几种常见的柔性事务模式：

1. TCC（Try-Confirm-Cancel）

• Try：预留资源（如冻结库存）；
• Confirm：确认操作（真正扣减）；
• Cancel：取消预留（释放资源）。

适用于对一致性要求较高的场景（如金融交易），但实现复杂，需业务侵入。

2. Saga 模式

• 将长事务拆分为多个本地事务；
• 每个步骤有对应的补偿操作；
• 通过事件或命令驱动。

适合长流程业务（如旅行预订），但补偿逻辑复杂。

3. 最大努力通知（Best-Effort Notification）

• 发起方不断重试通知接收方；
• 接收方提供查询接口供对账；
• 常用于支付回调等场景。

实现简单，但依赖人工对账兜底。

八、RabbitMQ 柔性事务的优缺点分析

优点

• 解耦：服务间通过消息通信，降低耦合度；
• 异步：提升系统吞吐量和响应速度；
• 可靠：通过 ACK、DLQ、重试机制保障消息不丢失；
• 最终一致：在可接受时间内达成数据一致；
• 扩展性强：新增消费者无需修改生产者。

缺点

• 最终一致性：不能满足强一致性需求；
• 复杂度：需处理幂等、重试、死信、监控等问题；
• 调试困难：分布式链路追踪成本高；
• 消息堆积风险：消费者宕机可能导致消息积压。

💡 建议：在非核心链路（如日志、通知、积分）优先使用柔性事务；核心资金链路考虑 TCC 或 Saga。

九、生产环境最佳实践

1. 消息幂等性必须保证

• 消费端通过唯一 ID（如 messageId）去重；
• 使用数据库唯一索引或 Redis SETNX 实现。

2. 合理设置重试策略

• 初次失败立即重试（3 次）；
• 后续采用指数退避（如 1m, 5m, 30m）；
• 最终进入人工审核队列。

3. 监控与告警

• 监控 RabbitMQ 队列长度、消费者速率；
• 对 DLQ 消息设置告警；
• 记录消息全链路 TraceID。

4. 消息内容设计

• 包含足够上下文（如订单 ID、用户 ID）；
• 避免大消息（建议 < 1MB）；
• 使用 JSON Schema 约束格式。

5. 本地事务与消息发送顺序

• 先写 DB，再发消息 → 可能消息丢失；
• 先发消息，再写 DB → 可能消息重复；
• 正确做法：本地事务 + 消息表（本文方案）。

十、常见问题与解决方案

Q1：消息发送成功，但消费者没收到？

• 检查 RabbitMQ 队列绑定是否正确；
• 查看消费者是否在线、是否有异常；
• 启用 mandatory=true + ReturnCallback 捕获路由失败。

Q2：消费者处理慢，导致消息堆积？

• 增加消费者实例（水平扩展）；
• 优化业务逻辑（如批量处理）；
• 设置合理的 prefetch count（如 10）。

Q3：如何防止消息重复消费？

• 消费端实现幂等（如数据库唯一约束）；
• 生产者生成全局唯一 messageId；
• 避免自动 ACK，使用手动 ACK。

Q4：RabbitMQ 宕机怎么办？

• 集群部署（3 节点以上）；
• 消息持久化（durable=true）；
• 生产者启用 confirm 模式。

十一、替代方案对比：Kafka vs RocketMQ vs RabbitMQ

💡 如果业务强依赖事务消息，可考虑 RocketMQ 的 Half Message 机制。

十二、总结：柔性事务是分布式系统的必修课

在微服务时代，没有银弹，只有权衡。RabbitMQ 提供的柔性事务方案，虽不能保证强一致性，但以较低的复杂度实现了高可用、高扩展的最终一致性，非常适合大多数互联网业务场景。

关键成功要素：

• 本地事务 + 消息表 保证发送可靠性；
• 手动 ACK + 幂等消费 防止数据错乱；
• 死信队列 + 重试机制 提升容错能力；
• 监控告警 + 人工兜底 保障系统健壮。

🌟 记住：柔性事务不是'不一致'，而是'延迟一致'。只要设计得当，它完全可以支撑亿级用户的高并发系统。

希望本文能为你在分布式事务的迷宫中点亮一盏灯。Happy Coding！🚀