RabbitMQ分布式事务解决方案

[主业务] --> [本地事务]
[本地事务] --> [RabbitMQ发送准备消息]
[RabbitMQ] --> [从业务]
[从业务] --> [事务协调]
[事务协调] --> [最终提交]

@Transactional
public void createOrder() {
    // 业务逻辑...
}

CREATE TABLE trans_coordinator (
  tx_id VARCHAR(64) PRIMARY KEY,
  status ENUM('prepared','committed','rollbacked'),
  create_time DATETIME
);

message.setCorrelationId(txId);
rabbitTemplate.send("prepared.exchange", 
                   "prepared.route", 
                   message);

UPDATE trans_coordinator 
SET status='prepared' 
WHERE tx_id='tx123';

@RabbitListener(queues = "queue.prepared")
public void handle(String txId) {
    // 幂等检查
    if(checkDuplicate(txId)) return;
  
    try {
        doBiz();
        sendAck(txId);
    } catch(Exception e) {
        sendNack(txId);
    }
}

// 成功响应
rabbitTemplate.send("ack.exchange", 
                   txId, 
                   "SUCCESS");

@RabbitListener(queues = "queue.ack")
public void coordinate(String txId, String result) {
    updateTxResult(txId, result);
  
    if(allAckReceived(txId)) {
        finalCommitOrRollback(txId);
    }
}

[定时任务] --> [扫描prepared状态]
[扫描prepared] --> [超时5分钟] --> [发起回滚]

SELECT tx_id FROM trans_coordinator 
WHERE status='prepared' 
AND create_time < NOW()-INTERVAL 5 MINUTE;

1. 主服务 -> MySQL: 开启事务
2. MySQL -> 主服务: 成功响应
3. 主服务 -> RabbitMQ: 发送准备消息
4. RabbitMQ -> 从服务: 投递消息
5. 从服务 -> RabbitMQ: 返回ACK
6. RabbitMQ -> 主服务: 反馈结果
7. 主服务 -> MySQL: 最终提交
8. MySQL -> 主服务: 确认完成