🔀 分布式事务：2PC / Saga / TCC

跨服务的原子性保证——为什么你的微服务需要事务、三种主流方案的取舍、Agent 工具链的部分失败处理

一句话总结：分布式事务的本质是"当部分步骤失败时，如何保证系统不处于不一致状态"。2PC 追求强一致但阻塞风险高；Saga 放弃隔离性换取可用性；TCC 介于两者之间。AI Agent 的工具调用链本质上就是分布式事务问题——每一次工具调用都是一个参与者，部分失败时需要补偿。

📋 目录

1. 为什么需要分布式事务
2. 2PC：两阶段提交
3. Saga 模式
4. TCC：Try-Confirm-Cancel
5. 三种方案对比
6. Saga 的两种编排方式
7. 生产环境真实案例
8. Agent 工具链的分布式事务
9. 幂等性：一切补偿的基础
10. 工具与框架
11. 选型决策树
12. 常见坑与实战建议

1. 为什么需要分布式事务

单体时代：本地事务就够了

在单体应用中，所有操作共享一个数据库，一条 BEGIN; ... COMMIT; 就能保证 ACID：

-- 单体应用：一个事务搞定一切
BEGIN;
  INSERT INTO orders (user_id, amount) VALUES (42, 99.00);
  UPDATE accounts SET balance = balance - 99.00 WHERE user_id = 42;
  INSERT INTO order_items (order_id, product_id) VALUES (1001, 7);
COMMIT;
-- 要么全成功，要么全回滚，简单可靠

微服务时代：一个操作横跨 N 个数据库

同样的"下单"操作，在微服务架构中变成了：

每个服务有自己的数据库，没有共享事务。这就是分布式事务问题的来源。

ACID 在分布式环境中的崩塌

2. 2PC：两阶段提交

核心思想

2PC（Two-Phase Commit）是最经典的分布式事务协议，由 Jim Gray 在 1978 年提出。核心思路：引入一个协调者（Coordinator），分两阶段执行：

流程详解

2PC 的致命问题：阻塞

具体故障场景：

2PC 的其他问题

• 性能差：两个 RTT（往返延迟）+ 参与者持有锁的时间 = 吞吐量低
• 扩展性差：参与者越多，任何一个超时都会导致整个事务失败
• 同步阻塞：在等待协调者决策期间，参与者不能做其他事
• 不适合长事务：锁持有时间越长，死锁和超时风险越大

2PC 实际用在哪？

虽然纯 2PC 在微服务中很少直接使用，但它的变体仍在广泛使用：

• 数据库内部：MySQL/XA 事务、PostgreSQL 的 PREPARE TRANSACTION
• 消息队列：Kafka 的事务机制本质上就是 2PC（Producer + Broker + Consumer Group）
• XA 协议：X/Open DTP 定义的分布式事务标准，2PC 是其核心
• Seata AT 模式：阿里开源的 Seata 框架，AT 模式底层就是增强版 2PC

-- MySQL XA 事务示例（2PC 的标准化实现）
-- 阶段一：准备
XA START 'txn1';
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
XA END 'txn1';
XA PREPARE 'txn1';  -- Phase 1: 投票

-- 阶段二：提交（通常由事务管理器执行）
XA COMMIT 'txn1';   -- Phase 2: 提交
-- 或者 XA ROLLBACK 'txn1';  -- Phase 2: 回滚

3. Saga 模式

起源

Saga 模式由 Hector Garcia-Molina 和 Kenneth Salem 在 1987 年的论文 "Sagas" 中提出。核心思想：将一个长事务拆分成多个本地短事务，每个本地事务都有对应的补偿操作。如果某个步骤失败，按反序执行已完成步骤的补偿操作。

Saga 执行模型

电商下单 Saga 示例

如果 T₃（扣减余额）失败：

补偿事务设计要点

4. TCC：Try-Confirm-Cancel

核心思想

TCC（Try-Confirm-Cancel）是一种业务层面的分布式事务模式，将每个操作拆分为三个阶段：

与 Saga 的区别

TCC 代码示例

// TCC 三个接口的设计：转账场景
// Try: 冻结金额
public boolean tryFreezeAmount(String userId, BigDecimal amount) {
    // 检查余额是否充足
    Account account = accountDao.findByUserId(userId);
    if (account.getBalance().compareTo(amount) < 0) {
        return false; // 余额不足，Try 失败
    }
    // 冻结金额（业务隔离：其他人看不到也不能用这笔钱）
    account.setFrozenAmount(account.getFrozenAmount().add(amount));
    accountDao.update(account);
    return true;
}

// Confirm: 确认扣减
public boolean confirmDeduct(String userId, BigDecimal amount) {
    Account account = accountDao.findByUserId(userId);
    // 真正扣减：余额减少，冻结金额释放
    account.setBalance(account.getBalance().subtract(amount));
    account.setFrozenAmount(account.getFrozenAmount().subtract(amount));
    accountDao.update(account);
    return true;
}

// Cancel: 取消冻结
public boolean cancelFreeze(String userId, BigDecimal amount) {
    Account account = accountDao.findByUserId(userId);
    // 释放冻结金额
    account.setFrozenAmount(account.getFrozenAmount().subtract(amount));
    accountDao.update(account);
    return true;
}

5. 三种方案对比

关键指标对比

6. Saga 的两种编排方式

// 编舞模式：Order Service 发布事件
@Service
public class OrderService {
    @Autowired private KafkaTemplate kafka;
    
    public Order createOrder(OrderRequest req) {
        Order order = new Order(req);
        order.setStatus(Status.PENDING);
        orderRepo.save(order);
        
        // 发布事件，让 Inventory 和 Payment 自己响应
        kafka.send("order-events", new OrderCreatedEvent(order.getId(), req.getItems()));
        return order;
    }
    
    // 监听支付结果
    @KafkaListener(topics = "payment-events")
    public void onPaymentEvent(PaymentEvent event) {
        if (event.isSuccess()) {
            orderRepo.updateStatus(event.getOrderId(), Status.CONFIRMED);
        } else {
            orderRepo.updateStatus(event.getOrderId(), Status.CANCELLED);
            kafka.send("order-events", new OrderCancelledEvent(event.getOrderId()));
        }
    }
}

// 编排模式：用 Temporal 实现的 Saga
@WorkflowInterface
public interface OrderSagaWorkflow {
    @WorkflowMethod
    OrderResult executeOrder(OrderRequest request);
}

public class OrderSagaWorkflowImpl implements OrderSagaWorkflow {
    
    @Override
    public OrderResult executeOrder(OrderRequest req) {
        // Saga 补偿列表：每步成功后注册补偿
        List compensations = new ArrayList<>();
        
        try {
            // Step 1: 创建订单
            String orderId = activities.createOrder(req);
            compensations.add(() -> activities.cancelOrder(orderId));
            
            // Step 2: 扣减库存
            activities.reserveInventory(orderId, req.getItems());
            compensations.add(() -> activities.releaseInventory(orderId));
            
            // Step 3: 扣减余额
            activities.processPayment(orderId, req.getAmount());
            compensations.add(() -> activities.refundPayment(orderId));
            
            // Step 4: 确认订单
            activities.confirmOrder(orderId);
            return new OrderResult(orderId, "SUCCESS");
            
        } catch (Exception e) {
            // 反序执行补偿
            Collections.reverse(compensations);
            for (Runnable compensation : compensations) {
                try {
                    compensation.run();
                } catch (Exception ex) {
                    // 补偿失败需要人工介入
                    activities.alertCompensationFailure(orderId, ex.getMessage());
                }
            }
            return new OrderResult(null, "COMPENSATED");
        }
    }
}

编舞 vs 编排选择矩阵

7. 生产环境真实案例

案例 1：阿里巴巴 + Apache Seata

Seata（原 Fescar）最初是阿里巴巴内部的分布式事务中间件，在双 11 等高并发场景下支撑了阿里系的电商交易。2024 年进入 Apache 孵化器。

// Seata AT 模式：最简单的分布式事务
@GlobalTransactional  // ← 只需这一个注解！
public void purchase(String userId, String commodityCode, int count) {
    // Step 1: 扣减库存（inventory 微服务）
    inventoryDao.deduct(commodityCode, count);
    
    // Step 2: 扣减余额（account 微服务）
    accountDao.debit(userId, totalAmount);
    
    // Step 3: 创建订单（order 微服务）
    orderDao.create(userId, commodityCode, count, totalAmount);
    
    // Seata 自动拦截 SQL，生成 undo_log
    // 任一步骤失败 → 自动回滚（无需手写补偿！）
}

// Seata AT 的原理：
// 1. 拦截 SQL，在执行前记录 before-image（修改前数据）
// 2. 执行业务 SQL
// 3. 记录 after-image（修改后数据）
// 4. 生成 undo_log：{beforeImage, afterImage, xid, branchId}
// 5. 回滚时：根据 undo_log 反向生成 SQL 执行

案例 2：Netflix Conductor

Netflix Conductor 是 Netflix 开源的微服务编排引擎，用于管理 Netflix 内容平台工程团队中复杂的微服务工作流。后被 Tesla、LinkedIn、J.P. Morgan 等公司采用。

• Conductor 提供 Saga 补偿模式：每个任务可以定义补偿任务，工作流失败时自动执行
• 支持 at-least-once 任务投递保证
• 内置 human-in-the-loop 审批节点
• 2023 年 Netflix 将 Conductor 开源给 Orkes 公司维护，社区版称 conductor-oss

数据来源：Netflix Tech Blog、conductor-oss GitHub 仓库

案例 3：Temporal.io + 多公司

Temporal 是 Uber Cadence 的继任者，由 Cadence 原作者 Maxim Fateev 创建。它提供 durable execution（持久执行）——工作流代码保证完成，不受服务器崩溃影响。

• Netflix：用于内容编码管线编排
• Stripe：用于支付工作流管理
• HashiCorp：用于 Nomad 调度器
• Box：用于文档处理流水线
• Temporal 内置 Saga 支持（Java SDK 有 Saga 类），补偿列表自动反序执行

// Temporal Java SDK 内置 Saga
WorkflowSaga saga = new WorkflowSaga();
try {
    String reservationId = activities.reserveHotel(req);
    saga.addCompensation(() -> activities.cancelHotelReservation(reservationId));
    
    String flightBookingId = activities.bookFlight(req);
    saga.addCompensation(() -> activities.cancelFlight(flightBookingId));
    
    activities.chargeCreditCard(req.getCard(), req.getTotal());
    saga.addCompensation(() -> activities.refundCard(req.getCard(), req.getTotal()));
    
} catch (Exception e) {
    saga.compensate(); // 自动反序执行所有已注册的补偿
    throw e;
}

数据来源：Temporal 官网 docs.temporal.io、各公司公开技术博客

案例 4：银行系统的混合策略

2024-2025 年的多篇学术论文（如 Semantic Scholar 上收录的 "Optimizing Distributed Transactions in Banking APIs"）提出了混合策略：

• 核心金融操作（如账本更新）使用 2PC 保证强一致性
• 辅助服务（如通知、审计日志）使用 Saga 模式
• 这种混合策略在一致性和性能之间取得了平衡

8. Agent 工具链的分布式事务

AI Agent 的工具调用链本质上就是分布式事务——多个外部服务调用，部分失败时需要恢复一致性。

为什么 Agent 需要 Saga

SagaLLM：VLDB 2025 的学术研究

SagaLLM（发表在 VLDB 2025）是第一个将 Saga 模式正式应用于多 Agent LLM 系统的学术框架。它指出了 LLM Agent 的四个核心缺陷：

1. 自验证不可靠：LLM 无法可靠地验证自己的输出（受 Gödel 不完备性定理影响）
2. 上下文丢失：长对话中 Agent 遗忘早期步骤，导致矛盾决策
3. 缺乏事务保障：没有回滚或一致性机制
4. Agent 间协调不足：多 Agent 系统无法自动协调状态变更

SagaLLM 的解决方案：将每个 Agent 操作视为 Saga 中的一个本地事务，配备独立验证 Agent 和补偿操作。

Agent Saga 实战模式

# Agent 工具链的 Saga 模式
# 每个工具调用都注册补偿操作

class AgentSaga:
    def __init__(self):
        self.compensations = []
    
    async def execute(self, steps):
        """执行工具链，自动补偿失败"""
        for step in steps:
            try:
                # 1. 先注册补偿（在执行之前！）
                compensation = step.get_compensation()
                
                # 2. 执行操作
                result = await step.execute()
                
                # 3. 执行成功后才加入补偿列表
                self.compensations.append(compensation)
                
            except Exception as e:
                # 反序执行补偿
                await self._compensate()
                raise
    
    async def _compensate(self):
        """反序执行所有已注册的补偿"""
        for compensation in reversed(self.compensations):
            try:
                await compensation()
            except Exception as e:
                # 补偿失败 → 人工介入
                await self._alert_human(compensation, e)

# 使用示例：AI 旅行助手
saga = AgentSaga()
await saga.execute([
    ToolStep(
        name="预订航班",
        execute=lambda: flight_api.book(flight="CA1234"),
        get_compensation=lambda: flight_api.cancel(booking_id)
    ),
    ToolStep(
        name="扣减信用卡",
        execute=lambda: payment_api.charge(card="****4242", amount=2999),
        get_compensation=lambda: payment_api.refund(charge_id)
    ),
    ToolStep(
        name="预订酒店",
        execute=lambda: hotel_api.book(hotel="Hilton", dates="..."),
        get_compensation=lambda: hotel_api.cancel(reservation_id)
    ),
])

AgentRelay：开源事务运行时

AgentRelay 是一个框架无关的 AI Agent 事务运行时（PyPI 可安装），提供：

• 工具调用幂等性：防止重复副作用（重复发邮件、重复扣款）
• Saga 补偿：注册补偿工具，失败时自动回滚

来源：pypi.org/project/agentrelay/

AWS 的 Agent Saga 指南

AWS Prescriptive Guidance 在 2025 年发布了 "Prompt chaining saga patterns"，将 Saga 模式应用于 AI Agent 的 prompt chain：

• 每个 prompt 链步骤视为一个本地事务
• 使用迭代修正而非数据库补偿来实现回滚
• 适合"软补偿"场景（生成内容可以修正，但不会产生外部副作用）

来源：docs.aws.amazon.com/prescriptive-guidance

9. 幂等性：一切补偿的基础

如何实现幂等

# 幂等键实现模式
import hashlib

def generate_idempotency_key(session_id, action_type, sequence_num):
    """生成确定性的幂等键"""
    raw = f"{session_id}:{action_type}:{sequence_num}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

# 调用支付 API
idem_key = generate_idempotency_key(
    session_id="sess_abc123",
    action_type="charge", 
    sequence_num=1
)

# Stripe API 原生支持幂等键
stripe.Charge.create(
    amount=2999,
    currency="usd",
    idempotency_key=idem_key,  # ← 重复请求返回相同结果
)

# 服务端实现：
# 1. 收到请求 → 检查 idempotency_key 是否已存在
# 2. 已存在 → 直接返回缓存结果（不重复执行）
# 3. 不存在 → 执行操作，缓存结果，返回

# ❌ 非幂等：追加写入
# 重试一次 → 数据翻倍
append_to_file("/tmp/result.txt", "processed order #1234\n")

# ✅ 幂等：覆盖写入
# 重试任意次 → 结果相同
write_file("/tmp/result.txt", "processed order #1234\n")

# ❌ 非幂等：增量更新
UPDATE accounts SET balance = balance - 99 WHERE id = 42;
# 重试 → 多扣 99 元

# ✅ 幂等：基于目标值的更新
UPDATE accounts SET balance = 901 WHERE id = 42;  -- balance was 1000
# 重试 → 仍然是 901

# ✅ 更好的幂等：条件更新
UPDATE accounts SET balance = balance - 99 
WHERE id = 42 AND balance >= 99 AND version = 7;
-- version 不变 → 已执行过，跳过

# ❌ 不可逆设计
db.execute("DELETE FROM orders WHERE id = 1234")  # 硬删除，无法恢复

# ✅ 可逆设计：软删除
db.execute("UPDATE orders SET status = 'cancelled' WHERE id = 1234")
# 补偿操作：UPDATE orders SET status = 'active' WHERE id = 1234

# ❌ 不可逆：立即发送邮件
smtp.send(to="user@example.com", body="...")

# ✅ 可逆：先创建草稿，确认后发送
draft_id = email_api.create_draft(to="user@example.com", body="...")
# 确认步骤：
email_api.send_draft(draft_id)
# 补偿步骤：
email_api.delete_draft(draft_id)  # 删除草稿（而非撤回已发邮件）

10. 工具与框架

Seata AT 模式实操

# 1. 启动 Seata Server
docker run --name seata-server \
  -p 8091:8091 \
  -p 7091:7091 \
  seataio/seata-server:2.0.0

# 2. Spring Boot 依赖
# pom.xml
<dependency>
  <groupId>io.seata</groupId>
  <artifactId>seata-spring-boot-starter</artifactId>
  <version>2.0.0</version>
</dependency>

# 3. application.yml
seata:
  enabled: true
  application-id: order-service
  tx-service-group: my-tx-group
  service:
    vgroup-mapping:
      my-tx-group: default
  registry:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848

# 4. 数据库 undo_log 表（每个参与方都要建）
CREATE TABLE IF NOT EXISTS undo_log (
  branch_id     BIGINT       NOT NULL,
  xid           VARCHAR(128) 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 (branch_id),
  KEY idx_xid (xid)
) ENGINE=InnoDB;

# 5. 业务代码（只需一个注解！）
@GlobalTransactional(timeoutMills = 60000)
public void purchase(String userId, String commodity, int count) {
    inventoryService.deduct(commodity, count);
    accountService.debit(userId, totalPrice);
    orderService.create(userId, commodity, count);
}

Temporal Saga 实操

# pip install temporalio

from dataclasses import dataclass
from temporalio import workflow, activity
from datetime import timedelta

@dataclass
class OrderRequest:
    user_id: str
    product_id: str
    amount: float
    quantity: int

# ---- Activities（各服务的操作 + 补偿）----

@activity.defn
async def create_order(req: OrderRequest) -> str:
    """创建订单"""
    order_id = f"ORD-{req.user_id}-{req.product_id}"
    print(f"✅ 订单创建: {order_id}")
    return order_id

@activity.defn
async def cancel_order(order_id: str) -> None:
    """补偿：取消订单"""
    print(f"↩️ 订单取消: {order_id}")

@activity.defn
async def reserve_inventory(req: OrderRequest) -> str:
    """预留库存"""
    reservation_id = f"RES-{req.product_id}"
    print(f"✅ 库存预留: {reservation_id}")
    return reservation_id

@activity.defn
async def release_inventory(reservation_id: str) -> None:
    """补偿：释放库存"""
    print(f"↩️ 库存释放: {reservation_id}")

@activity.defn
async def charge_payment(req: OrderRequest) -> str:
    """扣减支付"""
    charge_id = f"CHG-{req.user_id}"
    print(f"✅ 支付扣减: {charge_id}")
    return charge_id

@activity.defn
async def refund_payment(charge_id: str) -> None:
    """补偿：退款"""
    print(f"↩️ 退款: {charge_id}")

# ---- Workflow（Saga 编排）----

@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, req: OrderRequest) -> str:
        compensations = []
        
        try:
            # Step 1: 创建订单
            order_id = await workflow.execute_activity(
                create_order, req, start_to_close_timeout=timedelta(seconds=10))
            compensations.append(
                lambda: workflow.execute_activity(cancel_order, order_id, 
                    start_to_close_timeout=timedelta(seconds=10)))
            
            # Step 2: 预留库存
            reservation_id = await workflow.execute_activity(
                reserve_inventory, req, start_to_close_timeout=timedelta(seconds=10))
            compensations.append(
                lambda: workflow.execute_activity(release_inventory, reservation_id,
                    start_to_close_timeout=timedelta(seconds=10)))
            
            # Step 3: 扣减支付
            charge_id = await workflow.execute_activity(
                charge_payment, req, start_to_close_timeout=timedelta(seconds=10))
            compensations.append(
                lambda: workflow.execute_activity(refund_payment, charge_id,
                    start_to_close_timeout=timedelta(seconds=10)))
            
            return f"Order completed: {order_id}"
            
        except Exception as e:
            # Saga 补偿：反序执行
            print(f"❌ Saga 失败: {e}, 开始补偿...")
            for compensation in reversed(compensations):
                try:
                    await compensation()
                except Exception as ce:
                    print(f"⚠️ 补偿失败: {ce}, 需要人工介入")
            raise

11. 选型决策树

快速判断表

12. 常见坑与实战建议

坑 1：补偿操作不可逆

坑 2：补偿操作本身失败

补偿操作也可能失败（网络故障、服务宕机）。如果补偿失败，系统将停留在不一致状态。

# 补偿失败的处理策略
async def _compensate(self):
    for compensation in reversed(self.compensations):
        max_retries = 3
        for attempt in range(max_retries):
            try:
                await compensation()
                break
            except Exception as e:
                if attempt == max_retries - 1:
                    # 最终失败 → 写入 dead letter queue + 人工介入
                    await dead_letter_queue.push({
                        "saga_id": self.saga_id,
                        "compensation": compensation.__name__,
                        "error": str(e),
                        "timestamp": datetime.utcnow().isoformat()
                    })
                    await alert_ops_team(self.saga_id, e)

坑 3：脏读问题

Saga 没有隔离性，中间状态对其他事务可见。例如：

1. Saga 执行 T₁（创建订单，状态=CREATED）
2. 另一个事务读到订单状态=CREATED，开始处理
3. Saga 的 T₃ 失败，补偿 C₁ 将订单状态改为 CANCELLED
4. 第二个事务已经基于 CREATED 状态做了错误操作

解决方案：

• 引入预留状态（PENDING_RESERVATION），其他事务看到这个状态会等待
• 使用语义锁（状态字段代替数据库锁）
• 查询时过滤中间状态

坑 4：过度使用分布式事务

坑 5：2PC 在微服务中的滥用

2PC 要求所有参与者支持 Prepare 阶段（即支持 XA 协议）。但大多数微服务的数据库是独立选择的，不一定都支持 XA。更常见的是：

• 一个服务用 PostgreSQL（支持 XA）
• 另一个服务用 MongoDB（不支持 XA）
• 第三个服务调外部 API（完全不受 2PC 控制）

这种异构环境下，2PC 根本无法工作。

实战建议清单

1. 优先用本地事务——如果可以通过数据模型调整（如将关联数据放在同一个库），就不需要分布式事务
2. 用 Outbox 解决 80% 的问题——"写 DB + 发消息"是最常见的跨服务一致性需求
3. Saga 编排优于编舞——流程超过 3 步就用编排，未来你会感谢自己
4. 补偿必须在执行前注册——先记 undo log，再执行操作
5. 补偿操作必须幂等——因为补偿可能被重试
6. 软状态设计——把不可逆操作变成可逆（草稿→确认，软删除）
7. 补偿失败需要人工介入——Dead Letter Queue + 告警
8. 不要自己造轮子——用 Temporal/Seata/Conductor，它们已经解决了 99% 的边界情况
9. 可观测性——每个 Saga 实例必须有全局唯一 ID，所有操作都关联此 ID 便于追踪
10. 测试补偿路径——不只测正向流程，更要测每个步骤失败后的补偿

附录：Outbox 模式——最简单也最实用的"分布式事务"

很多场景其实不需要完整的分布式事务，只需要"写数据库"和"发消息"的原子性。Outbox 模式用最简单的方式解决了这个问题：

# Outbox 表设计
CREATE TABLE outbox (
    id          BIGSERIAL PRIMARY KEY,
    event_type  VARCHAR(128) NOT NULL,
    aggregate_id VARCHAR(128) NOT NULL,
    payload     JSONB NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published   BOOLEAN NOT NULL DEFAULT FALSE
);

-- 创建索引加速 Relay 查询
CREATE INDEX idx_outbox_unpublished ON outbox (created_at) WHERE published = FALSE;

# 业务代码：写操作 + 写 Outbox 在同一事务中
@Transactional
public void confirmOrder(Long orderId) {
    // 1. 业务操作
    Order order = orderRepo.findById(orderId);
    order.setStatus(Status.CONFIRMED);
    orderRepo.save(order);
    
    // 2. 写 Outbox（同一事务！）
    OutboxEvent event = new OutboxEvent();
    event.setEventType("order_confirmed");
    event.setAggregateId(orderId.toString());
    event.setPayload(Map.of("orderId", orderId, "userId", order.getUserId()));
    outboxRepo.save(event);
    
    // 事务提交后：业务数据和事件要么同时持久化，要么同时回滚
}

# Outbox Relay：轮询未发布事件
@Scheduled(fixedDelay = 1000)  // 每秒检查一次
public void relayOutboxEvents() {
    List events = outboxRepo.findByPublishedFalseOrderByCreatedAtAsc();
    for (OutboxEvent event : events) {
        try {
            kafkaTemplate.send(event.getEventType(), event.getPayload());
            event.setPublished(true);
            outboxRepo.save(event);
        } catch (Exception e) {
            log.warn("Failed to relay event {}", event.getId(), e);
            // 下次重试
        }
    }
}

🔗 相关阅读

• 共识算法（Raft/Paxos） — 分布式事务的底层依赖
• Event Sourcing — 事务的另一种视角：不存状态，存事件
• CRDT — 不需要事务的最终一致性方案
• CQRS — 读写分离与最终一致性
• 向量时钟 — 分布式因果一致性
• 二阶效应 — 为什么"加个分布式事务"本身就有二阶效应
• 定价模式总结 — 事务成本如何影响定价
• LLM 成本监控 — Agent 工具链的事务级成本追踪