Automatonymous状态机中如何在每个事件执行完成后发布下一个事件

完全可以在状态机定义中实现该需求，你当前的问题核心是将下一个步骤的发布动作放在了「触发当前步骤的事件」的处理分支中，该分支会在状态机接收到事件时立即执行，不会等待对应步骤的消费者业务逻辑执行完成。

正确实现思路 #

你需要为每个业务步骤新增执行完成回调事件，让消费者处理完当前步骤的业务逻辑后，主动发布对应的完成事件，状态机只有收到完成事件后，才会发布下一个步骤的执行命令，同时将需要跨步骤传递的参数提前存储在Saga状态实例中，解决参数传递问题。

具体实现代码 #

1. 扩展Saga状态类，存储跨步骤传递的参数 #

public class TestState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    // 按需添加所有需要跨步骤传递的参数
    public string SharedParam1 { get; set; }
    public int SharedParam2 { get; set; }
}

2. 定义步骤完成回调事件 #

// 步骤一执行完成回调事件
public interface IStepOneCompleted
{
    Guid CorrelationId { get; }
    // 若消费者执行完成后有返回参数，也可在此定义
}

// 步骤二执行完成回调事件
public interface IStepTwoCompleted
{
    Guid CorrelationId { get; }
}

3. 修改状态机定义 #

public class TestStateMachine : MassTransitStateMachine<TestState>
{
    public State StepOneProcessing { get; set; }
    public State StepOneCompleted { get; set; }
    public State StepTwoCompleted { get; set; }
    public State StepThreeCompleted { get; set; }

    public Event<ProcessStepOne> ProcessStepOne { get; set; }
    public Event<ProcessStepTwo> ProcessStepTwo { get; set; }
    public Event<ProcessStepThree> ProcessStepThree { get; set; }
    // 新增完成事件定义
    public Event<IStepOneCompleted> StepOneCompletedEvent { get; set; }
    public Event<IStepTwoCompleted> StepTwoCompletedEvent { get; set; }

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

        Initially(
            // 收到初始的步骤一执行请求
            When(ProcessStepOne)
                // 先把跨步骤参数存入Saga状态
                .Then(context => 
                {
                    context.Instance.SharedParam1 = context.Data.Param1;
                    context.Instance.SharedParam2 = context.Data.Param2;
                })
                .TransitionTo(StepOneProcessing));

        // 步骤一处理中状态下，只有收到完成事件才执行后续逻辑
        During(StepOneProcessing,
            When(StepOneCompletedEvent)
                // 从Saga状态取参数，发布步骤二执行命令
                .Publish(context => new ProcessStepTwo
                {
                    CorrelationId = context.Instance.CorrelationId,
                    Param1 = context.Instance.SharedParam1,
                    Param2 = context.Instance.SharedParam2
                })
                .TransitionTo(StepOneCompleted));

        During(StepOneCompleted,
            When(StepTwoCompletedEvent)
                .Publish(context => new ProcessStepThree
                {
                    CorrelationId = context.Instance.CorrelationId,
                    Param1 = context.Instance.SharedParam1
                })
                .TransitionTo(StepTwoCompleted));

        During(StepTwoCompleted, 
            When(ProcessStepThree)
                .TransitionTo(StepThreeCompleted));
    }
}

4. 调整消费者逻辑 #

在每个步骤的消费者中，业务处理完成后发布对应的完成事件即可：

public class ProcessStepOneConsumer : IConsumer<ProcessStepOne>
{
    public async Task Consume(ConsumeContext<ProcessStepOne> context)
    {
        // 执行步骤一的业务逻辑
        await DoStepOneBusiness(context.Data);
        
        // 业务执行完成，发布完成回调事件
        await context.Publish<IStepOneCompleted>(new 
        {
            context.Data.CorrelationId
        });
    }
}

补充说明 #

• 你之前尝试的WhenLeave确实不适合该场景，它的触发源不绑定特定事件，无法直接获取事件携带的参数，将共享参数存储在Saga状态实例中是MassTransit Saga的标准跨步骤参数传递方案。
• 复用的事件不需要修改，只需要在对应的消费者中统一加上发布完成事件的逻辑即可，多个状态机可以同时订阅同一个完成事件，互不影响。

内容的提问来源于stack exchange，提问作者Isard