背景与问题

订单系统几乎是分布式业务里最容易“表面正常、暗地翻车”的地方。

一个用户点击“下单”之后，背后通常不止一个动作：

创建订单
扣减库存
冻结或扣减优惠券
发起支付
更新营销资格
发送短信或站内通知

如果这些动作都发生在一个单体数据库事务里，事情反而简单。但一旦拆成多个服务，问题就会立刻冒出来：

请求重试导致重复下单
消息重复投递导致库存被扣两次
消费者处理成功，但 ack 失败，消息再次消费
数据库提交成功，但消息发送失败，订单状态卡住
某个下游服务超时，最终状态不一致

我第一次做订单系统时，最容易误判的一点是：消息队列并不会自动帮你实现一致性。
它解决的是“解耦、削峰、异步”，不是“天然强一致”。真正兜底的一般是两件事：

幂等设计
围绕消息投递与消费链路构建可追踪、可补偿的一致性机制

这篇文章我会从工程实战角度，把一套比较常用、也比较靠谱的方案串起来：
订单本地事务 + Outbox 消息表 + MQ 异步投递 + 消费端幂等处理 + 状态机推进

背景场景：一个典型订单链路

假设订单服务收到“创建订单”请求后，要做这几件事：

写入订单主表
写入订单明细
记录一条“订单已创建”的领域事件
由异步消费者去扣减库存
库存成功后，推进订单状态到 CONFIRMED
库存失败，则推进订单状态到 FAILED

这时最关键的问题不是“如何发消息”，而是：

如何保证订单不会被重复创建？
如何保证“订单创建成功”这件事一定能被发出去？
如何保证库存消费者不会重复扣减？
如何保证消息乱序、重复、延迟时，订单状态不被写坏？

核心原理

这一类问题，我建议拆成四个层次理解。

1. 请求入口幂等：防止重复下单

用户、网关、调用方都可能重试。
所以创建订单接口必须支持一个业务幂等键，比如：

request_id
client_token
biz_order_no

这个键必须由调用方生成，并在服务端落库做唯一约束。

设计原则

幂等键必须与“同一次业务意图”绑定
唯一索引必须落在数据库，而不是只放 Redis
如果请求重复，返回第一次处理结果，而不是简单报错

2. 本地事务 + Outbox：防止“数据库成功但消息丢了”

经典问题：

先发 MQ，再写库：消息发出去了，但数据库回滚了
先写库，再发 MQ：数据库成功了，但 MQ 发送失败

这就是著名的“双写不一致”。

一个非常实用的方案是 Transactional Outbox：

在同一个数据库事务里：
- 写订单表
- 写 outbox_event 事件表
事务提交后，由独立投递程序扫描 outbox 表并发送到 MQ
发送成功后，把事件标记为 SENT

这样即使 MQ 短暂不可用，事件也还在库里，后续可以补发。

3. 消费端幂等：防止重复扣库存

大部分 MQ 都只能承诺 至少一次投递，而不是“恰好一次”。

这意味着消费者必须假设：

同一条消息会被重复消费
ack 失败会重投
消费成功但进程挂了，也可能重投

所以库存服务不能写成“来一条就扣一次”，而要写成：

先检查 message_id 或 biz_key 是否处理过
没处理过再执行业务
在同一事务里记录“已处理”

4. 状态机约束：防止乱序更新

订单状态如果只是字符串字段任意改，非常容易乱。

建议把订单状态设计成有限状态机，例如：

INIT
CREATED
CONFIRMED
FAILED
CANCELLED

并明确哪些状态迁移是合法的：

INIT -> CREATED
CREATED -> CONFIRMED
CREATED -> FAILED
CONFIRMED -> CANCELLED（某些业务允许）
FAILED 之后不能再到 CONFIRMED

这样即使重复消息、延迟消息来了，也能通过条件更新避免状态回滚。

整体架构图

flowchart LR
    A[客户端提交下单请求] --> B[订单服务]
    B --> C[(orders表)]
    B --> D[(outbox_event表)]
    D --> E[Outbox投递器]
    E --> F[MQ]
    F --> G[库存服务消费者]
    G --> H[(inventory_txn表)]
    G --> I[(inventory表)]
    G --> J[发送库存结果事件]
    J --> F
    F --> K[订单结果消费者]
    K --> C

时序图：成功路径

sequenceDiagram
    participant Client as 客户端
    participant Order as 订单服务
    participant DB as 数据库
    participant Outbox as Outbox投递器
    participant MQ as 消息队列
    participant Inv as 库存服务

    Client->>Order: createOrder(requestId, items)
    Order->>DB: 事务写 orders + outbox_event
    DB-->>Order: commit success
    Order-->>Client: 返回订单已创建

    Outbox->>DB: 扫描未发送事件
    Outbox->>MQ: 投递 OrderCreated
    MQ-->>Inv: 推送消息
    Inv->>DB: 幂等检查 + 扣库存 + 记录处理
    Inv->>MQ: 发送 InventoryReserved
    MQ-->>Order: 推送库存成功事件
    Order->>DB: 更新订单状态 CREATED -> CONFIRMED

方案对比与取舍分析

在订单一致性设计里，常见有三条路。

方案一：分布式事务（2PC / XA）

优点：

理论上强一致
业务代码看起来更“线性”

缺点：

对中间件和数据库要求高
性能差
可用性差
很多云原生系统并不适合

适用场景：

核心链路极短
参与方少
基础设施成熟

方案二：TCC

优点：

一致性强于纯异步最终一致
适合资金、库存冻结等场景

缺点：

业务侵入大
Try/Confirm/Cancel 开发复杂
悬挂、空回滚、防重都要自己处理

适用场景：

高价值交易
可以明确定义预留资源

方案三：本地事务 + MQ + 幂等 + 补偿

优点：

工程上最常见
易于扩展
解耦好
对基础设施要求相对低

缺点：

不是强一致，而是最终一致
需要建设对账、监控、补偿机制

对订单类业务来说，大多数互联网场景我更推荐第三种。
不是因为它最“高级”，而是因为它在复杂性、可用性、吞吐和实施成本之间更平衡。

数据模型设计

下面给一套简化但实战可用的表结构。

订单表

CREATE TABLE orders (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(64) NOT NULL UNIQUE,
    request_id VARCHAR(64) NOT NULL UNIQUE,
    user_id BIGINT NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status VARCHAR(32) NOT NULL,
    version INT NOT NULL DEFAULT 0,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

Outbox 事件表

CREATE TABLE outbox_event (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    event_id VARCHAR(64) NOT NULL UNIQUE,
    aggregate_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(64) NOT NULL,
    payload TEXT NOT NULL,
    status VARCHAR(16) NOT NULL DEFAULT 'NEW',
    retry_count INT NOT NULL DEFAULT 0,
    next_retry_at TIMESTAMP NULL DEFAULT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

消费幂等记录表

CREATE TABLE message_consume_log (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    consumer_name VARCHAR(64) NOT NULL,
    message_id VARCHAR(64) NOT NULL,
    biz_key VARCHAR(64) NOT NULL,
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_consumer_message (consumer_name, message_id),
    UNIQUE KEY uk_consumer_biz (consumer_name, biz_key)
);

库存表

CREATE TABLE inventory (
    sku_id BIGINT PRIMARY KEY,
    available INT NOT NULL,
    reserved INT NOT NULL DEFAULT 0,
    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

实战代码（可运行）

下面用 Python + SQLite 模拟一套最小可运行示例。
这个示例重点演示：

下单接口幂等
订单与 outbox 同事务落库
outbox 投递
消费端幂等
重复消息不会重复扣库存

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

import sqlite3
import json
import uuid
import time
from contextlib import contextmanager

DB_FILE = "order_demo.db"


@contextmanager
def get_conn():
    conn = sqlite3.connect(DB_FILE)
    conn.isolation_level = None
    try:
        yield conn
    finally:
        conn.close()


def init_db():
    with get_conn() as conn:
        cur = conn.cursor()
        cur.executescript("""
        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            order_no TEXT NOT NULL UNIQUE,
            request_id TEXT NOT NULL UNIQUE,
            user_id INTEGER NOT NULL,
            sku_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL,
            status TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS outbox_event (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            event_id TEXT NOT NULL UNIQUE,
            aggregate_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            payload TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'NEW',
            retry_count INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS inventory (
            sku_id INTEGER PRIMARY KEY,
            available INTEGER NOT NULL
        );

        CREATE TABLE IF NOT EXISTS message_consume_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            consumer_name TEXT NOT NULL,
            message_id TEXT NOT NULL,
            biz_key TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE(consumer_name, message_id),
            UNIQUE(consumer_name, biz_key)
        );
        """)
        conn.commit()

        cur.execute("INSERT OR IGNORE INTO inventory(sku_id, available) VALUES(?, ?)", (1001, 10))
        conn.commit()


def begin(conn):
    conn.execute("BEGIN")


def commit(conn):
    conn.execute("COMMIT")


def rollback(conn):
    conn.execute("ROLLBACK")


def create_order(request_id, user_id, sku_id, quantity):
    with get_conn() as conn:
        cur = conn.cursor()
        try:
            begin(conn)

            # 幂等检查：request_id 唯一
            cur.execute("SELECT order_no, status FROM orders WHERE request_id = ?", (request_id,))
            row = cur.fetchone()
            if row:
                commit(conn)
                return {"ok": True, "duplicate": True, "order_no": row[0], "status": row[1]}

            order_no = "ORD-" + uuid.uuid4().hex[:12].upper()
            cur.execute("""
                INSERT INTO orders(order_no, request_id, user_id, sku_id, quantity, status)
                VALUES (?, ?, ?, ?, ?, ?)
            """, (order_no, request_id, user_id, sku_id, quantity, "CREATED"))

            event = {
                "message_id": str(uuid.uuid4()),
                "order_no": order_no,
                "user_id": user_id,
                "sku_id": sku_id,
                "quantity": quantity
            }

            cur.execute("""
                INSERT INTO outbox_event(event_id, aggregate_id, event_type, payload, status)
                VALUES (?, ?, ?, ?, ?)
            """, (event["message_id"], order_no, "OrderCreated", json.dumps(event), "NEW"))

            commit(conn)
            return {"ok": True, "duplicate": False, "order_no": order_no, "status": "CREATED"}
        except Exception as e:
            rollback(conn)
            return {"ok": False, "error": str(e)}


def dispatch_outbox():
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute("""
            SELECT id, event_id, event_type, payload
            FROM outbox_event
            WHERE status = 'NEW'
            ORDER BY id
        """)
        rows = cur.fetchall()

        mq = []
        for row in rows:
            event_db_id, event_id, event_type, payload = row
            mq.append({
                "message_id": event_id,
                "event_type": event_type,
                "payload": json.loads(payload)
            })
            cur.execute("UPDATE outbox_event SET status = 'SENT' WHERE id = ?", (event_db_id,))
        conn.commit()
        return mq


def consume_order_created(message):
    consumer_name = "inventory-consumer"
    payload = message["payload"]
    message_id = message["message_id"]
    order_no = payload["order_no"]
    sku_id = payload["sku_id"]
    quantity = payload["quantity"]

    with get_conn() as conn:
        cur = conn.cursor()
        try:
            begin(conn)

            # 消费幂等
            cur.execute("""
                SELECT id FROM message_consume_log
                WHERE consumer_name = ? AND message_id = ?
            """, (consumer_name, message_id))
            if cur.fetchone():
                commit(conn)
                return {"ok": True, "duplicate": True, "order_no": order_no}

            # 检查库存
            cur.execute("SELECT available FROM inventory WHERE sku_id = ?", (sku_id,))
            row = cur.fetchone()
            if not row:
                raise Exception(f"sku {sku_id} not found")

            available = row[0]
            if available < quantity:
                cur.execute("UPDATE orders SET status = 'FAILED' WHERE order_no = ? AND status = 'CREATED'", (order_no,))
            else:
                cur.execute("UPDATE inventory SET available = available - ? WHERE sku_id = ?", (quantity, sku_id))
                cur.execute("UPDATE orders SET status = 'CONFIRMED' WHERE order_no = ? AND status = 'CREATED'", (order_no,))

            # 记录已消费
            cur.execute("""
                INSERT INTO message_consume_log(consumer_name, message_id, biz_key)
                VALUES (?, ?, ?)
            """, (consumer_name, message_id, order_no))

            commit(conn)
            return {"ok": True, "duplicate": False, "order_no": order_no}
        except Exception as e:
            rollback(conn)
            return {"ok": False, "error": str(e)}


def print_state():
    with get_conn() as conn:
        cur = conn.cursor()
        print("\n=== orders ===")
        for row in cur.execute("SELECT order_no, request_id, sku_id, quantity, status FROM orders ORDER BY id"):
            print(row)

        print("\n=== inventory ===")
        for row in cur.execute("SELECT sku_id, available FROM inventory ORDER BY sku_id"):
            print(row)

        print("\n=== outbox_event ===")
        for row in cur.execute("SELECT event_id, event_type, status FROM outbox_event ORDER BY id"):
            print(row)

        print("\n=== message_consume_log ===")
        for row in cur.execute("SELECT consumer_name, message_id, biz_key FROM message_consume_log ORDER BY id"):
            print(row)


if __name__ == "__main__":
    init_db()

    # 模拟用户重复点击下单
    req_id = "REQ-001"
    print(create_order(req_id, user_id=1, sku_id=1001, quantity=2))
    print(create_order(req_id, user_id=1, sku_id=1001, quantity=2))  # 幂等返回

    # outbox 投递
    mq_messages = dispatch_outbox()
    print("\nDispatch to MQ:", mq_messages)

    # 模拟 MQ 重复投递同一条消息两次
    for msg in mq_messages:
        print(consume_order_created(msg))
        print(consume_order_created(msg))  # 重复消费

    print_state()

运行效果说明

这段程序运行后，你会看到几个关键结果：

同一个 request_id 第二次创建订单，不会生成新订单
outbox_event 在订单创建时和订单数据一起提交
同一条消息即使被消费两次，也只会真正扣一次库存
订单状态从 CREATED 变成 CONFIRMED，且不会重复推进

这就是一条最小但完整的一致性闭环。

状态机设计建议

实际项目里，我建议把订单状态迁移收口到一个统一方法，而不是分散在多个服务里随意更新。

stateDiagram-v2
    [*] --> INIT
    INIT --> CREATED
    CREATED --> CONFIRMED
    CREATED --> FAILED
    CREATED --> CANCELLED
    CONFIRMED --> CANCELLED
    FAILED --> [*]
    CANCELLED --> [*]

如果用 SQL 更新状态，尽量带上旧状态条件：

UPDATE orders
SET status = 'CONFIRMED', version = version + 1
WHERE order_no = ? AND status = 'CREATED';

这样做有两个好处：

防止乱序消息把状态改回去
利用受影响行数判断是否真的完成了状态迁移

容量估算与架构边界

中级工程师做方案设计时，不能只停留在“能不能工作”，还要问一句：流量上来后还能不能工作。

1. Outbox 扫描压力

如果每秒订单创建 3000 笔，那么 outbox 事件也差不多是 3000/s。

你要估算：

扫描频率是多少
单次扫描多少条
是否有 status + created_at 索引
是否需要分片表

常见做法：

每 100ms 扫描一次
每次拉 500~2000 条
多 worker 并发发送
按 id 范围或时间窗口分批

2. 消费幂等表膨胀

message_consume_log 会增长很快。
如果不做归档，几个月后索引会明显变重。

建议：

只保留 7~30 天
按月分表或按时间分区
对历史数据定期归档

3. 热点库存行竞争

爆款商品会出现热点行更新：

UPDATE inventory SET available = available - 1 WHERE sku_id = 1001;

这时会遇到锁竞争、RT 上升、吞吐下降。

应对思路：

预扣库存分桶
Redis 预减 + DB 异步对账
队列按 sku_id 分区，保证同商品串行消费
库存中心独立出来，不让多个服务直接写

常见坑与排查

这一节我尽量写得“接地气”一点，因为这些坑多数不是理论问题，而是线上告警把人叫醒的问题。

坑一：只做接口幂等，没做消费幂等

现象：

下单只生成一个订单
但库存被扣了两次

原因：

开发者以为请求入口挡住重复就够了
实际 MQ 是至少一次投递，消费者一定要防重

排查方法：

查 MQ 消费日志，看同一个 message_id 是否出现多次
查消费幂等表是否缺失
查 ack 是否在业务提交前发送

坑二：先发消息后落库

现象：

下游已经收到“订单创建成功”
但订单表里查不到该订单

原因：

消息先发，数据库事务后失败
典型双写不一致

排查方法：

对比订单表与消息轨迹
看业务日志中是否存在“发送成功但事务回滚”

止血方案：

改成 outbox 模式
短期通过对账任务补偿脏数据

坑三：消费成功了，但幂等记录没写进去

现象：

第一次消费已经扣了库存
第二次重试又扣一次

原因：

“扣库存”和“写已消费记录”不在同一事务
先执行业务，后写日志，中间失败了

正确做法：

业务处理和幂等记录必须在一个本地事务里提交

坑四：状态更新没带前置条件

现象：

订单本来已经 FAILED
延迟消息又把它改成 CONFIRMED

原因：

更新 SQL 写成： UPDATE orders SET status='CONFIRMED' WHERE order_no=?
没有限制旧状态

正确做法：

必须写成： WHERE order_no=? AND status='CREATED'

坑五：幂等键设计错误

现象：

用户重试时没有命中幂等
或者不同订单被错误判成同一单

原因：

幂等键不是业务唯一意图
比如只用 user_id + sku_id，导致连续购买同商品被误判重复

建议：

用调用方生成的 request_id
生命周期要覆盖重试窗口
具备全局唯一性

排查路径：出了不一致先看什么

如果你线上遇到“订单状态不对”这类问题，我建议按这条链路排查：

请求日志
是否存在重复请求？是否 request_id 相同？
订单表
订单创建成功了吗？状态是什么？
outbox_event 表
事件是否写入？状态是 NEW、SENT 还是卡住？
MQ 轨迹
消息是否投递？投递了几次？是否死信？
消费幂等表
同一消息是否已处理？处理了几次？
下游业务表
比如库存是否真的扣减？是否有回滚或补偿记录？
状态机迁移日志
哪条消息把状态推进到了当前值？

这个顺序有个好处：
能快速定位问题落在“入口、投递、消费、状态推进”哪个环节，而不是一上来就全链路瞎翻。

安全/性能最佳实践

这部分我分成两类说。

安全最佳实践

1. 幂等键不要直接信任客户端格式

客户端传 request_id 很正常，但服务端要校验：

长度
字符合法性
是否为空
是否存在异常重复模式

否则有人恶意构造超长字符串，可能把索引和日志打爆。

2. 消息体要做签名或最少做字段校验

尤其在跨团队、跨系统场景下，消息不能“拿到就信”。

建议至少校验：

message_id
event_type
timestamp
biz_key
必填字段完整性

3. 关键操作保留审计日志

订单状态推进、库存扣减、补偿执行，都要有审计日志。
这不只是为了排障，也是为了风控和合规。

性能最佳实践

1. 唯一索引是幂等的底线，但不要让每次请求都走多次查询

更好的处理方式通常是：

尝试插入
命中唯一键冲突后再查结果返回

比“先查再插”更抗并发。

2. Outbox 投递要批量化

不要一条一条扫、一条一条发。
高吞吐下建议：

批量拉取
批量发送
批量更新状态

3. 消费者以业务键分区

例如按 order_no、sku_id 或 user_id 分区，可以减少并发写冲突。
尤其库存类场景，按 sku_id 分区通常很有效。

4. 死信队列一定要接

死信队列不是“有了再看”，而是要一开始就接进监控：

堆积数量
首次出现时间
重试次数
消息类型分布

5. 建对账与补偿任务

只靠在线链路，不足以覆盖所有极端异常。
建议每天或每小时跑这些对账：

订单表 vs outbox 表
outbox 表 vs MQ 投递结果
订单状态 vs 库存事务结果
支付状态 vs 订单最终状态

一套实战落地清单

如果你准备在团队里落地这套方案，我建议至少完成下面这些项：

这份清单不花哨，但很有用。很多系统出事，不是因为没上“高级架构”，而是这些基础动作漏了两三个。

总结

订单系统一致性，真正有效的不是某一个“银弹组件”，而是一整套协作机制：

入口幂等：防重复下单
Outbox 模式：防数据库与消息双写不一致
消费端幂等：防重复消费导致副作用放大
状态机约束：防乱序和回写污染
对账补偿：为最终一致性兜底

如果你的业务不是金融级强一致，大多数场景下：

本地事务 + Outbox + MQ + 消费幂等 + 状态机
是一套性价比很高、并且能长期演进的方案。

最后给几个可执行建议：

先把幂等落到数据库唯一约束上，不要只靠 Redis。
消息发送一定走 outbox，不要手写“先写库再发 MQ”侥幸过关。
消费者永远按重复投递来设计，别假设 MQ 只来一次。
状态更新必须条件化，别让延迟消息把正确状态覆盖掉。
一定建设对账和补偿，因为最终一致不是“最终一定一致”，而是“最终可被修复到一致”。

边界条件也要说清楚：
如果你做的是资金实时扣款、证券成交、强监管账务，光靠这套最终一致架构通常还不够，可能要进一步引入 TCC、账务分录体系，甚至更严格的事务模型。

但如果你面对的是大多数电商、零售、履约型订单系统，这套方法已经足够实用，而且经得住线上流量和故障场景的考验。