1
# 2PC 参与者的阻塞状态（简化示意）
2
class Participant:
3
    def handle_prepare(self, txn_id):
4
        """阶段一：准备"""
5
        if self.can_commit(txn_id):
6
            self.lock_resources(txn_id)     # 锁定资源！
7
            self.write_prepare_log(txn_id)  # 写入 Prepare 日志
8
            return "YES"
9
        return "NO"
10

11
    def handle_commit(self, txn_id):
12
        """阶段二：提交"""
13
        self.commit(txn_id)
14
        self.release_locks(txn_id)  # 只有提交/回滚后才能释放锁
15
        self.write_commit_log(txn_id)
16

17
    #  问题：如果协调者崩溃，参与者将永远停在"已锁定"状态
18
    # 直到协调者恢复并重新发送决定

1
// Java 中使用 XA 事务
2
XADataSource xaDs1 = createXADataSource("jdbc:mysql://db1:3306/order");
3
XADataSource xaDs2 = createXADataSource("jdbc:mysql://db2:3306/inventory");
4
XAResource xaRes1 = xaDs1.getXAConnection().getXAResource();
5
XAResource xaRes2 = xaDs2.getXAConnection().getXAResource();
6

7
TransactionManager tm = getTransactionManager();
8
tm.begin();
9
tm.getTransaction().enlistResource(xaRes1);
10
tm.getTransaction().enlistResource(xaRes2);
11
// 在两个数据源上执行操作... 事务管理器自动执行 2PC
12
tm.commit();  // 内部执行 Prepare → Commit

1
# 3PC 参与者的超时处理
2
class ThreePhaseParticipant:
3
    def handle_can_commit(self, txn_id):
4
        if self.can_commit(txn_id): return "YES"  # 不锁定资源
5
        return "NO"
6

7
    def handle_pre_commit(self, txn_id):
8
        self.lock_resources(txn_id)
9
        self.write_precommit_log(txn_id)
10
        return "ACK"
11

12
    def handle_do_commit(self, txn_id):
13
        self.commit(txn_id)
14
        self.release_locks(txn_id)
15

16
    def on_timeout(self, txn_id):
17
        if self.state == "PRE_COMMITTED":
18
            self.handle_do_commit(txn_id)  # 超时自行提交
19
        elif self.state == "INIT":
20
            self.abort(txn_id)  # 未到 PreCommit，自行回滚

1
// 编排式 Saga 示例
2
public class OrderSagaOrchestrator {
3
    public SagaDefinition<OrderState> saga() {
4
        return step()
5
            .invokeParticipant(this::createOrder)
6
            .withCompensation(this::cancelOrder)
7
        .step()
8
            .invokeParticipant(this::deductInventory)
9
            .withCompensation(this::restoreInventory)
10
        .step()
11
            .invokeParticipant(this::deductBalance)
12
            .withCompensation(this::restoreBalance)
13
        .build();
14
    }
15
}

1
// 协同式 Saga — 订单服务
2
@Transactional
3
public Order createOrder(CreateOrderRequest req) {
4
    Order order = new Order(req);
5
    order.setStatus("PENDING");
6
    orderRepository.save(order);
7
    eventBus.publish(new OrderCreatedEvent(order));  // 触发下游
8
    return order;
9
}
10

11
@Transactional
12
@EventHandler
13
public void handlePaymentFailed(PaymentFailedEvent event) {
14
    Order order = orderRepository.findById(event.getOrderId());
15
    order.setStatus("CANCELLED");  // 补偿：取消订单
16
    orderRepository.save(order);
17
    eventBus.publish(new OrderCancelledEvent(order));
18
}
19

20
// 协同式 Saga — 库存服务
21
@Transactional
22
@EventHandler
23
public void handleOrderCreated(OrderCreatedEvent event) {
24
    inventoryRepository.deduct(event.getProductId(), event.getQuantity());
25
    eventBus.publish(new InventoryDeductedEvent(event));
26
}
27

28
@Transactional
29
@EventHandler
30
public void handleOrderCancelled(OrderCancelledEvent event) {
31
    inventoryRepository.restore(event.getProductId(), event.getQuantity());
32
}

1
原则 1：补偿操作必须是幂等的（网络重试可能导致多次执行）
2
原则 2：补偿操作本身不能失败（否则需人工介入）
3
原则 3：补偿是"语义撤销"而非"物理回滚"（反向转账 vs 删除记录）
4
原则 4：必须逆序补偿（先补偿最后完成的步骤）

1
// TCC 模式实现示例（以库存服务为例）
2
public class InventoryTccService {
3
    // Try 阶段：冻结库存
4
    @Transactional
5
    public boolean tryDeduct(String xid, String productId, int quantity) {
6
        Inventory inv = inventoryRepository.findByProductId(productId);
7
        if (inv.getTotal() - inv.getFrozen() < quantity) return false;
8
        inv.setFrozen(inv.getFrozen() + quantity);  // 冻结，而非扣减
9
        tccBranchRepository.save(new TccBranch(xid, productId, quantity, "TRY"));
10
        return true;
11
    }
12
    // Confirm 阶段：确认扣减
13
    @Transactional
14
    public boolean confirm(String xid) {
15
        TccBranch branch = tccBranchRepository.findByXid(xid);
16
        Inventory inv = inventoryRepository.findByProductId(branch.getProductId());
17
        inv.setTotal(inv.getTotal() - branch.getQuantity());
18
        inv.setFrozen(inv.getFrozen() - branch.getQuantity());
19
        branch.setStatus("CONFIRMED");
20
        return true;
21
    }
22
    // Cancel 阶段：释放冻结
23
    @Transactional
24
    public boolean cancel(String xid) {
25
        TccBranch branch = tccBranchRepository.findByXid(xid);
26
        if (branch.getStatus().equals("CANCELLED")) return true;  // 幂等
27
        Inventory inv = inventoryRepository.findByProductId(branch.getProductId());
28
        inv.setFrozen(inv.getFrozen() - branch.getQuantity());
29
        branch.setStatus("CANCELLED");
30
        return true;
31
    }
32
}

1
-- 本地消息表（Outbox Pattern）实现
2

3
-- 1. 创建消息表
4
CREATE TABLE outbox (
5
    id          BIGINT PRIMARY KEY AUTO_INCREMENT,
6
    topic       VARCHAR(100) NOT NULL,       -- 消息主题
7
    payload     JSON NOT NULL,               -- 消息内容
8
    status      ENUM('NEW', 'SENT', 'FAILED') DEFAULT 'NEW',
9
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
10
    retry_count INT DEFAULT 0,
11
    INDEX idx_status_created (status, created_at)
12
);
13

14
-- 2. 业务操作 + 写入消息表（原子性保证）
15
BEGIN;
16
INSERT INTO orders (user_id, product_id, amount, status)
17
VALUES (1001, 2001, 99.00, 'CREATED');
18

19
INSERT INTO outbox (topic, payload)
20
VALUES ('order-created', '{"orderId": 10001, "userId": 1001, "amount": 99.00}');
21
COMMIT;
22

23
-- 3. 消息发送器（定时轮询）
24
-- 由后台线程定期执行
25
UPDATE outbox SET status = 'SENT'
26
WHERE id = ? AND status = 'NEW';
27

28
-- 4. 失败重试
29
UPDATE outbox
30
SET retry_count = retry_count + 1,
31
    status = CASE WHEN retry_count >= 3 THEN 'FAILED' ELSE 'NEW' END
32
WHERE id = ? AND status = 'FAILED';

1
// RocketMQ 事务消息示例
2
public class OrderService {
3
    @Autowired private RocketMQTemplate rocketMQTemplate;
4

5
    public void createOrder(CreateOrderRequest req) {
6
        // 发送半消息：对消费者不可见，等待本地事务确认
7
        rocketMQTemplate.sendMessageInTransaction(
8
            "order-topic", MessageBuilder.withPayload(req).build(), req);
9
    }
10

11
    @Transactional
12
    public void executeLocalTransaction(Message msg) {
13
        orderRepository.save(new Order(parseMessage(msg)));
14
        // 本地事务提交成功 → 消息对消费者可见
15
    }
16

17
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
18
        // 事务回查：不确定本地事务状态时调用
19
        Order order = orderRepository.findById(parseOrderId(msg));
20
        return order != null ? COMMIT : ROLLBACK;
21
    }
22
}

1
// 幂等消费者：唯一消息 ID 去重
2
@Transactional
3
public void handleOrderCreated(OrderCreatedEvent event) {
4
    String msgId = event.getMessageId();
5
    if (idempotentRepo.existsByMessageId(msgId)) return;  // 重复消息跳过
6
    inventoryService.deduct(event.getProductId(), event.getQuantity());
7
    idempotentRepo.save(new IdempotentRecord(msgId, "PROCESSED"));
8
}

1
1. 优先避免分布式事务 — 合理划分服务，将强关联数据放同一服务
2
2. 选择合适的方案 — 不要为"强一致性"强行用 2PC，多数场景可接受最终一致性
3
3. 做好补偿和重试 — 所有操作必须幂等，补偿操作必须有重试机制
4
4. 可观测性 — 记录事务步骤状态，实现事务看板排查悬挂事务
5
5. 测试故障场景 — 模拟网络分区、节点崩溃，验证补偿与幂等正确性