分布式事务：2PC、TCC、Saga 与最终一致性

// 下单流程涉及三个微服务，三个数据库

订单服务（订单库）：创建订单记录
库存服务（库存库）：扣减商品库存
账户服务（账户库）：扣减用户余额

// 问题：如果订单创建成功，但库存扣减失败？
// 问题：如果库存扣减成功，但余额扣减失败？
// 需要保证三个操作要么全成功，要么全回滚

// 传统 @Transactional 只能管一个数据库
// 跨数据库、跨服务的事务 → 需要分布式事务方案

// 2PC 的四个致命问题

1. 同步阻塞：
   准备阶段所有参与者持有本地锁（如行锁），等待协调者的提交/回滚指令
   → 如果协调者响应慢，所有参与者的资源都被锁住
   → 高并发下性能极差

2. 单点故障：
   协调者挂了 → 所有参与者无限期阻塞
   参与者挂了 → 协调者等待超时后回滚

3. 数据不一致：
   阶段二发送 COMMIT 时，如果部分参与者收到、部分没收到
   → 有的提交了，有的没提交 → 数据不一致

4. 性能差：
   两次网络往返（prepare + commit），且全程同步阻塞
   → 吞吐量极低，不适合互联网高并发场景

// XA 是 X/Open 组织定义的分布式事务标准接口
// 大多数关系型数据库都支持 XA：MySQL、Oracle、PostgreSQL

XA 事务的四个阶段：
1. XA_START  — 开启分布式事务分支
2. XA_END    — 结束事务分支（但不提交）
3. XA_PREPARE — 准备阶段（参与者确认能否提交）
4. XA_COMMIT / XA_ROLLBACK — 提交或回滚

// MySQL 中的 XA 使用
XA START 'tx1';
UPDATE account SET balance = balance - 100 WHERE user_id = 'A';
XA END 'tx1';
XA PREPARE 'tx1';  -- 此时事务已持久化，但未提交
-- 协调者确认所有参与者都 PREPARE 成功后：
XA COMMIT 'tx1';

// Java 中使用 XA
XADataSource xaDs = new MysqlXADataSource();
xaDs.setUrl("jdbc:mysql://localhost/db1");
XAConnection xaConn = xaDs.getXAConnection();
XAResource xaRes = xaConn.getXAResource();

Xid xid = new XidImpl(1, "global-tx-1".getBytes(), "branch-1".getBytes());
xaRes.start(xid, XAResource.TMNOFLAGS);
// 执行 SQL...
xaRes.end(xid, XAResource.TMSUCCESS);
xaRes.prepare(xid);
xaRes.commit(xid, false);

阶段一：CanCommit（询问）
  协调者问参与者"你能提交吗？"
  参与者只做预检查，不执行事务
  → 返回 YES / NO

阶段二：PreCommit（预提交）
  如果全部 YES → 协调者发送 PreCommit 指令
  参与者执行本地事务（但不提交），写 redo/undo 日志
  → 返回 ACK

阶段三：DoCommit（提交）
  协调者收到所有 ACK → 发送 DoCommit
  参与者提交本地事务

// 3PC 相比 2PC 的改进
// 1. CanCommit 阶段不锁资源，减少阻塞时间
// 2. 引入超时机制：如果阶段三超时，参与者自动提交
//    （假设能进入阶段三说明之前都同意了）

// 但 3PC 仍然无法完全解决数据不一致：
// 网络分区时，部分参与者超时提交，部分等待回滚 → 仍然不一致

public class InventoryTccService {

    // Phase 1: Try — 冻结库存（不实际扣减）
    public boolean tryDeduct(String txId, String productId, int qty) {
        // 幂等检查：是否已经 Try 过
        if (tccLogRepository.exists(txId, "TRY")) return true;

        // 检查库存是否充足
        Inventory inv = inventoryRepo.findByProductId(productId);
        if (inv.getAvailable() < qty) return false;

        // 冻结库存：available -= qty, frozen += qty
        inv.setAvailable(inv.getAvailable() - qty);
        inv.setFrozen(inv.getFrozen() + qty);
        inventoryRepo.save(inv);

        // 记录 TCC 事务日志
        tccLogRepository.save(txId, "TRY", productId, qty);
        return true;
    }

    // Phase 2a: Confirm — 实际扣减冻结库存
    public void confirm(String txId, String productId, int qty) {
        // 幂等：已确认过就跳过
        if (tccLogRepository.exists(txId, "CONFIRM")) return;

        Inventory inv = inventoryRepo.findByProductId(productId);
        inv.setFrozen(inv.getFrozen() - qty); // 冻结减少
        inventoryRepo.save(inv);
        tccLogRepository.save(txId, "CONFIRM", productId, qty);
    }

    // Phase 2b: Cancel — 释放冻结库存
    public void cancel(String txId, String productId, int qty) {
        // 幂等：已取消过就跳过
        if (tccLogRepository.exists(txId, "CANCEL")) return;

        // 空回滚处理：Try 没执行就收到 Cancel
        if (!tccLogRepository.exists(txId, "TRY")) {
            tccLogRepository.save(txId, "CANCEL_EMPTY", productId, qty);
            return;
        }

        Inventory inv = inventoryRepo.findByProductId(productId);
        inv.setFrozen(inv.getFrozen() - qty);
        inv.setAvailable(inv.getAvailable() + qty);
        inventoryRepo.save(inv);
        tccLogRepository.save(txId, "CANCEL", productId, qty);
    }
}

public class OrderSagaOrchestrator {

    // Saga 步骤定义
    public Saga<OrderContext> buildSaga() {
        return Saga
            .<OrderContext>create()
            .step("createOrder")
                .action(ctx -> orderService.create(ctx))
                .compensation(ctx -> orderService.cancel(ctx))
            .step("deductInventory")
                .action(ctx -> inventoryService.deduct(ctx))
                .compensation(ctx -> inventoryService.rollback(ctx))
            .step("deductBalance")
                .action(ctx -> accountService.deduct(ctx))
                .compensation(ctx -> accountService.refund(ctx))
            .step("sendNotification")
                .action(ctx -> notificationService.send(ctx))
                .compensation(ctx -> {})  // 通知无需补偿
            .build();
    }

    // 执行 Saga
    public void execute(OrderContext ctx) {
        saga.execute(ctx);
        // 如果某步失败，自动反向执行已完成步骤的 compensation
    }
}

// 核心问题：本地事务 + 发送 MQ 消息 不是原子的
// 如果本地事务成功但消息发送失败 → 下游收不到通知
// 如果消息发送成功但本地事务回滚 → 下游收到"幽灵消息"

事务发件箱模式：
// 1. 同一个本地事务中：写业务表 + 写消息表
@Transactional
public void createOrder(Order order) {
    orderRepository.save(order); // 写业务表

    OutboxMessage msg = new OutboxMessage();
    msg.setTopic("order-created");
    msg.setPayload(toJson(order));
    msg.setStatus("PENDING");
    outboxRepository.save(msg); // 写消息表（同一个数据库，同一个事务）
}

// 2. 独立线程/定时任务：扫描消息表，发送到 MQ
@Scheduled(fixedRate = 1000)
public void publishMessages() {
    List<OutboxMessage> pending = outboxRepository
        .findByStatus("PENDING");

    for (OutboxMessage msg : pending) {
        mqProducer.send(msg.getTopic(), msg.getPayload());
        msg.setStatus("SENT");
        outboxRepository.save(msg);
    }
}

// RocketMQ 的事务消息流程

1. 发送半消息（Half Message）
   → 消息对消费者不可见（存在半消息队列）

2. 执行本地事务
   → 成功 → 发送 Commit → 消息对消费者可见
   → 失败 → 发送 Rollback → 删除半消息

3. 如果 RocketMQ 没收到 Commit/Rollback
   → 定时回查（TransactionListener.checkLocalTransaction）
   → 检查本地事务状态，决定 Commit 还是 Rollback

// 本质是把"事务发件箱"的逻辑搬到了 MQ Broker 侧

// AT 模式的核心思想：自动帮你做补偿

// 阶段一：业务 SQL 执行
// Seata 拦截 SQL，自动记录 undo_log（回滚日志）
UPDATE inventory SET stock = stock - 1 WHERE product_id = 'P001';
// Seata 记录：
// before_image: {product_id: P001, stock: 100}
// after_image:  {product_id: P001, stock: 99}

// 阶段二：
// 全部成功 → 异步删除 undo_log（非常轻量）
// 任一失败 → 根据 undo_log 的 before_image 反向执行
// → UPDATE inventory SET stock = 100 WHERE product_id = 'P001'

// AT vs XA 的关键区别：
// XA：Prepare 阶段持有数据库行锁，直到 Commit → 同步阻塞
// AT：Prepare 阶段直接提交本地事务，释放行锁 → 异步补偿
// 所以 AT 性能比 XA 好很多，但只能保证最终一致性

需要强一致性？
├── 是 → 2PC / XA（传统金融系统，能接受低性能）
└── 否（最终一致性可接受）
    ├── 业务能拆分出 Try/Confirm/Cancel？
    │   ├── 是 → TCC（核心业务，如交易、支付）
    │   └── 否 → Saga / AT
    ├── 流程长、步骤多？
    │   ├── 是 → Saga（跨机构、跨系统）
    │   └── 否 → AT（简单业务，不想改代码）
    └── 能接受消息异步方案？
        └── 是 → 消息最终一致性（性价比最高）