分布式架构中基于 Saga 模式的订单服务一致性设计与落地实践

在订单场景里，“创建订单成功”从来都不是一张 orders 表插入一行那么简单。它往往还要扣库存、冻结余额、发优惠券核销、生成履约单、通知物流，甚至还要写风控事件。单体时代，一个本地事务还能勉强兜住；一旦拆成多个服务，事情就开始变得微妙：每个服务都有自己的数据库，但业务又要求整体看起来像“一次完成”。

这篇文章我会从订单链路这个最常见、也最容易踩坑的场景出发，讲清楚 Saga 模式怎么设计、怎么落地、代码怎么写，以及生产环境里容易翻车的地方该怎么排查。

背景与问题

先看一个典型下单链路：

1. 订单服务创建订单草稿
2. 库存服务预留库存
3. 账户服务冻结用户余额
4. 营销服务核销优惠券
5. 订单服务将订单状态改为 CONFIRMED

听起来很顺，但真正上线后，很容易遇到这些问题：

• 库存扣了，订单没成
• 订单超时取消了，冻结资金没解冻
• 重试导致重复扣库存
• 消息发出去了，但本地事务没提交
• 补偿动作执行了一半，又失败了

很多团队一开始会想用分布式事务，比如 2PC / XA。但在高并发订单场景下，这条路通常不太现实：

• 锁持有时间长，吞吐受影响
• 各服务、数据库、中间件适配复杂
• 故障恢复和运维成本高
• 云原生和异构存储环境下支持不理想

所以更常见的选择是：接受“最终一致性”，通过业务补偿把系统拉回正确状态。
这就是 Saga 模式擅长的地方。

为什么订单场景适合 Saga

Saga 的核心思路很朴素：

• 把一个长事务拆成多个本地事务
• 每一步成功后进入下一步
• 如果某一步失败，就按相反顺序执行已经成功步骤的补偿操作

对订单业务来说，这个思路非常贴合：

• 预留库存失败，可以释放已冻结的余额
• 扣款失败，可以撤销优惠券占用
• 下单超时，可以统一触发补偿回滚

但注意，Saga 不是银弹。它的前提是：

1. 每一步都能设计出可执行的补偿动作
2. 业务能接受短时间不一致
3. 服务接口具备幂等能力
4. 你愿意为“恢复机制”投入设计成本

如果你的业务是“钱一分都不能错，且必须强一致立即可见”，那就不能只靠最基础的 Saga 方案。

核心原理

Saga 常见有两种实现方式：

• Choreography（事件编排 / 事件驱动）

  • 每个服务监听上游事件，再发自己的事件
  • 优点是松耦合
  • 缺点是链路复杂时不好追踪，容易变成“事件意大利面”

• Orchestration（中心编排）

  • 由一个 Saga 协调器统一驱动流程
  • 优点是流程清晰、易观测、易治理
  • 缺点是协调器需要高可用设计

在订单场景里，如果链路已经跨了库存、账户、优惠券、履约几个服务，我更推荐中心编排。因为一旦进入生产环境，定位问题时你会非常感谢自己当初把状态机和补偿路径画清楚了。

一次下单 Saga 的状态流转

stateDiagram-v2
    [*] --> PENDING
    PENDING --> RESERVE_INVENTORY
    RESERVE_INVENTORY --> FREEZE_BALANCE
    FREEZE_BALANCE --> USE_COUPON
    USE_COUPON --> CONFIRMED

    RESERVE_INVENTORY --> COMPENSATING: 库存预留失败
    FREEZE_BALANCE --> COMPENSATING: 余额冻结失败
    USE_COUPON --> COMPENSATING: 优惠券核销失败

    COMPENSATING --> CANCELLED
    CONFIRMED --> [*]
    CANCELLED --> [*]

时序关系

sequenceDiagram
    participant Client as Client
    participant Orchestrator as Saga Orchestrator
    participant Order as Order Service
    participant Inventory as Inventory Service
    participant Account as Account Service
    participant Coupon as Coupon Service

    Client->>Orchestrator: createOrder(request)
    Orchestrator->>Order: createPendingOrder()
    Order-->>Orchestrator: orderCreated

    Orchestrator->>Inventory: reserve(orderId, items)
    Inventory-->>Orchestrator: success/fail

    alt inventory success
        Orchestrator->>Account: freeze(orderId, amount)
        Account-->>Orchestrator: success/fail
    end

    alt account success
        Orchestrator->>Coupon: useCoupon(orderId, couponId)
        Coupon-->>Orchestrator: success/fail
    end

    alt all success
        Orchestrator->>Order: confirm(orderId)
        Order-->>Client: success
    else any step failed
        Orchestrator->>Coupon: compensateUseCoupon()
        Orchestrator->>Account: compensateFreeze()
        Orchestrator->>Inventory: compensateReserve()
        Orchestrator->>Order: cancel(orderId)
        Order-->>Client: failed
    end

三个关键设计点

1. 正向动作与补偿动作成对出现

比如：

补偿并不一定要“恢复到绝对原样”，而是恢复到业务可接受状态。这点非常重要。

2. 每个动作必须幂等

因为网络抖动、超时重试、消息重复投递都很常见。
“冻结余额”这个接口如果重复调用一次就多冻一笔钱，那系统一定会出事故。

所以常见做法是：

• 每个 Saga 步骤带唯一 sagaId / stepId
• 下游服务记录执行日志
• 重复请求直接返回上次结果

3. Saga 状态要持久化

很多示例代码喜欢把流程状态放内存里，Demo 看着爽，生产环境会出事。协调器重启后，如果不知道执行到哪一步，就没法补偿，也没法续跑。

因此至少要持久化：

• saga_id
• order_id
• 当前步骤
• 已完成步骤
• 补偿状态
• 最近一次错误原因
• 重试次数
• 更新时间

方案对比与取舍分析

Saga vs 2PC/XA

编排式 vs 事件驱动式

我的经验是：

• 核心交易主链路：优先编排式
• 外围通知、积分、埋点：优先事件驱动

不要把所有事情都塞进一个大 Saga。主链路越长，失败概率越高，补偿成本越大。

架构设计建议

一个更稳妥的落地方式，通常是下面这套组合：

1. Saga 协调器

   • 负责流程编排、状态流转、重试、补偿

2. 业务服务本地事务

   • 每个服务只保证自己的数据库事务

3. Outbox 模式

   • 本地事务和事件写入同库，避免“数据库提交了，消息没发出去”

4. 幂等表 / 执行日志表

   • 保证动作和补偿可重试

5. 死信队列 / 补偿任务

   • 兜底异常流程

一张简化架构图

flowchart LR
    A[Client] --> B[Saga Orchestrator]
    B --> C[Order Service]
    B --> D[Inventory Service]
    B --> E[Account Service]
    B --> F[Coupon Service]

    C --> CDB[(Order DB)]
    D --> DDB[(Inventory DB)]
    E --> EDB[(Account DB)]
    F --> FDB[(Coupon DB)]

    C --> O1[Outbox]
    D --> O2[Outbox]
    E --> O3[Outbox]
    F --> O4[Outbox]

    O1 --> MQ[Message Broker]
    O2 --> MQ
    O3 --> MQ
    O4 --> MQ

实战代码（可运行）

下面我用 Python 写一个可运行的简化版 Saga 编排器。它不是完整生产代码，但足够把关键设计点讲清楚：状态流转、补偿、幂等、失败回滚。

你可以直接保存为 saga_order_demo.py 运行。

from dataclasses import dataclass, field
from typing import Dict, List, Callable
import uuid


class SagaStepError(Exception):
    pass


@dataclass
class SagaContext:
    saga_id: str
    order_id: str
    user_id: str
    amount: int
    items: Dict[str, int]
    coupon_id: str = ""
    logs: List[str] = field(default_factory=list)


class IdempotencyStore:
    def __init__(self):
        self.done = set()

    def executed(self, key: str) -> bool:
        return key in self.done

    def mark_done(self, key: str):
        self.done.add(key)


class OrderService:
    def __init__(self):
        self.orders = {}

    def create_pending(self, ctx: SagaContext):
        self.orders[ctx.order_id] = "PENDING"
        ctx.logs.append(f"Order {ctx.order_id} created as PENDING")

    def confirm(self, ctx: SagaContext):
        self.orders[ctx.order_id] = "CONFIRMED"
        ctx.logs.append(f"Order {ctx.order_id} confirmed")

    def cancel(self, ctx: SagaContext):
        self.orders[ctx.order_id] = "CANCELLED"
        ctx.logs.append(f"Order {ctx.order_id} cancelled")


class InventoryService:
    def __init__(self, stock: Dict[str, int], idem: IdempotencyStore):
        self.stock = stock
        self.reserved = {}
        self.idem = idem

    def reserve(self, ctx: SagaContext):
        key = f"{ctx.saga_id}:inventory:reserve"
        if self.idem.executed(key):
            ctx.logs.append("Inventory reserve skipped by idempotency")
            return

        for sku, qty in ctx.items.items():
            if self.stock.get(sku, 0) < qty:
                raise SagaStepError(f"Not enough stock for {sku}")

        for sku, qty in ctx.items.items():
            self.stock[sku] -= qty
            self.reserved[(ctx.order_id, sku)] = qty

        self.idem.mark_done(key)
        ctx.logs.append("Inventory reserved")

    def release(self, ctx: SagaContext):
        key = f"{ctx.saga_id}:inventory:release"
        if self.idem.executed(key):
            ctx.logs.append("Inventory release skipped by idempotency")
            return

        for sku, qty in ctx.items.items():
            reserved_qty = self.reserved.get((ctx.order_id, sku), 0)
            if reserved_qty > 0:
                self.stock[sku] += reserved_qty
                self.reserved[(ctx.order_id, sku)] = 0

        self.idem.mark_done(key)
        ctx.logs.append("Inventory released")


class AccountService:
    def __init__(self, balances: Dict[str, int], idem: IdempotencyStore):
        self.balances = balances
        self.frozen = {}
        self.idem = idem

    def freeze(self, ctx: SagaContext):
        key = f"{ctx.saga_id}:account:freeze"
        if self.idem.executed(key):
            ctx.logs.append("Account freeze skipped by idempotency")
            return

        balance = self.balances.get(ctx.user_id, 0)
        if balance < ctx.amount:
            raise SagaStepError("Insufficient balance")

        self.balances[ctx.user_id] -= ctx.amount
        self.frozen[ctx.order_id] = ctx.amount
        self.idem.mark_done(key)
        ctx.logs.append("Account balance frozen")

    def unfreeze(self, ctx: SagaContext):
        key = f"{ctx.saga_id}:account:unfreeze"
        if self.idem.executed(key):
            ctx.logs.append("Account unfreeze skipped by idempotency")
            return

        amount = self.frozen.get(ctx.order_id, 0)
        if amount > 0:
            self.balances[ctx.user_id] += amount
            self.frozen[ctx.order_id] = 0

        self.idem.mark_done(key)
        ctx.logs.append("Account balance unfrozen")


class CouponService:
    def __init__(self, available_coupons: Dict[str, str], idem: IdempotencyStore):
        self.available_coupons = available_coupons
        self.used = {}
        self.idem = idem

    def use(self, ctx: SagaContext):
        if not ctx.coupon_id:
            ctx.logs.append("No coupon to use")
            return

        key = f"{ctx.saga_id}:coupon:use"
        if self.idem.executed(key):
            ctx.logs.append("Coupon use skipped by idempotency")
            return

        owner = self.available_coupons.get(ctx.coupon_id)
        if owner != ctx.user_id:
            raise SagaStepError("Coupon invalid or not owned by user")

        del self.available_coupons[ctx.coupon_id]
        self.used[ctx.order_id] = ctx.coupon_id
        self.idem.mark_done(key)
        ctx.logs.append("Coupon used")

    def rollback_use(self, ctx: SagaContext):
        if not ctx.coupon_id:
            ctx.logs.append("No coupon rollback needed")
            return

        key = f"{ctx.saga_id}:coupon:rollback"
        if self.idem.executed(key):
            ctx.logs.append("Coupon rollback skipped by idempotency")
            return

        coupon_id = self.used.get(ctx.order_id)
        if coupon_id:
            self.available_coupons[coupon_id] = ctx.user_id
            self.used[ctx.order_id] = ""

        self.idem.mark_done(key)
        ctx.logs.append("Coupon rolled back")


class SagaOrchestrator:
    def __init__(self, order_svc, inventory_svc, account_svc, coupon_svc):
        self.order_svc = order_svc
        self.inventory_svc = inventory_svc
        self.account_svc = account_svc
        self.coupon_svc = coupon_svc

    def execute(self, ctx: SagaContext):
        completed_compensations: List[Callable[[SagaContext], None]] = []

        try:
            self.order_svc.create_pending(ctx)
            completed_compensations.append(self.order_svc.cancel)

            self.inventory_svc.reserve(ctx)
            completed_compensations.append(self.inventory_svc.release)

            self.account_svc.freeze(ctx)
            completed_compensations.append(self.account_svc.unfreeze)

            self.coupon_svc.use(ctx)
            completed_compensations.append(self.coupon_svc.rollback_use)

            self.order_svc.confirm(ctx)
            ctx.logs.append("Saga completed successfully")
            return True

        except SagaStepError as e:
            ctx.logs.append(f"Saga failed: {e}")
            for compensate in reversed(completed_compensations):
                try:
                    compensate(ctx)
                except Exception as ce:
                    ctx.logs.append(f"Compensation failed: {ce}")
            ctx.logs.append("Saga compensated")
            return False


def main():
    idem = IdempotencyStore()

    order_svc = OrderService()
    inventory_svc = InventoryService(stock={"SKU-1": 10}, idem=idem)
    account_svc = AccountService(balances={"U1001": 200}, idem=idem)
    coupon_svc = CouponService(available_coupons={"C10": "U1001"}, idem=idem)

    orchestrator = SagaOrchestrator(
        order_svc=order_svc,
        inventory_svc=inventory_svc,
        account_svc=account_svc,
        coupon_svc=coupon_svc,
    )

    ctx = SagaContext(
        saga_id=str(uuid.uuid4()),
        order_id="O20221202001",
        user_id="U1001",
        amount=120,
        items={"SKU-1": 2},
        coupon_id="C10",
    )

    success = orchestrator.execute(ctx)

    print("SUCCESS:", success)
    print("ORDER STATUS:", order_svc.orders.get(ctx.order_id))
    print("STOCK:", inventory_svc.stock)
    print("BALANCE:", account_svc.balances)
    print("COUPONS:", coupon_svc.available_coupons)
    print("\nLOGS:")
    for log in ctx.logs:
        print("-", log)


if __name__ == "__main__":
    main()

这段代码体现了什么

• 订单、库存、账户、优惠券都是本地服务
• 编排器负责流程控制
• 每一步都有补偿动作
• 使用 saga_id + step 做简单幂等
• 一旦失败，按已完成步骤逆序补偿

如何验证补偿是否生效

把余额改小一点：

account_svc = AccountService(balances={"U1001": 50}, idem=idem)

再次执行时你会看到：

• 库存先预留成功
• 账户冻结失败
• Saga 触发补偿
• 库存被释放
• 订单被取消

这就是最基本的 Saga 行为。

生产落地时，代码还需要补什么

上面这份代码只是“能跑通流程”。真上生产，至少还要补这些能力。

1. Saga 状态持久化表

一个简化表示例：

CREATE TABLE saga_instance (
    saga_id VARCHAR(64) PRIMARY KEY,
    order_id VARCHAR(64) NOT NULL,
    state VARCHAR(32) NOT NULL,
    current_step VARCHAR(64),
    retry_count INT DEFAULT 0,
    error_message TEXT,
    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL
);

步骤执行日志表：

CREATE TABLE saga_step_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    saga_id VARCHAR(64) NOT NULL,
    step_name VARCHAR(64) NOT NULL,
    action_type VARCHAR(16) NOT NULL,
    status VARCHAR(16) NOT NULL,
    idempotency_key VARCHAR(128) NOT NULL,
    error_message TEXT,
    created_at TIMESTAMP NOT NULL
);

2. 本地事务 + Outbox

订单服务里，不要这样做：

1. 更新订单状态
2. 再发 MQ 消息

因为如果更新成功、发消息失败，下游根本不知道状态变了。

应改为：

• 同一个本地事务里写订单表和 outbox 表
• 由异步任务扫描 outbox 表投递消息
• 投递成功后标记已发送

示例表：

CREATE TABLE outbox_event (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    aggregate_type VARCHAR(64) NOT NULL,
    aggregate_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(64) NOT NULL,
    payload TEXT NOT NULL,
    status VARCHAR(16) NOT NULL DEFAULT 'NEW',
    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL
);

3. 超时驱动补偿

不要假设每一步都能及时返回。生产里常见的是：

• 下游超时，但实际已经执行成功
• 消息积压导致状态延迟
• 协调器重启后流程中断

所以要有一个超时扫描任务：

• 超过阈值未推进的 Saga 标记为 TIMEOUT
• 根据步骤日志决定是继续重试还是开始补偿
• 对人工介入的单据打上告警标签

常见坑与排查

这部分我想讲得实战一点，因为很多 Saga 方案不是设计时死掉，而是上线后慢慢烂掉。

坑 1：补偿不是回滚，业务语义设计错了

有些同学会把补偿理解成数据库 rollback，这其实是错的。

比如“发放优惠券”之后，用户已经看到并使用了，这时再“删除优惠券”就可能引发新问题。
更合理的做法可能是：

• 标记优惠券失效
• 生成一张等额补偿券
• 写审计日志

排查方法：

• 检查每个补偿动作是否真的符合业务语义
• 问自己一句：如果补偿晚到 10 分钟，还成立吗？

坑 2：接口没做幂等，重试变事故

这是我见过最多的问题。协调器超时重试一次，下游执行两次，库存、余额、优惠券全乱。

排查方法：

• 检查请求里是否有全局唯一业务键
• 检查下游是否有幂等表或唯一索引
• 查日志时按 saga_id 聚合，看是否出现重复 step

坑 3：补偿顺序错了

顺序一般应该和正向动作相反。
例如先冻结余额、后核销优惠券，那么失败补偿时通常先退券、再解冻余额。

排查方法：

• 把主流程和补偿流程都画成图
• 对每个步骤标注前置依赖
• 在测试环境注入故障，验证逆序补偿是否正确

坑 4：只处理“失败”，没处理“未知”

分布式系统里最烦的不是明确失败，而是不知道成功没成功。

例如库存服务调用超时：

• 可能真没执行
• 也可能执行成功了，但响应丢了

这时不能直接补偿，更不能盲目重试。

排查方法：

• 下游提供查询接口：按 saga_id 查询步骤执行结果
• 协调器在“超时未知”时先查询，再决定重试或补偿
• 把状态从二值改成三值：成功 / 失败 / 未知

坑 5：监控只有错误日志，没有流程视角

订单出问题时，如果你只能看到“库存服务 500”，其实定位效率很低。
真正有用的是：这一笔订单的 Saga 当前处在哪一步，做过哪些补偿，重试了几次。

排查方法：

• 为每个 Saga 生成全链路 traceId
• 日志字段统一：saga_id、order_id、step_name、action_type
• 做一个简单的 Saga 看板，比堆日志好用太多

安全/性能最佳实践

Saga 设计里，大家容易只盯一致性，忽略安全和性能。实际上订单链路是高价值链路，这两点同样重要。

安全最佳实践

1. 补偿接口不能裸奔

补偿接口本质上拥有“撤销资产动作”的权限，比如解冻余额、归还库存、恢复优惠券。
所以一定要做：

• 服务间身份认证
• 请求签名或 mTLS
• 细粒度权限校验
• 防重放机制

不要让外部系统或者误调用直接触发补偿。

2. 审计日志要完整

尤其涉及余额、库存、券这些资产变动时，要记录：

• 请求方
• 幂等键
• 前后状态
• 结果码
• 时间戳

这不仅是排障需要，也是审计需要。

3. 避免敏感数据在事件里裸传

事件消息里尽量不要直接塞：

• 完整支付信息
• 用户敏感身份数据
• 明文手机号等

可以传业务主键，下游按权限再查。

性能最佳实践

1. 缩短主链路，只保留必要步骤

我见过一些团队把“下单后发站内信”“打埋点”“推推荐系统”也放进 Saga 主流程。结果就是：

• 链路长
• 失败点多
• 延迟高

建议做法：

• 主链路只保留“订单成立必须依赖”的步骤
• 其他非关键动作走异步事件

2. 步骤接口要轻量，避免大事务

比如库存服务的预留接口，只做：

• 校验库存
• 扣减可售/增加预留
• 写执行日志

不要顺手再查一堆扩展信息，更不要把重计算塞进去。

3. 重试要有退避策略

瞬时故障时，重试是有用的；但如果所有协调器同时对一个下游服务猛烈重试，就会形成放大效应。

建议：

• 指数退避
• 随机抖动
• 最大重试次数
• 熔断后人工介入

4. 容量估算别忽略补偿流量

很多人压测只看主交易流量，不看补偿流量。
但在下游抖动时，补偿可能瞬间把请求量再翻一倍。

一个简单估算：

• 正常下单 TPS：1000
• 某一步失败率：3%
• 每笔失败平均触发 2 次补偿调用

那么补偿流量约为：

1000 × 3% × 2 = 60 次/秒

如果叠加重试，实际峰值会更高。
因此库存、账户这类服务必须评估：

• 正向请求峰值
• 补偿请求峰值
• 重试叠加峰值

一套更务实的落地建议

如果你正在设计订单一致性方案，我建议按下面顺序推进，而不是一口气搞一个“大而全”的平台。

第一阶段：先把链路跑通

目标：

• 核心订单主链路使用编排式 Saga
• 每个步骤有明确补偿
• 接口具备幂等能力
• 可人工排查和人工重放

第二阶段：补齐工程能力

目标：

• Saga 状态持久化
• Outbox 落地
• 超时扫描与重试
• 统一日志与监控面板

第三阶段：平台化治理

目标：

• 通用编排框架
• 统一步骤模板
• 告警分级
• 死信处理
• 人工干预后台

很多团队的问题不是“不懂 Saga”，而是一上来就想做平台，结果业务规则没吃透。我的建议是：先用 1~2 条关键链路把补偿语义打磨稳定，再考虑抽象。

总结

订单服务的一致性，真正难的不是“选一个模式”，而是把业务动作、失败路径、补偿语义、幂等机制、观测能力一起设计完整。Saga 之所以在分布式订单场景里常用，不是因为它最完美，而是因为它在一致性、性能、复杂度之间做了相对合理的平衡。

如果你只记住几个最关键的落地建议，我建议是这几条：

1. 核心交易链路优先用编排式 Saga
2. 每个步骤都要有业务上成立的补偿动作
3. 幂等是底线，不是优化项
4. 状态必须持久化，不能只放内存
5. Outbox、超时扫描、死信处理要作为标配
6. 把主链路做短，把非关键动作异步化
7. 监控要围绕 saga_id 和订单维度展开

最后补一句边界条件：
如果你的业务不能接受短暂不一致、且补偿成本极高，那么不要盲目上 Saga，应该重新审视是否需要更强的一致性机制，或者收缩服务边界，把关键操作合并到单服务内处理。

Saga 不是“分布式事务的平替”，它更像一种工程化妥协：承认分布式系统会失败，然后设计出一套可恢复的业务机制。
对于订单系统来说，这种妥协，往往比追求表面上的绝对完美更现实。