如何处理可能乱序的 Domain (integration) events？

2026-10-24支付老李

结论

不要按照事件到达的顺序直接覆盖 Order.Status。

Ordering BC 需要限制状态只能合法迁移，但更好的建模方式是，不要把”已付款”和”已发货”压缩到一个二选一的字段里。这是两个独立且存在因果关系的事实，应该分别记录：

PaymentStatus = Processed
ShippingStatus = Shipped

如果业务仍然需要统一的 Order.Status，应从这些事实推导，而不是让每个 integration event 直接为它赋值。这样，即使 PaymentProcessed 晚于 OrderShipped 到达，也不会把已经发货的订单降级为 Payed。

问题的根源

下面的事件处理方式假定消息会严格按照业务发生的顺序到达：

void Handle(PaymentProcessed @event)
{
    order.Status = OrderStatus.Payed;
}

void Handle(OrderShipped @event)
{
    order.Status = OrderStatus.Shipped;
}

这个假定在异步系统中并不可靠。消息可能因为重试、不同的 topic 或 queue、消费者并发、网络延迟等原因乱序到达。即使消息系统提供顺序保证，通常也只限于特定的 partition、session 或 message group，并不会自动保证跨 BC、跨消息流的全局顺序。

因此，不能用事件的到达顺序判断业务事实的发生顺序。

首选方案：分别保存独立事实

“付款完成”和”订单已发货”不是同一个维度的状态。可以把模型改成：

public enum PaymentStatus
{
    Pending,
    Processed,
    Failed
}

public enum ShippingStatus
{
    NotReady,
    Ready,
    Shipped
}

public sealed class Order
{
    public Guid Id { get; private set; }

    public PaymentStatus PaymentStatus { get; private set; }
    public ShippingStatus ShippingStatus { get; private set; }

    public OrderStatus Status =>
        ShippingStatus == ShippingStatus.Shipped
            ? OrderStatus.Shipped
            : PaymentStatus == PaymentStatus.Processed
                ? OrderStatus.Payed
                : OrderStatus.Pending;

    public void ApplyPaymentProcessed()
    {
        PaymentStatus = PaymentStatus.Processed;
    }

    public void ApplyOrderShipped()
    {
        ShippingStatus = ShippingStatus.Shipped;
    }
}

即使按下面的顺序处理事件：

OrderShipped
PaymentProcessed

最终结果仍然是：

PaymentStatus  = Processed
ShippingStatus = Shipped
Status         = Shipped

模型保存的是已经确认发生的事实。后到的付款事件只会补充付款信息，不会撤销已经发生的发货事实。

实际项目中，Payed 通常应拼写为 Paid。如果现有 API 或数据库枚举已经使用 Payed，重命名时需要考虑兼容性。

状态机仍然有必要

如果 Order.Status 确实表示领域模型中的单一生命周期状态，那么使用状态机或显式的状态迁移规则是合理的。处理事件时应拒绝状态降级，而不是无条件覆盖：

public sealed class Order
{
    public OrderStatus Status { get; private set; }

    public void Apply(PaymentProcessed @event)
    {
        switch (Status)
        {
            case OrderStatus.Pending:
                Status = OrderStatus.Payed;
                break;

            case OrderStatus.Payed:
            case OrderStatus.Shipped:
                // 已处理，或者当前状态已经更靠后。
                break;

            default:
                throw new InvalidOperationException(
                    $"Cannot apply PaymentProcessed when order is {Status}.");
        }
    }

    public void Apply(OrderShipped @event)
    {
        switch (Status)
        {
            case OrderStatus.Pending:
            case OrderStatus.Payed:
                Status = OrderStatus.Shipped;
                break;

            case OrderStatus.Shipped:
                // 幂等处理。
                break;

            default:
                throw new InvalidOperationException(
                    $"Cannot apply OrderShipped when order is {Status}.");
        }
    }
}

这段代码包含一个需要明确的业务决定：是否允许尚未观察到 PaymentProcessed 的订单进入 Shipped。

如果 OrderShipped 是 Shipping BC 已确认发生的事实，Ordering BC 通常不应因为本地尚未收到付款事件而否认它。此时可以接受 Pending -> Shipped，再由迟到的 PaymentProcessed 补齐付款信息。

如果业务规定”未确认付款绝不能发货”，这一约束主要应由负责启动发货流程的 BC 或 process manager 保证，不能只依赖 Ordering BC 的投影状态。

不要仅依赖事件时间戳

为事件增加 OccurredAt 有助于审计，但不能单独解决事件顺序问题：

{
  "eventId": "2fa3...",
  "orderId": "8471...",
  "occurredAt": "2026-09-01T08:30:00Z",
  "eventType": "PaymentProcessed"
}

不同服务的时钟可能存在偏差，事件的时间精度也可能不够。消费者不能简单地认为时间戳更大的事件一定对应更高的业务状态。

如果多个事件来自同一个有序事件流，可以使用生产者分配的单调递增版本号：

{
  "eventId": "2fa3...",
  "aggregateId": "8471...",
  "aggregateVersion": 12,
  "eventType": "PaymentProcessed"
}

消费者可以记录最后处理的版本：

if (@event.AggregateVersion <= checkpoint.LastProcessedVersion)
{
    return;
}

if (@event.AggregateVersion != checkpoint.LastProcessedVersion + 1)
{
    await StoreForLater(@event);
    return;
}

await Apply(@event);
checkpoint.AdvanceTo(@event.AggregateVersion);

但当 PaymentProcessed 和 OrderShipped 分别由不同 BC 发布时，通常没有共同的 aggregateVersion，两个 BC 各自的版本号也不能直接比较。跨 BC 的业务顺序需要通过明确的因果流程、独立事实建模或专门的协调器来表达。

用流程协调表达因果关系

如果 Shipping 必须等 Payment 完成后才能开始，可以使用 saga 或 process manager：

PaymentProcessed
        │
        ▼
FulfillmentProcess
        │  Send StartShipping command
        ▼
Shipping BC
        │
        ▼
OrderShipped

具体流程如下：

Payment BC 完成付款，并可靠地发布 PaymentProcessed。
process manager 消费该事件，然后发送 StartShipping command。
Shipping BC 校验命令并执行发货。
Shipping BC 发布 OrderShipped。
Ordering BC 分别记录付款和发货事实。

这个流程建立了业务上的因果顺序，但 Ordering BC 仍不一定会按相同顺序收到两个事件，因为它们可能经过不同的消息通道。因此，Ordering BC 依然需要支持幂等和乱序处理。

每个消费者都要实现幂等

乱序往往伴随着重复投递。integration event 应携带稳定的 EventId，消费者则应在同一个本地事务中记录处理结果和事件 ID：

public async Task Handle(PaymentProcessed @event)
{
    if (await inbox.Contains(@event.EventId))
    {
        return;
    }

    var order = await orders.Get(@event.OrderId);

    order.ApplyPaymentProcessed();

    await inbox.MarkAsProcessed(@event.EventId);
    await unitOfWork.Commit();
}

如果 MarkAsProcessed 与业务数据更新不在同一个事务中，消费者崩溃时仍可能造成重复更新。常见的处理方式包括：

生产端使用 transactional outbox，保证业务提交与事件发布不会丢失。
消费端使用 inbox 或 processed-message table，避免重复事件产生重复副作用。
更新数据时使用 optimistic concurrency，防止多个消费者同时覆盖同一订单。
对无法立即处理的事件执行重试、暂存或死信处理，同时保留诊断信息。

区分领域状态和查询投影

还需要确认 Order.Status 属于哪类数据。

如果它决定 Ordering BC 内部能否取消订单、修改地址或退款，那么它就是领域状态，必须通过 aggregate 的方法和不变量更新，事件处理器不能直接修改字段。

如果它只是订单列表中的展示字段，就更接近 read model。read model 可以使用可交换、单调的更新方式：

UPDATE order_view
SET payment_processed = TRUE
WHERE order_id = @order_id;

UPDATE order_view
SET shipped = TRUE
WHERE order_id = @order_id;

查询时再计算展示状态：

CASE
    WHEN shipped = TRUE THEN 'Shipped'
    WHEN payment_processed = TRUE THEN 'Payed'
    ELSE 'Pending'
END

这种更新能很好地容忍乱序。无论两个事件以什么顺序执行，最终结果都相同。

实施时的判断原则

处理这类问题时，可以按以下顺序判断：

确认事件表达的是可以撤销的命令，还是已经发生且不能否认的事实。
分别保存付款、发货等不同维度，避免多个事件争抢同一个字段。
如果必须保留统一状态，就从事实推导，或者使用合法且单调的状态迁移。
只有确实需要建立跨 BC 的因果关系时，才引入 saga 或 process manager。
使用 outbox、inbox、EventId 和 optimistic concurrency 处理消息丢失、重复投递及并发更新。
只有当事件属于同一个有序流时，才使用统一的 sequence 或 version 检测缺口。
对长期缺失的前置事件设置告警和 reconciliation，不要无限等待。

状态机可以防止 Payed 覆盖 Shipped，但它只是最后一道领域保护。更稳妥的模型是分别保存独立事实，让事件处理具备幂等、可交换和单调更新的性质，再把统一的订单状态作为这些事实的派生结果。