PAYATHON 2026

如何正确配置 MassTransit Order state machine？

支付老李

结论

问题的根源不是 MassTransit 无法识别状态，而是事件与状态的含义错位了：

  • StartOrderProcess 一到达，Saga 就被直接切换到了 Validated。
  • 随后验证服务发布 OrderValidated，但 Validated 状态下只配置了 CheckoutSessionCreatedEvent，没有配置 OrderValidated。
  • 因此 MassTransit 正确地抛出了 Not accepted in state Validated。

正确做法是区分：

  1. 当前已经完成的业务状态；
  2. 发给下游服务的命令；
  3. 下游服务返回的结果事件。

例如，订单启动后应进入 Created，同时发布 ValidateOrder 命令。收到 OrderValidated 后才进入 Validated，然后再发布创建支付会话的命令。

另外，当前取消逻辑调用的是 .Finalize()，它会进入 MassTransit 内置的 Final 状态，并不会进入声明的 Cancelled 状态。若要保留 Cancelled，应使用 .TransitionTo(Cancelled)。

当前配置中的主要问题

1. 过早进入 Validated

当前启动逻辑是：

Initially(
    When(StartOrderProcess)
        .Publish(context => new OrderValidatedEvent(context.Instance))
        .TransitionTo(Validated)
);

这段代码把 OrderValidatedEvent 当成了“请求验证订单”的消息，同时又把 Saga 标记为已经验证完成。

但从名字看，OrderValidatedEvent 表示“订单已经验证完成”，属于结果事件，不应该在验证开始前发布。

推荐将消息分为两类：

ValidateOrder            // 命令：请验证订单
OrderValidated           // 事件：订单已经验证成功
CreateCheckoutSession    // 命令：请创建支付会话
CheckoutSessionCreated   // 事件：支付会话已经创建

命令表示希望另一个服务执行某个动作，事件表示该动作已经完成。

2. OrderValidated 没有对应的处理器

错误信息已经明确说明：

The OrderValidated event is not handled during the Validated state

状态机只会处理当前状态下通过 When(...) 注册的事件。

当前配置中：

During(
    Validated,
    When(CheckoutSessionCreatedEvent)
        // ...
);

因此，当 Saga 已经处于 Validated 时，只有 CheckoutSessionCreatedEvent 能被处理，OrderValidated 会被拒绝。

简单地在 Validated 中增加 When(OrderValidated) 虽然可以消除异常，但会让业务含义变成“已经验证完成的订单再次处理验证完成事件”。更合理的修复是调整前一个状态，使 OrderValidated 在 Created 状态中处理。

3. 发布的消息与触发转换的消息混淆

类似问题在后续步骤中也存在：

When(OrderPayed)
    .Publish(context => new OrderPayedEvent(context.Instance))

收到 OrderPayed 后又发布一个 OrderPayedEvent，如果两者实际映射到相同或相互订阅的消息契约，可能造成重复消费甚至消息循环。

收到完成事件以后，通常应该发布下一步骤的命令。例如：

收到 OrderPaid
    -> 进入 Paid
    -> 发布 CollectOrder

而不是再次发布 OrderPaid。

4. Cancelled 状态从未使用

当前代码声明了：

public State Cancelled { get; private set; }

但取消行为是：

DuringAny(
    When(OrderCancelled)
        .Finalize()
);

.Finalize() 会进入框架内置的 Final 状态。它不会把 CurrentState 保持为 Cancelled。

同时：

SetCompletedWhenFinalized();

表示进入 Final 后 Saga 被视为完成。根据使用的 Saga repository，完成实例通常会被删除或不再作为活动 Saga 保留。

因此需要在以下两种方案中选择：

  • 希望数据库中保留 Cancelled 状态：使用 .TransitionTo(Cancelled)，不要立即 Finalize()。
  • 不需要保留已取消实例：记录取消信息后直接 .Finalize()，并删除没有实际用途的 Cancelled 声明。

推荐的状态转换

如果状态名称表示“最后一个已经完成的业务步骤”，流程可以配置为：

Initial
  └─ StartOrderProcess
       -> Created
       -> 发布 ValidateOrder

Created
  └─ OrderValidated
       -> Validated
       -> 发布 CreateCheckoutSession

Validated
  └─ CheckoutSessionCreated
       -> CheckoutSessionCreated
       -> 发布 ProcessPayment

CheckoutSessionCreated
  └─ OrderPaid
       -> Paid
       -> 发布 CollectOrder

Paid
  └─ OrderCollected
       -> Collected

任意活动状态
  └─ OrderCancelled
       -> Cancelled

这里故意把“请求执行某一步”和“某一步已经完成”拆成了不同消息。

配置示例

下面的代码重点展示状态与事件的对应关系。消息契约中的具体属性可以根据项目实际情况调整。

public class OrderStateMachine : MassTransitStateMachine<OrderStateData>
{
    public State Created { get; private set; }
    public State Validated { get; private set; }
    public State CheckoutSessionCreated { get; private set; }
    public State Paid { get; private set; }
    public State Collected { get; private set; }
    public State Cancelled { get; private set; }

    public Event<IStartOrderProcess> StartOrderProcess { get; private set; }
    public Event<IOrderValidated> OrderValidated { get; private set; }
    public Event<ICheckoutSessionCreated> CheckoutSessionCreatedEvent { get; private set; }
    public Event<IOrderPaid> OrderPaid { get; private set; }
    public Event<IOrderCollected> OrderCollected { get; private set; }
    public Event<IOrderCancelled> OrderCancelled { get; private set; }

    public OrderStateMachine()
    {
        InstanceState(x => x.CurrentState);

        Event(() => StartOrderProcess, e =>
        {
            e.CorrelateById(x => x.Message.OrderId);

            // 如果当前 MassTransit 版本或 repository 配置要求初始事件
            // 显式指定 Saga ID，可使用：
            e.SelectId(x => x.Message.OrderId);
        });

        Event(() => OrderValidated,
            e => e.CorrelateById(x => x.Message.OrderId));

        Event(() => CheckoutSessionCreatedEvent,
            e => e.CorrelateById(x => x.Message.OrderId));

        Event(() => OrderPaid,
            e => e.CorrelateById(x => x.Message.OrderId));

        Event(() => OrderCollected,
            e => e.CorrelateById(x => x.Message.OrderId));

        Event(() => OrderCancelled,
            e => e.CorrelateById(x => x.Message.OrderId));

        Initially(
            When(StartOrderProcess)
                .Then(context =>
                {
                    context.Instance.OrderId = context.Data.OrderId;
                    context.Instance.PackagesId = context.Data.PackagesIds;
                    context.Instance.OrderCreationDateTime = DateTime.UtcNow;
                })
                .TransitionTo(Created)
                .Publish(context => new ValidateOrder
                {
                    OrderId = context.Instance.OrderId,
                    PackagesIds = context.Instance.PackagesId
                })
        );

        During(
            Created,

            When(OrderValidated)
                .Then(context =>
                {
                    context.Instance.OrderId = context.Data.OrderId;
                })
                .TransitionTo(Validated)
                .Publish(context => new CreateCheckoutSession
                {
                    OrderId = context.Instance.OrderId,
                    PackagesIds = context.Instance.PackagesId
                })
        );

        During(
            Validated,

            When(CheckoutSessionCreatedEvent)
                .Then(context =>
                {
                    context.Instance.CheckoutSessionId =
                        context.Data.SessionId;
                })
                .TransitionTo(CheckoutSessionCreated)
                .Publish(context => new ProcessOrderPayment
                {
                    OrderId = context.Instance.OrderId,
                    CheckoutSessionId =
                        context.Instance.CheckoutSessionId
                })
        );

        During(
            CheckoutSessionCreated,

            When(OrderPaid)
                .Then(context =>
                {
                    context.Instance.PaymentMethod =
                        context.Data.PaymentMethod;
                })
                .TransitionTo(Paid)
                .Publish(context => new CollectOrder
                {
                    OrderId = context.Instance.OrderId,
                    PackagesIds = context.Instance.PackagesId
                })
        );

        During(
            Paid,

            When(OrderCollected)
                .Then(context =>
                {
                    context.Instance.OrderCollectedDateTime =
                        DateTime.UtcNow;
                })
                .TransitionTo(Collected)
        );

        DuringAny(
            When(OrderCancelled)
                .Then(context =>
                {
                    context.Instance.OrderCancelDateTime =
                        DateTime.UtcNow;
                })
                .TransitionTo(Cancelled)
        );

        During(
            Cancelled,
            Ignore(OrderCancelled)
        );

        During(
            Collected,
            Ignore(OrderCollected)
        );
    }
}

上述示例假设 ValidateOrder、CreateCheckoutSession、ProcessOrderPayment 和 CollectOrder 是发送给各个微服务的命令，而 IOrderValidated、ICheckoutSessionCreated、IOrderPaid 和 IOrderCollected 是这些服务发布的结果事件。

如果完成后需要删除 Saga

如果 Collected 和 Cancelled 都只是终止结果，不需要长期保留 Saga 实例，可以直接结束 Saga：

During(
    Paid,
    When(OrderCollected)
        .Then(context =>
        {
            context.Instance.OrderCollectedDateTime = DateTime.UtcNow;
        })
        .Finalize()
);

DuringAny(
    When(OrderCancelled)
        .Then(context =>
        {
            context.Instance.OrderCancelDateTime = DateTime.UtcNow;
        })
        .Finalize()
);

SetCompletedWhenFinalized();

这种配置最终保存的是框架内置的 Final 状态，而不是 Collected 或 Cancelled。如果业务需要查询最终结果，应将结果写入订单表、历史表或其他业务存储，而不要依赖已经完成的 Saga 实例。

关联 ID 必须保持一致

所有流程消息都必须携带相同的 OrderId：

e.CorrelateById(x => x.Message.OrderId)

若选择让 OrderId 同时充当 Saga 的主键，初始事件可以配置：

e.SelectId(x => x.Message.OrderId);

此时应保证：

context.Instance.CorrelationId == context.Data.OrderId

后续服务不能为每个步骤生成新的 OrderId，否则事件将无法找到原 Saga。

还要注意，通常只有 Initially(...) 中配置的事件能够创建 Saga。后续事件如果先于启动事件到达，默认情况下可能进入跳过队列或按缺失实例策略处理，而不会自动创建完整订单流程。

重复和乱序消息也要考虑

RabbitMQ 和分布式服务不能保证业务事件只出现一次。消费者重试、网络超时以及服务重新发布，都可能产生重复事件。

对于确认可以安全忽略的重复消息，可以在后续状态中显式配置：

During(
    Validated,
    Ignore(OrderValidated)
);

不要为了消除异常而对所有事件统一使用 Ignore。乱序事件可能意味着上游服务存在问题，静默忽略会掩盖真正的流程错误。更稳妥的做法是只忽略已确认完成步骤的重复事件，同时记录异常乱序事件。

发布下一步骤消息时也建议启用 MassTransit 的 outbox。否则可能出现 Saga 状态尚未可靠持久化，下游结果事件却已经返回的竞态问题。具体配置方式取决于 MassTransit 版本以及所用的 Saga repository。

最后，建议统一使用 Paid，而不是 Payed。后者虽然可以作为合法标识符使用，但在支付语境下通常是拼写错误。统一状态名、事件名和消息契约名称，能显著减少这种“看起来是同一个步骤，实际上是不同消息”的配置问题。

备注：内容仅供参考。