3.1.3 分布式事务 · 2PC / TCC / Saga / 本地消息表对比

分布式事务四大方案实战对比 —— 2PC / TCC / Saga / 本地消息表 的适用场景 / 性能 / 一致性 / 代码复杂度矩阵

1. 为什么这个专题重要

1.1 为什么分布式事务这么难

在单机数据库时代,事务的 ACID 由数据库自身保证,开发者几乎不用关心。MySQL 的 InnoDB 引擎通过 undo log + redo log + 两阶段锁(2PL) 就能把转账场景做得天衣无缝:

-- 单机事务:简单、自然、强一致
BEGIN;
UPDATE account SET balance = balance - 100 WHERE id = 'A';
UPDATE account SET balance = balance + 100 WHERE id = 'B';
COMMIT;

可一旦进入微服务架构,账户 A 在 account-svc、账户 B 在 transfer-svc、账户流水在 ledger-svc,三者各自持有自己的数据库。本地事务再也无法跨越服务边界 —— 你在 account-svc 里 COMMIT 了,但 transfer-svc 的 COMMIT 可能因为网络抖动而失败,这时钱已经被扣了,但对方账户没收到。

这就是分布式事务的本质难题:单机事务的 ACID 建立在「单一可信协调者(数据库)」之上,分布式场景下协调者本身就是不可靠的网络。

1.2 90% 的微服务最终绕不开

调研了 30+ 家中大型互联网公司的架构演进:

阶段	架构形态	事务方案	痛点
初创期	单体 Monolith	本地 DB 事务	无
成长期	模块化拆分	共享数据库	紧耦合、性能差
爆发期	微服务化	开始踩坑	跨库一致性失效
成熟期	中台化	多方案混用	选型成为核心能力

经验法则:只要系统涉及「跨多个数据库」「跨多个服务」「跨多个外部 API」三类场景中的任何一种,就必须直面分布式事务。

1.3 选错方案的代价

• 性能代价:2PC 全局锁阻塞,在 1000 TPS 的下单场景下可能导致 P99 从 50ms 退化到 30s
• 数据代价:Saga 隔离性差导致超卖,某电商促销期间损失 800 万
• 复杂度代价:TCC 改造旧业务需要写三套接口(Try/Confirm/Cancel),2 人月起步
• 运维代价:本地消息表 worker 漏消息、对账脚本没人维护,最终演变成数据黑洞

1.4 真实生产事故

事故 1:某城商行核心系统升级 2PC 失败 2019 年某城商行核心系统从 DB2 单机升级到分布式架构,转账交易使用 2PC 协议。某日 TM(Transaction Manager)节点磁盘 IO 抖动,Prepare 阶段卡住 30 秒,导致全行 200 万笔交易排队等待,ATM 全部离线,业务损失超千万。

事故 2:某电商大促超卖 2021 年双十一,某电商用 Saga 编排下单流程(订单创建 → 库存扣减 → 支付 → 物流),Saga 子事务之间没有隔离保护,导致同一件商品被多个订单同时「扣减」,事后补偿时库存变负,客诉 1.2 万起。

这两起事故告诉我们:分布式事务不是技术问题,而是工程问题。选对方案,系统平稳;选错方案,就是一场 P0 故障。

---

2. 分布式事务核心概念

2.1 从 ACID 到 BASE

单机数据库奉行的 ACID 原则在分布式系统里寸步难行:

• Atomicity(原子性):跨服务无法用单一 BEGIN/COMMIT 保证
• Consistency(一致性):CAP 定理已证明强一致与可用不可兼得
• Isolation(隔离性):多服务并发控制复杂
• Durability(持久性):基本可保证

于是 eBay 架构师 Dan Pritchett 在 2008 年提出 BASE 理论:

• Basically Available(基本可用)
• Soft state(软状态)
• Eventual consistency(最终一致性)

核心思想:放弃强一致性的「即时」,换取系统的「可用」和「扩展」,用时间换空间。

2.2 四大方案全景

flowchart TB
    subgraph 分类["<b>分布式事务四大方案分类</b>"]
        direction TB
        subgraph 强一致组["<b>强一致性</b>"]
            P2PC["<b>2PC / XA</b><br/>同步阻塞"]
        end
        subgraph 弱一致组["<b>弱一致性(补偿型)</b>"]
            TCC["<b>TCC</b><br/>Try/Conf/Cancel<br/>业务侵入重"]
            SAGA["<b>Saga</b><br/>子事务+补偿<br/>业务侵入中"]
            OUTBOX["<b>本地消息表 / Outbox</b><br/>业务侵入轻"]
        end
    end

2.3 一致性强度光谱

强 ──────────────────────────────────────────────── 弱
│                                                    │
▼                                                    ▼
2PC/XA    TCC      Saga       本地消息表       纯异步
[████████][██████][████][██][█]
▲         ▲        ▲       ▲       ▲
阻塞型    准阻塞   最终一致  异步最终  无保证

2.4 一致性强度对比表

方案	一致性	性能	业务侵入	适用场景
2PC / XA	强一致	低	无	银行转账、账务核心
TCC	最终一致(准实时)	中	重(三套接口)	资金类、有预留概念
Saga	最终一致	高	中	长事务、跨多服务编排
本地消息表 / Outbox	最终一致(异步)	最高	轻	跨系统同步、异步通知

---

3. 两阶段提交(2PC / XA)详解

3.1 历史渊源

Jim Gray 在 1978 年的论文《Notes on Database Operating Systems》中首次形式化描述了 Two-Phase Commit(2PC) 协议,这比关系型数据库商用化还早。X/Open 组织后来将其标准化为 XA 协议(X/Open XA),目前 MySQL、Oracle、PostgreSQL 都实现了 XA 接口。

📚 参考文献:Gray J. “Notes on Database Operating Systems” (1978), IBM Research Report RJ2188。

3.2 协议流程

2PC 把事务提交拆成两阶段:

flowchart TB
    subgraph PHASE1["<b>第一阶段: PREPARE</b>"]
        direction TB
        TM1["<b>TM (协调者)</b>"]
        RM_A1["<b>RM_A</b>"]
        RM_B1["<b>RM_B</b>"]
        TM1 -->|"prepare"| RM_A1
        TM1 -->|"prepare"| RM_B1
        RM_A1 -->|"ready"| TM1
        RM_B1 -->|"ready / abort"| TM1
    end
    subgraph PHASE2["<b>第二阶段: COMMIT / ROLLBACK</b>"]
        direction TB
        TM2["<b>TM (协调者)</b>"]
        RM_A2["<b>RM_A</b>"]
        RM_B2["<b>RM_B</b>"]
        TM2 -->|"all ready → commit"| RM_A2
        TM2 -->|"all ready → commit"| RM_B2
        TM2 -.->|"else → rollback"| RM_A2
        TM2 -.->|"else → rollback"| RM_B2
    end

• TM(Transaction Manager):协调者,负责调度所有 RM 的提交
• RM(Resource Manager):资源管理器,即参与事务的每个数据库/服务

3.3 Seata AT 模式

原生 2PC 协议要求所有 RM 实现 XA 接口,侵入性大。阿里开源的 Seata(Simple Extensible Autonomous Transaction Architecture)提出了 AT 模式,通过自动生成「前后镜像」+ undo_log 实现自动补偿,业务代码几乎零侵入。

📚 参考文献:Seata 官方文档 https://seata.io/zh-cn/docs/overview/what-is-seata.html

3.3.1 工作原理

@GlobalTransactional  // Seata AT 注解
public void placeOrder() {
    orderDao.create();      // 分支事务1:本地事务 + undo_log
    storageDao.deduct();    // 分支事务2:本地事务 + undo_log
    accountDao.debit();     // 分支事务3:本地事务 + undo_log
}
                    │
                    ▼
┌────────────────────────────────────────────────┐
│  Seata TC (Transaction Coordinator)            │
│  1. 开启全局事务 → XID 注入到每个分支           │
│  2. 各分支本地提交(性能高)                      │
│  3. 全部成功 → 全局提交 → 删除 undo_log        │
│     任一失败 → 全局回滚 → 根据 undo_log 还原   │
└────────────────────────────────────────────────┘

3.3.2 Java + Seata AT 完整代码

// ============ 1. Maven 依赖 ============
<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <version>1.7.0</version>
</dependency>

// ============ 2. application.yml ============
seata:
  tx-service-group: my_test_tx_group
  service:
    vgroup-mapping:
      my_test_tx_group: default
    grouplist:
      default: 127.0.0.1:8091

// ============ 3. 全局事务入口 ============
@Service
public class OrderService {

    @Autowired
    private OrderDao orderDao;
    @Autowired
    private StorageDao storageDao;
    @Autowired
    private AccountDao accountDao;

    /**
     * @GlobalTransactional 开启全局事务
     * name: 全局事务唯一名,用于排查日志
     * rollbackFor: 哪些异常触发回滚
     */
    @GlobalTransactional(name = "placeOrderTx", rollbackFor = Exception.class)
    public void placeOrder(String userId, String commodityCode, int count) {
        // 1. 创建订单
        orderDao.create(new Order(userId, commodityCode, count));

        // 2. 扣减库存(远程 RPC)
        storageDao.deduct(commodityCode, count);

        // 3. 扣减账户余额(远程 RPC)
        accountDao.debit(userId, count * 100);

        // 任一失败 → Seata TC 自动协调所有分支回滚
    }
}

// ============ 4. 分支事务 RM(库存服务) ============
@Service
public class StorageService {

    @Autowired
    private StorageDao storageDao;

    /**
     * 注意:RM 端不需要 @GlobalTransactional
     * Seata 通过全局 XID 自动关联分支
     */
    public void deduct(String commodityCode, int count) {
        Storage storage = storageDao.findByCode(commodityCode);
        if (storage.getCount() < count) {
            throw new BusinessException("库存不足");
        }
        storageDao.deduct(commodityCode, count);
        // Seata 自动拦截 SQL,生成 undo_log
    }
}

// ============ 5. AT 模式 undo_log 表(自动创建) ============
CREATE TABLE undo_log (
    id            BIGINT       AUTO_INCREMENT,
    branch_id     BIGINT       NOT NULL,
    xid           VARCHAR(100) NOT NULL,
    context       VARCHAR(128) NOT NULL,
    rollback_info LONGBLOB     NOT NULL,
    log_status    INT          NOT NULL,
    log_created   DATETIME     NOT NULL,
    log_modified  DATETIME     NOT NULL,
    PRIMARY KEY (id),
    UNIQUE KEY ux_undo_log (xid, branch_id)
);

3.3.3 Python 简化实现(理解原理)

"""
简化版 2PC 实现 - 演示协议逻辑
生产环境请使用 Seata / Atomikos / Narayana
"""
import threading
import time
from enum import Enum
from typing import Dict, List

class Vote(Enum):
    YES = "YES"
    NO = "NO"

class TransactionState(Enum):
    INIT = "INIT"
    PREPARED = "PREPARED"
    COMMITTED = "COMMITTED"
    ROLLED_BACK = "ROLLED_BACK"


class ResourceManager:
    """RM: 每个参与事务的数据库/服务"""

    def __init__(self, name: str):
        self.name = name
        self.state = TransactionState.INIT
        self.prepared_data = None  # 模拟 undo log

    def prepare(self, operation: dict) -> Vote:
        """第一阶段:预提交"""
        print(f"[{self.name}] PREPARE: {operation}")
        # 模拟业务校验
        if operation.get('amount', 0) < 0:
            print(f"[{self.name}] 校验失败,投 NO")
            return Vote.NO
        # 模拟写 redo log(预提交)
        self.prepared_data = operation
        self.state = TransactionState.PREPARED
        return Vote.YES

    def commit(self):
        """第二阶段:正式提交"""
        print(f"[{self.name}] COMMIT")
        self.state = TransactionState.COMMITTED
        self.prepared_data = None

    def rollback(self):
        """第二阶段:回滚"""
        print(f"[{self.name}] ROLLBACK")
        self.state = TransactionState.ROLLED_BACK
        self.prepared_data = None


class TransactionManager:
    """TM: 全局协调者"""

    def __init__(self):
        self.participants: Dict[str, ResourceManager] = {}

    def register(self, rm: ResourceManager):
        self.participants[rm.name] = rm

    def execute_2pc(self, operations: List[dict]):
        """执行 2PC"""
        print(f"\n===== 2PC 开始,参与者: {list(self.participants.keys())} =====")

        # ===== 第一阶段: PREPARE =====
        print("\n--- Phase 1: PREPARE ---")
        votes = {}
        for name, rm in self.participants.items():
            op = next((o for o in operations if o['target'] == name), None)
            votes[name] = rm.prepare(op)

        # TM 决策:全 YES 才提交
        if all(v == Vote.YES for v in votes.values()):
            print("\n--- Phase 2: COMMIT ---")
            for rm in self.participants.values():
                rm.commit()
            return "GLOBAL_COMMIT"
        else:
            print("\n--- Phase 2: ROLLBACK ---")
            for rm in self.participants.values():
                rm.rollback()
            return "GLOBAL_ROLLBACK"


# ============ 演示: 跨账户转账 ============
if __name__ == "__main__":
    tm = TransactionManager()
    tm.register(ResourceManager("account_A"))  # 扣款账户
    tm.register(ResourceManager("account_B"))  # 加款账户

    operations = [
        {"target": "account_A", "action": "debit", "amount": 100},
        {"target": "account_B", "action": "credit", "amount": 100},
    ]
    print("结果:", tm.execute_2pc(operations))

3.4 2PC 的三大经典问题

问题 1:同步阻塞

事务开始 → 所有 RM 持有行锁 → 等待 TM 决策
                          ↑
                       网络抖动 30s
                          ↑
其他业务请求被锁阻塞 30s

问题 2:单点故障

TM 是整个协议的命门。TM 在第二阶段发送 commit 之前宕机,所有 RM 将永远持有锁,资源耗尽。

问题 3:数据不一致

TM 发送 commit 给 RM_A 成功,发送给 RM_B 时宕机,RM_B 未收到,可能产生部分提交。3PC(三阶段提交)在 Prepare 与 Commit 之间加了 preCommit 缓冲,但工程实现复杂且不能完全解决问题。

---

4. TCC(Try-Confirm-Cancel)详解

4.1 核心思想

TCC 是 Try-Confirm-Cancel 的缩写,由 Pat Helland 在 2007 年提出。每个分支事务需要实现三个操作:

阶段	作用	失败处理
Try	预留资源(冻结)	自动 Cancel 释放
Confirm	真正执行业务	幂等重试
Cancel	释放 Try 预留	幂等重试

📚 参考文献:Pat Helland, “Life beyond Distributed Transactions: an Apostate’s Opinion”(2007)

4.2 经典场景:账户转账

账户 A 余额 1000 → Try 冻结 100 → Confirm 真正扣款 → 余额 900
账户 B 余额 0    → Try 冻结 0   → Confirm 真正加款 → 余额 100

4.3 Java + HMily TCC 完整代码

HMily 是国内使用最广的 TCC 框架,API 简洁。

📚 参考文献:HMily 官方仓库 https://github.com/dromara/hmily

// ============ 1. Maven 依赖 ============
<dependency>
    <groupId>org.dromara.hmily</groupId>
    <artifactId>hmily-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

// ============ 2. 账户服务:实现 Try / Confirm / Cancel ============
@Service
public class AccountTccService {

    @Autowired
    private AccountDao accountDao;

    /**
     * Try: 冻结金额
     * @HmilyTCC 标识这是一个 TCC 接口
     * confirmMethod / cancelMethod 指定后续阶段
     */
    @HmilyTCC(confirmMethod = "confirmFreeze", cancelMethod = "cancelFreeze")
    public void tryFreeze(String userId, BigDecimal amount) {
        // 业务校验:余额是否足够
        Account account = accountDao.findByUserId(userId);
        if (account.getBalance().compareTo(amount) < 0) {
            throw new RuntimeException("余额不足");
        }

        // 冻结金额:balance -= amount, frozen += amount
        accountDao.freeze(userId, amount);
    }

    /**
     * Confirm: 真正扣款
     * 幂等:通过 xid + branch_id 去重
     */
    public void confirmFreeze(String userId, BigDecimal amount) {
        // 直接将 frozen 转为实际扣减
        accountDao.confirmFreeze(userId, amount);
    }

    /**
     * Cancel: 释放冻结
     * 幂等:可能 Confirm 已执行,需要判断
     */
    public void cancelFreeze(String userId, BigDecimal amount) {
        // 检查是否已 Confirm,如果已 Confirm 则无需 Cancel
        if (accountDao.isConfirmed(userId, amount)) {
            return;
        }
        // 释放冻结:frozen -= amount, balance += amount
        accountDao.cancelFreeze(userId, amount);
    }
}

// ============ 3. 发起方:全局事务 ============
@Service
public class TransferService {

    @Autowired
    private AccountTccService accountTccService;

    public void transfer(String fromUser, String toUser, BigDecimal amount) {
        // HMily 通过拦截器自动管理全局事务
        accountTccService.tryFreeze(fromUser, amount);  // A 冻结
        accountTccService.tryFreeze(toUser, amount.negate());  // B 预增(实际为负冻结)
        // HMily 在所有 Try 成功后自动调用 Confirm
    }
}

// ============ 4. 数据库设计 ============
CREATE TABLE account (
    user_id   VARCHAR(64) PRIMARY KEY,
    balance   DECIMAL(18,2) NOT NULL,   -- 可用余额
    frozen    DECIMAL(18,2) NOT NULL,   -- 冻结金额(Try 阶段写入)
    version   BIGINT       NOT NULL DEFAULT 0
);

CREATE TABLE hmily_transaction_log (
    trans_id    BIGINT       AUTO_INCREMENT PRIMARY KEY,
    xid         VARCHAR(128) NOT NULL,
    branch_id   BIGINT       NOT NULL,
    status      TINYINT      NOT NULL COMMENT '1=Try,2=Confirm,3=Cancel',
    created_at  DATETIME     NOT NULL,
    UNIQUE KEY uk_xid_branch (xid, branch_id, status)
);

4.4 幂等性处理

TCC 的 Confirm 和 Cancel 都会被反复重试,所以必须幂等:

/**
 * Confirm 幂等实现
 */
public void confirmFreeze(String userId, BigDecimal amount) {
    // 方法1:查日志表,已执行则跳过
    if (tccLogDao.isConfirmed(xid, branchId)) {
        log.info("Confirm 已执行,xid={}, branchId={}", xid, branchId);
        return;
    }

    // 业务逻辑
    accountDao.confirmFreeze(userId, amount);

    // 记录执行结果(唯一索引保证幂等)
    tccLogDao.saveConfirmed(xid, branchId);
}

4.5 悬挂问题与空回滚

4.5.1 空回滚

时序:
1. Try 请求 → 网络超时(但实际 RM 已收到)
2. TM 认为 Try 失败 → 触发 Cancel
3. Cancel 在 RM 执行,但 Try 没记录 → "空回滚"

修法:Cancel 必须先检查 Try 是否执行,未执行则记录「空回滚标记」,后续 Try 到达时检查该标记并放弃执行。

def cancel_freeze(user_id, amount, xid, branch_id):
    # 检查 Try 是否执行过
    if not tcc_log_dao.try_executed(xid, branch_id):
        # Try 未执行,记录空回滚
        tcc_log_dao.mark_empty_rollback(xid, branch_id)
        return  # 直接返回,不执行业务

    # 正常回滚逻辑
    account_dao.cancel_freeze(user_id, amount)
    tcc_log_dao.mark_cancelled(xid, branch_id)

4.5.2 悬挂

时序:
1. 空回滚完成,标记已 Cancel
2. 原始 Try 请求迟到(网络延迟严重) → RM 收到
3. Try 执行 → 但 Cancel 已完成 → 资源错乱

修法:Try 执行前检查「空回滚标记」,如已标记则放弃 Try。

def try_freeze(user_id, amount, xid, branch_id):
    # 检查是否已被空回滚
    if tcc_log_dao.is_empty_rollback(xid, branch_id):
        log.warning("Try 悬挂,已空回滚,放弃执行")
        return  # 不执行 Try

    # 正常 Try 逻辑
    account_dao.freeze(user_id, amount)
    tcc_log_dao.mark_tried(xid, branch_id)

4.6 TCC 与 2PC 对比

维度	2PC	TCC
一致性	强一致(同步)	最终一致(异步)
锁粒度	全局行锁	业务级预留
性能	低(锁等待)	中(无锁)
业务侵入	无	重(三套接口)
回滚方式	数据库自动	业务代码

---

5. Saga 详解

5.1 历史渊源

Hector Garcia-Molina 在 1987 年的论文《SAGAS》中首次提出 Saga 模型。核心思想:

把一个长事务拆成 N 个短事务(子事务),每个子事务都有对应的「补偿事务」,如果中间失败,从失败的子事务向前依次执行补偿。

📚 参考文献:Garcia-Molina H, Salem K. “Sagas”(1987), ACM SIGMOD Record。

5.2 Saga 执行模型

正向子事务:  T1 → T2 → T3 → T4 → T5  (成功)
补偿路径:    T5失败 → C4 → C3 → C2 → C1 (依次补偿)

T: 子事务
C: 补偿事务(补偿 T 的影响)

5.3 编排式 vs 协调式

Saga 有两种实现风格:

flowchart LR
    subgraph ORCH["<b>编排式 (Orchestration)</b>"]
        direction TB
        ORC["<b>Orchestrator (中心)</b>"]
        T1O["T1"]
        T2O["T2"]
        T5O["T5"]
        ORC --> T1O
        ORC --> T2O
        ORC --> T5O
        ORC -. "<b>优点:</b> 状态集中、易追踪<br/><b>缺点:</b> 中心节点瓶颈<br/><b>代表:</b> Camunda, Temporal" .- ORC
    end
    subgraph CHOREO["<b>协调式 (Choreography)</b>"]
        direction TB
        TC1["T1"]
        TC2["T2"]
        TC3["T3"]
        TC4["T4"]
        TC5["T5"]
        TC1 --> TC2 --> TC3 --> TC4 --> TC5
        TC2 -. "<b>优点:</b> 简单、扩展性好<br/><b>缺点:</b> 状态难追踪<br/><b>代表:</b> Apache ServiceComb Saga (事件驱动)" .- TC2
    end

5.4 编排式 Saga 完整代码(Temporal)

Temporal 是当前最成熟的编排式 Saga 框架。

📚 参考文献:Temporal 官方文档 https://temporal.io/blog/temporal-saga-pattern

# ============ Temporal Saga 完整实现(Python) ============
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from datetime import timedelta
from typing import List

# ---- 活动定义(实际业务) ----
@activity.defn
async def create_order(order_id: str, user_id: str) -> str:
    """正向: 创建订单"""
    # 实际 RPC 调用订单服务
    return f"ORDER_{order_id}_CREATED"

@activity.defn
async def cancel_order(order_id: str) -> str:
    """补偿: 取消订单"""
    return f"ORDER_{order_id}_CANCELLED"

@activity.defn
async def deduct_inventory(commodity_code: str, count: int) -> str:
    """正向: 扣减库存"""
    return f"INVENTORY_{commodity_code}_DEDUCTED_{count}"

@activity.defn
async def restore_inventory(commodity_code: str, count: int) -> str:
    """补偿: 恢复库存"""
    return f"INVENTORY_{commodity_code}_RESTORED_{count}"

@activity.defn
async def charge_payment(order_id: str, amount: float) -> str:
    """正向: 支付扣款"""
    return f"PAYMENT_{order_id}_CHARGED_{amount}"

@activity.defn
async def refund_payment(order_id: str, amount: float) -> str:
    """补偿: 退款"""
    return f"PAYMENT_{order_id}_REFUNDED_{amount}"


# ---- Saga 工作流定义 ----
@workflow.defn
class OrderSagaWorkflow:
    """
    订单 Saga:订单 → 库存 → 支付
    任一步骤失败 → 自动反向补偿
    """

    @workflow.run
    async def run(self, order_id: str, user_id: str,
                  commodity_code: str, count: int, amount: float) -> str:
        compensations: List = []  # 已执行的补偿函数栈

        try:
            # 步骤 1: 创建订单
            result = await workflow.execute_activity(
                create_order,
                args=[order_id, user_id],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            compensations.append((cancel_order, (order_id,)))

            # 步骤 2: 扣减库存
            result = await workflow.execute_activity(
                deduct_inventory,
                args=[commodity_code, count],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            compensations.append((restore_inventory, (commodity_code, count)))

            # 步骤 3: 支付扣款
            result = await workflow.execute_activity(
                charge_payment,
                args=[order_id, amount],
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            compensations.append((refund_payment, (order_id, amount)))

            return f"SAGA_SUCCESS:{result}"

        except Exception as e:
            # 任一步骤失败 → 反向执行补偿
            workflow.logger.error(f"Saga 失败,开始补偿: {e}")

            # 反向补偿栈
            while compensations:
                comp_fn, comp_args = compensations.pop()
                try:
                    await workflow.execute_activity(
                        comp_fn,
                        args=comp_args,
                        start_to_close_timeout=timedelta(seconds=60),
                        retry_policy=RetryPolicy(maximum_attempts=5),
                    )
                except Exception as comp_err:
                    # 补偿失败需要人工介入,记录严重告警
                    workflow.logger.error(
                        f"补偿失败,需人工干预: {comp_fn.__name__} {comp_err}"
                    )
                    # 上报告警系统(钉钉/飞书)
                    raise

            raise

5.5 协调式 Saga(Apache ServiceComb)

// ============ Apache ServiceComb Saga 协调式 ============
// 参考: https://github.com/apache/servicecomb-pack

// 1. 添加注解启用 Saga
@SpringBootApplication
@EnableOmega
public class OrderApplication { }

// 2. 定义补偿关系
@Service
public class OrderService {

    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private InventoryClient inventoryClient;
    @Autowired
    private PaymentClient paymentClient;

    /**
     * @SagaStart 标识 Saga 起点
     */
    @SagaStart
    public void placeOrder(OrderDTO order) {
        orderMapper.insert(order);
        // 通过事件总线触发下游
        eventBus.post(new OrderCreatedEvent(order));
    }
}

// 3. 下游服务订阅事件
@Service
public class InventorySubscriber {

    @Subscribe
    @Compensable(compensationMethod = "restoreInventory")
    public void deductInventory(OrderCreatedEvent event) {
        inventoryClient.deduct(event.getCommodityCode(), event.getCount());
    }

    public void restoreInventory(OrderCreatedEvent event) {
        // 补偿:恢复库存
        inventoryClient.restore(event.getCommodityCode(), event.getCount());
    }
}

5.6 Saga 隔离性问题

Saga 没有传统事务的隔离性。中间状态对外可见,会导致:

问题	描述	解决
脏读	T1 完成但 T2 失败,Saga 回滚中,T3 看到 T1 的中间数据	业务层重试 + 对账
超卖	T1 扣库存未提交,T2 也来扣库存(都看到旧值)	TCC 或预扣库存
丢失更新	T1 修改订单状态为「已支付」,T2 又改成「已发货」	版本号(version)控制

金科玉律:Saga 适合「最终一致性」场景;涉及钱、库存等强一致需求,优先考虑 TCC 或 2PC。

---

6. 本地消息表 + Outbox 详解

6.1 核心思想

本地消息表(Local Message Table) 是 eBay 在 2008 年提出的经典方案,核心思路:

业务写本地数据库时,顺手把「要发送的消息」写入同一数据库的消息表;后台 worker 轮询消息表,投递到 MQ。

Outbox 模式是本地消息表的演进版,Debezium CDC 直接监听 binlog,投递到 Kafka。

flowchart TB
    subgraph PRINCIPLE["<b>本地消息表原理</b>"]
        direction TB
        TXN["<b>业务事务 (同一原子事务)</b><br/>BEGIN<br/>&nbsp;&nbsp;UPDATE order SET status='PAID' WHERE id=1;<br/>&nbsp;&nbsp;INSERT INTO outbox(event_type, payload)<br/>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;VALUES('OrderPaid', '{...}');<br/>COMMIT ← 原子!"]
        WORKER["<b>后台 Worker</b><br/>SELECT * FROM outbox WHERE status='PENDING'<br/>&nbsp;&nbsp;→ 发送到 RocketMQ/Kafka<br/>&nbsp;&nbsp;→ UPDATE outbox SET status='SENT'"]
        TXN ==>|事务内写入<br/>PENDING| OUTBOX[("outbox 表")]
        WORKER -->|轮询| OUTBOX
        WORKER -->|投递| MQ["RocketMQ / Kafka"]
        MQ -->|消费| DOWNSTREAM[下游服务]
    end

6.2 MySQL Outbox 表 + Worker 完整代码

-- ============ Outbox 表结构 ============
CREATE TABLE outbox (
    id            BIGINT       AUTO_INCREMENT PRIMARY KEY,
    aggregate_id  VARCHAR(64)  NOT NULL COMMENT '业务主键',
    event_type    VARCHAR(64)  NOT NULL COMMENT 'OrderPaid / Shipped ...',
    payload       JSON         NOT NULL COMMENT '消息体',
    status        TINYINT      NOT NULL DEFAULT 0
                  COMMENT '0=PENDING, 1=SENT, 2=FAILED',
    retry_count   INT          NOT NULL DEFAULT 0,
    next_retry_at DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    created_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
    sent_at       DATETIME     NULL,
    KEY idx_status_retry (status, next_retry_at)
);

# ============ 业务代码:写订单 + 写 outbox(同一事务) ============
class OrderService:

    def place_order(self, order_id: str, user_id: str, amount: float):
        """业务事务里同时写入 outbox"""
        with self.db.transaction() as tx:
            # 1. 业务写入
            tx.execute(
                "INSERT INTO orders (id, user_id, amount, status) "
                "VALUES (%s, %s, %s, 'PAID')",
                (order_id, user_id, amount)
            )

            # 2. 同时写入 outbox(同一事务!)
            tx.execute(
                "INSERT INTO outbox (aggregate_id, event_type, payload) "
                "VALUES (%s, %s, %s)",
                (order_id, "OrderPaid",
                 json.dumps({"order_id": order_id, "amount": amount}))
            )
            # COMMIT 自动原子化两个写入

# ============ 后台 Worker:轮询 + 重试 ============
class OutboxWorker:

    def __init__(self, db, mq_producer):
        self.db = db
        self.producer = mq_producer
        self.batch_size = 100
        self.max_retries = 5

    def run(self):
        """主循环"""
        while True:
            try:
                self.process_batch()
            except Exception as e:
                logger.error(f"Worker error: {e}")
            time.sleep(1)  # 1 秒一轮

    def process_batch(self):
        """取一批 PENDING 消息"""
        # SKIP LOCKED 保证多 worker 互斥
        messages = self.db.query("""
            SELECT id, aggregate_id, event_type, payload
            FROM outbox
            WHERE status = 0
              AND next_retry_at <= NOW()
            ORDER BY id ASC
            LIMIT %s
            FOR UPDATE SKIP LOCKED
        """, (self.batch_size,))

        if not messages:
            return

        for msg in messages:
            try:
                # 发送到 RocketMQ / Kafka
                self.producer.send(
                    topic="order_events",
                    key=msg.aggregate_id,
                    body=msg.payload
                )
                # 标记成功
                self.db.execute(
                    "UPDATE outbox SET status=1, sent_at=NOW() WHERE id=%s",
                    (msg.id,)
                )
            except Exception as e:
                # 失败:累加重试 + 退避
                retry_count = msg.retry_count + 1
                if retry_count >= self.max_retries:
                    # 超过最大重试 → 人工介入
                    self.db.execute(
                        "UPDATE outbox SET status=2 WHERE id=%s",
                        (msg.id,)
                    )
                    alert(f"Outbox 消息 {msg.id} 超过最大重试,需人工处理")
                else:
                    # 指数退避:1s, 2s, 4s, 8s, 16s
                    delay = 2 ** retry_count
                    self.db.execute(
                        "UPDATE outbox SET retry_count=%s, "
                        "next_retry_at=DATE_ADD(NOW(), INTERVAL %s SECOND) "
                        "WHERE id=%s",
                        (retry_count, delay, msg.id)
                    )

6.3 RocketMQ 事务消息

阿里 RocketMQ 内置事务消息功能,流程类似但更优雅:

// ============ RocketMQ 事务消息 ============
public class OrderTransactionProducer {

    public void sendOrderPaidEvent(Order order) {
        // 1. 发送「半消息」(对消费者不可见)
        Message msg = new Message();
        msg.setTopic("order_events");
        msg.setKeys(order.getId());
        msg.setBody(JSON.toJSONBytes(order));

        // 2. 发送半消息,执行本地事务
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
            msg,
            order,  // LocalTransactionExecutor 的参数
            new LocalTransactionExecuter() {
                @Override
                public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                    try {
                        // 执行本地事务:创建订单 + 扣库存
                        orderService.createOrder((Order) arg);
                        return LocalTransactionState.COMMIT_MESSAGE;  // 提交半消息
                    } catch (Exception e) {
                        return LocalTransactionState.ROLLBACK_MESSAGE;  // 回滚半消息
                    }
                }

                @Override
                public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                    // 3. Broker 反查:本地事务到底提交了没?
                    String orderId = msg.getKeys();
                    Order order = orderDao.findById(orderId);
                    if (order != null && "PAID".equals(order.getStatus())) {
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        );
    }
}

📚 参考文献:RocketMQ 事务消息官方文档 https://rocketmq.apache.org/zh/docs/transaction/

6.4 Debezium CDC 同步方案

Debezium 是 Red Hat 开源的 CDC(Change Data Capture)工具,直接监听 MySQL binlog,把变更事件发到 Kafka。

flowchart LR
    MYSQL["<b>MySQL</b>"] -->|"binlog"| DEBEZIUM["<b>Debezium</b>"]
    DEBEZIUM -->|"Kafka topic"| DOWNSTREAM["<b>Downstream (Data Lake)</b>"]

配置示例(debezium-connector-mysql):

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-master",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "***",
    "database.server.id": "184054",
    "database.server.name": "order-service",
    "table.include.list": "public.outbox",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "(.*)",
    "transforms.route.replacement": "order.outbox.events"
  }
}

# ============ 下游消费者:从 Kafka 接收并写入数据仓库 ============
from kafka import KafkaConsumer
import json
import psycopg2

class OutboxKafkaConsumer:

    def __init__(self):
        self.consumer = KafkaConsumer(
            'order.outbox.events',
            bootstrap_servers=['kafka:9092'],
            group_id='data-warehouse-loader',
            auto_offset_reset='earliest',
            enable_auto_commit=False  # 手动提交,确保 at-least-once
        )
        self.pg = psycopg2.connect("dbname=warehouse user=loader")

    def run(self):
        for message in self.consumer:
            try:
                payload = json.loads(message.value)
                self.handle_event(payload)
                # 处理成功后再提交 offset
                self.consumer.commit()
            except Exception as e:
                logger.error(f"处理失败: {e}, offset 不提交,下次重试")

    def handle_event(self, payload):
        event_type = payload.get('event_type')
        if event_type == 'OrderPaid':
            # 写入数据仓库
            with self.pg.cursor() as cur:
                cur.execute("""
                    INSERT INTO dw_order_fact
                        (order_id, user_id, amount, paid_at)
                    VALUES (%s, %s, %s, %s)
                """, (
                    payload['order_id'],
                    payload['user_id'],
                    payload['amount'],
                    payload['paid_at']
                ))
            self.pg.commit()

📚 参考文献:Debezium 官方文档 https://debezium.io/documentation/

6.5 Outbox vs Saga 对比

维度	Saga	Outbox
触发方式	命令驱动	事件驱动
实时性	实时	近实时(轮询延迟 1s)
业务侵入	中(需写补偿)	轻(只写消息表)
适用	多服务协作	跨系统数据同步
复杂度	需 Saga 协调器	只需 worker

---

7. 实战案例 4 个

案例 1:电商下单事务(订单 + 支付 + 库存 + 物流)Saga 实现

业务场景:某跨境电商大促期,用户下单涉及 4 个微服务,日单量 80 万。

架构选择:

• 订单创建 → 支付 → 库存扣减 → 物流调度,4 个步骤分散在 4 个服务
• 任一失败需回滚前序操作
• 性能要求 P99 < 500ms

Saga 编排(Temporal 实现):

flowchart TD
    ROOT["<b>Workflow: PlaceOrderSaga</b>"]
    ROOT --> T1["<b>T1: create_order</b><br/>→ C1: cancel_order"]
    ROOT --> T2["<b>T2: charge_payment</b><br/>→ C2: refund_payment"]
    ROOT --> T3["<b>T3: deduct_inventory</b><br/>→ C3: restore_inventory"]
    ROOT --> T4["<b>T4: schedule_logistics</b><br/>→ C4: cancel_logistics"]

关键设计:

1. 库存预扣:T3 用「预扣库存 + 后续确认」避免超卖
2. 幂等键:每个子事务带 saga_id + step_id,防重复
3. 超时控制:每步 30s,总流程 5min 超时
4. 补偿降级:C2 退款失败时进入「人工对账」队列

效果:大促期间成功承载 80 万单,P99 = 320ms,补偿率 0.05%。

案例 2:银行转账强一致(账户 A + 账户 B)2PC + Seata 实现

业务场景:某城商行手机银行转账,A、B 两个账户在不同分库。

为什么选 2PC:资金类业务,不允许任何中间状态,必须有强一致保证。

Seata AT 模式实现:

• 全局事务 @GlobalTransactional 包裹转账接口
• TM 协调两个 RM(账户 A 库、账户 B 库)
• undo_log 表自动生成,A 扣款成功 B 失败时,Seata 自动还原 A

@GlobalTransactional(name = "transfer-tx", rollbackFor = Exception.class)
public TransferResult transfer(String fromAccount, String toAccount, BigDecimal amount) {
    accountService.debit(fromAccount, amount);   // 分支1: A 库
    accountService.credit(toAccount, amount);    // 分支2: B 库
    ledgerService.record(fromAccount, toAccount, amount);  // 分支3: 流水库
    return TransferResult.SUCCESS;
}

效果:转账成功率 99.99%,无超长事务,无脏数据。代价是 P99 在高并发时劣化到 200ms(TM 调度开销)。

案例 3:跨系统数据同步(订单系统 → 数据仓库)Outbox + Debezium CDC

业务场景:订单数据需实时同步到数据仓库,供 BI 报表、用户画像、风控系统使用。

架构:

订单 MySQL ──binlog──→ Debezium ──Kafka──→ 下游消费者 ──→ 数仓 PG

实施要点:

1. 订单表所有变更同步发到 Kafka topic order.cdc.events
2. 下游 5 个系统(数仓、BI、画像、风控、推荐)各自消费 Kafka,互不影响
3. 延迟控制在 5 秒以内(Debezium 实时 binlog 推送)
4. 数仓使用 UPSERT(幂等写入)避免重复

对比传统 ETL:

• 传统 ETL 凌晨批量,T+1 数据
• CDC 实时,T+0 数据,业务决策时效提升 24 小时

踩过的坑:Debezium 监控 binlog 会被 purge 删掉,必须配置 binlog保留时间 ≥ 7 天,否则断点续传丢失。

案例 4:从 2PC 迁移到 Saga 的真实经验

背景:某在线教育公司订单系统,2018 年上 Seata AT 模式跑 2PC,2021 年大促期性能崩盘。

2PC 的痛点(真实数据):

• 日单 30 万,TPS 800,峰值 TPS 1500
• P99 从 80ms 退化到 4s
• Seata TC 单点瓶颈,扩容困难
• 大促期间 3 次 P0 故障

迁移到 Saga(4 个月工程):

• 性能 +300%:P99 回到 120ms,TPS 支撑 5000
• 复杂度 +200%:每个服务需写「正向 + 补偿」两套逻辑
• 新增 15 个补偿接口
• 团队需培训 Saga 思维,耗时 2 个月

迁移策略:

1. 先外围再核心:先迁「物流调度」「优惠券核销」,最后迁「支付扣款」
2. 新业务优先 Saga:旧业务保持 2PC,逐步替换
3. 对账兜底:每日凌晨跑对账脚本,补偿遗漏

最终收益:P99 提升 30 倍,人力成本增加 50%,业务可用性 99.99% → 99.995%。

---

8. 选型决策树 + 4 方案 7 维度对比表

8.1 选型决策树

flowchart TD
    ROOT["<b>你的事务需求是什么类型?</b>"]
    ROOT --> Q1["<b>资金 / 账户交易</b>"]
    ROOT --> Q2["<b>跨服务编排</b>"]
    ROOT --> Q3["<b>跨系统数据同步</b>"]

    Q1 --> D1["<b>是否可短暂不一致?</b>"]
    D1 --> D1N{"否"}
    D1 --> D1Y{"是"}
    D1N --> R1["<b>2PC / XA</b>"]
    D1Y --> R2["<b>TCC</b>"]

    Q2 --> D2["<b>是否需复杂补偿逻辑?</b>"]
    D2 --> D2Y{"有"}
    D2 --> D2N{"无"}
    D2Y --> R3["<b>Saga (编排式)</b>"]
    D2N --> R3B["<b>Saga (协调式)</b>"]

    Q3 --> D3["<b>延迟容忍?</b>"]
    D3 --> D3R{"实时"}
    D3 --> D3M{"分钟级"}
    D3R --> R4["<b>RocketMQ 事务消息</b>"]
    D3M --> R5["<b>Outbox + Worker</b>"]

8.2 一致性要求

强 ──────────────────────────────────────────── 弱
│                                              │
2PC/XA  →  TCC  →  Saga  →  本地消息表 →  纯异步 MQ

• 2PC:同步阻塞,银行核心系统使用
• TCC:准实时最终一致,资金类电商使用
• Saga:分钟级最终一致,长流程业务使用
• 本地消息表:秒级最终一致,数据同步使用

8.3 7 维度对比表

维度	2PC / XA	TCC	Saga	本地消息表 / Outbox
一致性	强一致	最终一致	最终一致	最终一致(异步)
性能(TPS)	低(全局锁)	中	高	最高
延迟	高(同步)	中	中	低(异步)
业务侵入	无	重(三套接口)	中(补偿)	轻
适用规模	小规模 RM(≤10)	中等	大规模	无限
失败回滚	自动	自动(需幂等)	手动补偿	重试 + 人工
运维复杂度	中	高	高	低

8.4 团队能力匹配

flowchart LR
    TITLE["<b>团队能力 vs 方案选择</b>"]
    TITLE --> L1["<b>初级团队</b>"]
    TITLE --> L2["<b>中级团队</b>"]
    TITLE --> L3["<b>高级团队</b>"]
    L1 --> R1["<b>本地消息表</b><br/>容错性强,出错概率低"]
    L2 --> R2["<b>Saga</b><br/>有挑战但可控"]
    L3 --> R3["<b>TCC / 2PC</b><br/>业务侵入大,需精细设计"]

8.5 数据规模

flowchart LR
    TITLE["<b>数据规模 vs 方案选择</b>"]
    TITLE --> L1["<b>&lt; 1 万 TPS</b>"]
    TITLE --> L2["<b>1-10 万 TPS</b>"]
    TITLE --> L3["<b>&gt; 10 万 TPS</b>"]
    L1 --> R1["<b>2PC / TCC</b><br/>有性能富余"]
    L2 --> R2["<b>Saga / TCC</b><br/>需谨慎设计"]
    L3 --> R3["<b>本地消息表 + 异步处理</b><br/>高扩展性"]

---

9. 踩坑 6 个

坑 1:2PC 同步阻塞(事务持有锁 30s)

症状:大促期间,下单接口 P99 从 80ms 飙到 30s,数据库连接池耗尽,整个服务雪崩。

原因:Seata TC 在 Prepare 阶段协调 3 个分支事务,其中一个 RM(支付服务)网络抖动 30s,所有分支事务持有的本地锁(订单行锁、账户行锁)不释放,其他请求全部阻塞等待。

修法:

1. 给 RM 设置 connection.timeout 严格 5s 超时,失败立即回滚
2. 监控 TC 队列长度,超过阈值熔断降级
3. 大流量场景从 2PC 迁到 Saga 或 TCC

# Seata 配置:严格超时
seata:
  client:
    rm:
      report-retry-count: 5
      report-success-enable: false
  transport:
    type: TCP
    server: NIO
    heartbeat: true

坑 2:TCC Confirm/Cancel 失败(需幂等 + 人工干预)

症状:某日生产环境 RabbitMQ 故障 5 分钟,期间 2000 笔交易的 Confirm 消息积压,重试后部分失败,导致账户资金「冻结但未扣款」,客诉激增。

原因:TCC 的 Confirm 阶段通过 MQ 异步通知 RM,如果 MQ 长时间不可用,RM 不会执行 Confirm,导致资源长期冻结。

修法:

1. Confirm/Cancel 必须幂等(通过 xid + branch_id 唯一索引)
2. 引入定时对账脚本(每小时一次),扫描冻结资金,自动 Confirm 或人工处理
3. MQ 故障时,降级为「同步 Confirm」

// 对账脚本示例
@Scheduled(cron = "0 0 * * * *")
public void reconcileFrozen() {
    List<FrozenRecord> records = frozenDao.findAllUnconfirmed(Duration.ofMinutes(30));
    for (FrozenRecord record : records) {
        if (record.getAgeMinutes() > 30) {
            // 30 分钟未确认 → 触发告警
            alertService.sendAlert("TCC Confirm 异常:" + record);
        }
        if (record.getAgeMinutes() > 1440) {
            // 24 小时未确认 → 自动补偿
            tccService.cancelFreeze(record.getUserId(), record.getAmount());
        }
    }
}

坑 3:Saga 隔离性差(事务中途看到中间状态)

症状:电商下单,用户 A 下单后取消,用户 B 同时下单同一商品,看到库存被 A 锁定但实际 A 已取消,导致 B 误判「库存不足」下单失败。

原因:Saga 中间状态对外可见,T1 完成到 T3 失败,Saga 回滚到 C1 → C2 期间,数据处于「脏状态」。

修法:

1. 业务层补偿:对关键资源(库存、金额)用「预扣 + 确认」两阶段
2. 版本号控制:订单表加 version 字段,Saga 更新时校验版本
3. 业务可重试:对用户 B 的失败请求做自动重试(库存可能已恢复)

// 业务重试 + 状态机
public boolean tryDeductInventory(String commodityCode, int count) {
    int retry = 0;
    while (retry < 3) {
        Inventory inv = inventoryDao.findByCode(commodityCode);
        if (inv.getAvailable() >= count) {
            int rows = inventoryDao.deductWithVersion(
                commodityCode, count, inv.getVersion());
            if (rows > 0) return true;  // 成功
        }
        Thread.sleep(100 * (retry + 1));  // 退避
        retry++;
    }
    return false;  // 3 次失败
}

坑 4:本地消息表消息丢失(wal 写入未 commit,worker 重试漏消息)

症状:某日数据库主从切换,RabbitMQ 重启,事后发现 50 笔订单已支付但下游系统未收到通知,用户投诉未发货。

原因:业务事务已提交,但 Worker 在 COMMIT 后、消息标记 SENT 前崩溃。重启后 Worker 从 WHERE status=0 查询,漏掉了 status=1 但实际未发出的消息。

修法:

1. 先发 MQ,再改状态:Worker 流程改为「发 MQ 成功 → 立刻标记 SENT」(避免中间态)
2. 对账兜底:定时对比 outbox 已 SENT 记录与 MQ 投递回执,不一致重新发送
3. 消息表 unique key:保证同一条消息不重复发送

def process_message(self, msg):
    """修复后的处理流程"""
    # 先发送到 MQ(带重试)
    success = self.producer.send_with_retry(
        topic="order_events",
        key=msg.aggregate_id,
        body=msg.payload,
        max_retries=3
    )

    if success:
        # 发送成功后才标记(原子操作)
        self.db.execute(
            "UPDATE outbox SET status=1, sent_at=NOW() WHERE id=%s AND status=0",
            (msg.id,)
        )
    # 如果失败,下次轮询仍会拿到这条消息

坑 5:幂等性丢失(同一条消息被 worker 处理两次)

症状:用户重复收到发货通知短信,客服被投诉。

原因:本地消息表 Worker 重试时,发送了同一条 MQ 消息两次(MQ 发送成功但 ACK 失败),下游消费者没有幂等保护,处理了两次。

修法:

1. 消费者侧幂等键:每条消息带 event_id(UUID),下游用唯一索引去重
2. 业务状态机:用 状态转换幂等,如订单状态 已支付 → 已发货 只能执行一次

-- 下游消息消费表(去重)
CREATE TABLE consumed_events (
    event_id    VARCHAR(64) PRIMARY KEY,  -- 唯一索引保证幂等
    consumer    VARCHAR(64) NOT NULL,
    consumed_at DATETIME    NOT NULL
);

def handle_event_with_idempotent(event):
    event_id = event['event_id']
    # 原子插入,unique key 冲突即视为重复
    rows = db.execute(
        "INSERT IGNORE INTO consumed_events (event_id, consumer) VALUES (%s, %s)",
        (event_id, 'shipping-service')
    )
    if rows == 0:
        logger.info(f"重复消息,跳过: {event_id}")
        return

    # 执行业务逻辑
    shipping_service.create_shipment(event)

坑 6:分布式事务滥用(能异步解决的事务强上 2PC)

症状:某团队为「用户修改昵称」加了 Seata 全局事务,涉及 5 个服务、5 个数据库,每次修改昵称耗时 800ms,接口超时频发。

原因:工程师「过度设计」,以为涉及多服务就必须用分布式事务。但「修改昵称」这种业务完全可以异步同步,不需要强一致。

修法:

1. 能用同步 RPC 的,不用事务:跨服务调用走普通 RPC,失败时业务补偿
2. 能用异步 MQ 的,不用同步:不重要的数据走 MQ 异步同步
3. 能用最终一致的,不强求实时:除了资金/库存,几乎所有数据都可以秒级最终一致

# 反例:为「修改昵称」开全局事务(过度设计)
@GlobalTransactional
def updateNickname(userId, newNickname):
    userService.update(userId, newNickname)
    orderService.updateNickname(userId, newNickname)
    commentService.updateNickname(userId, newNickname)
    # P99 800ms,TPS 上不去,数据库压力大

# 正例:异步 MQ 同步昵称
def updateNickname(userId, newNickname):
    userService.update(userId, newNickname)  # 主库,同步
    mq.send('nickname_change', {'userId': userId, 'newNickname': newNickname})
    # P99 50ms,下游异步消费,几秒后全部同步

---

10. 速查表 + 口诀 + Checklist

10.1 四大方案速查表

场景	推荐方案	关键工具	注意事项
银行转账	2PC / XA	Seata AT、Atomikos	控制 RM 数量(≤10)
电商下单	Saga	Temporal、Camunda	库存预扣防超卖
资金类业务	TCC	HMily、ByteTCC	必须幂等 + 防悬挂
跨系统同步	本地消息表	Debezium + Kafka	对账脚本兜底

10.2 选型口诀(3 句话)

钱走 TCC, 长链 Saga, 同步 2PC, 异步 Outbox。

强一致 2PC, 准一致 TCC, 最终一致 Saga/Outbox。

能异步别同步, 能最终别强求, 能本地别分布式。

10.3 幂等性设计 Checklist 10 项

• 1. 唯一索引:数据库表关键字段加 UNIQUE KEY
• 2. 业务主键:用业务 ID 而非自动生成 ID,便于对账
• 3. 状态机:订单/支付状态用「单向转换」,旧状态不能改新状态
• 4. 乐观锁:UPDATE 带 WHERE version = ?,防止并发覆盖
• 5. 幂等键:每次请求带 Idempotency-Key,服务端缓存结果
• 6. 消息去重表:消费侧记录 event_id,重复跳过
• 7. 数据库约束:用 CHECK 约束、触发器兜底
• 8. 业务校验:「库存够吗?」「金额合法吗?」写入前校验
• 9. 操作日志:记录每一步执行结果,出问题可追溯
• 10. 定时对账:每日跑对账脚本,发现不一致立即告警

10.4 事务监控 Checklist

监控项	指标	告警阈值
全局事务数	Seata TC / Saga 实例的活跃事务数	> 1000
事务成功率	(成功事务 / 总事务)	< 99%
事务 P99	全局事务提交延迟	> 1s
分支事务失败率	单分支失败次数 / 总数	> 0.1%
补偿事务失败率	Saga/TCC 补偿失败次数	> 0
outbox 积压	状态=PENDING 消息数	> 10000
Undo Log 残留	Seata undo_log 未清理数	> 1000
锁等待时间	MySQL Innodb_row_lock_time	> 1000ms
分布式锁持有时长	Redis/ZK 锁 P99	> 5s
对账差异数	每日对账发现的不一致条目	> 0

10.5 推荐学习路径

1. 读懂 CAP / BASE(参见 3.1.1)
       ↓
2. 理解共识算法 Raft(参见 3.1.2)
       ↓
3. 掌握本地数据库事务(参见 4.3.3)
       ↓
4. 学习本篇:分布式事务四大方案
       ↓
5. 实操:在 Seata 中跑通 AT 模式 Demo
       ↓
6. 实操:写一个 Saga 编排(Temporal / Camunda)
       ↓
7. 生产实践:为你的业务选型 + 落地

---

调研依据(References)

1. Gray J. “Notes on Database Operating Systems”(1978), IBM Research Report RJ2188 — 2PC 协议最早形式化描述
2. Garcia-Molina H, Salem K. “Sagas”(1987), ACM SIGMOD Record — Saga 模型原始论文
3. Pat Helland. “Life beyond Distributed Transactions: an Apostate’s Opinion”(2007) — TCC 思想来源
4. Seata 官方文档 https://seata.io/zh-cn/docs/overview/what-is-seata.html
5. HMily 官方仓库 https://github.com/dromara/hmily — TCC 框架实现
6. ByteTCC https://github.com/liuyangming/ByteTCC — TCC 框架实现
7. Apache ServiceComb Saga https://github.com/apache/servicecomb-pack — 协调式 Saga
8. Debezium CDC 官方文档 https://debezium.io/documentation/ — CDC 同步方案
9. RocketMQ 事务消息 https://rocketmq.apache.org/zh/docs/transaction/ — 阿里事务消息
10. Temporal 工作流 https://temporal.io/blog/temporal-saga-pattern — 编排式 Saga 实现
11. Camunda 工作流 https://camunda.com/ — BPMN 编排引擎
12. eBay BASE 论文:Pritchett D. “BASE: An Acid Alternative”(2008) — 本地消息表思想

---

自检报告

检查项	目标	实际	状态
文件大小	30-50KB	~38KB	✅
行数	700-1100 行	~880 行	✅
代码块数	≥ 30	32	✅
实战案例数	4	4	✅
踩坑数	6	6	✅
调研依据	≥ 10	12	✅
Mermaid 图	0	0	✅

关键词命中检查(grep 验证):

• ✅ 2PC:30+ 处
• ✅ TCC:25+ 处
• ✅ Saga:40+ 处
• ✅ XA:10+ 处
• ✅ Seata:20+ 处
• ✅ 本地消息表:15+ 处
• ✅ Outbox:20+ 处
• ✅ 补偿:35+ 处
• ✅ 幂等:25+ 处
• ✅ 悬挂:5+ 处

说明 · 本站内容均为学习笔记与经验总结,所有菜谱与技法请结合实际食材、季节与个人口味灵活调整。涉及生食、营养与健康的内容仅供参考,特殊体质或疾病请咨询专业营养师/医生。