MassTransit Saga状态机活动：如何使用IStateMachineActivity的Faulted方法？

要让IStateMachineActivity的Faulted方法触发，核心是让Execute方法抛出的异常能传递到MassTransit的Activity错误处理机制中，同时避免异常被上层提前拦截。以下是具体实现步骤和注意事项：

1. 实现自定义Activity #

首先定义一个实现IStateMachineActivity<TInstance, TMessage>的类，同时实现Execute和Faulted方法：

public class ProcessOrderActivity : IStateMachineActivity<OrderSaga, OrderSubmitted>
{
    public void Accept(StateMachineVisitor visitor)
    {
        visitor.Visit(this);
    }

    public async Task Execute(BehaviorContext<OrderSaga, OrderSubmitted> context, Behavior<OrderSaga, OrderSubmitted> next)
    {
        // 模拟业务执行时抛出异常
        throw new InvalidOperationException("订单库存不足，处理失败");
        
        // 无异常时继续执行后续行为
        await next.Execute(context);
    }

    public async Task Faulted<TException>(BehaviorExceptionContext<OrderSaga, OrderSubmitted, TException> context, Behavior<OrderSaga, OrderSubmitted> next) 
        where TException : Exception
    {
        // 统一异常处理逻辑：更新Saga状态、记录错误信息
        context.Instance.OrderStatus = OrderStatus.Failed;
        context.Instance.ErrorDetail = context.Exception.ToString();
        
        // 可选：让异常继续传递到后续错误处理流程（比如状态机的Fault状态转换）
        await next.Faulted(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateScope("process-order-activity");
    }
}

2. 在状态机中配置Activity #

将自定义Activity添加到状态机的行为链中，不要在该行为链中添加Catch拦截异常，否则Faulted方法不会被触发：

public class OrderStateMachine : MassTransitStateMachine<OrderSaga>
{
    public OrderStateMachine()
    {
        InstanceState(x => x.OrderStatus);

        // 定义事件
        Event(() => OrderSubmitted, x => x.CorrelateById(m => m.Message.OrderId));

        // 初始状态流程
        Initially(
            When(OrderSubmitted)
                .Activity(x => x.OfType<ProcessOrderActivity>()) // 使用自定义Activity
                .Then(context => Console.WriteLine($"订单 {context.Instance.OrderId} 处理完成"))
                .TransitionTo(OrderProcessed)
        );

        // 可选：捕获未被Activity处理的全局异常（如果需要）
        // 注意：如果在这里捕获异常，会覆盖Activity的Faulted处理，需谨慎使用
        // AnyException(
        //     x => x.Then(context => Console.WriteLine($"全局异常：{context.Exception.Message}"))
        // );
    }

    // 状态定义
    public State OrderProcessed { get; private set; }
    public State OrderFailed { get; private set; }

    // 事件定义
    public Event<OrderSubmitted> OrderSubmitted { get; private set; }
}

3. Saga配置注意事项 #

在配置MassTransit Saga时，避免为接收端点添加全局异常过滤器，否则异常会被提前拦截，导致Faulted方法无法触发：

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderSaga>()
        .InMemoryRepository(); // 可替换为其他持久化仓库

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.ReceiveEndpoint("order-saga-endpoint", e =>
        {
            e.ConfigureSaga<OrderSaga>(context);
            
            // 不要在这里添加全局异常过滤器，否则会优先处理异常
            // e.UseExceptionFilter(async context => 
            // {
            //     // 全局异常处理逻辑
            // });
        });
    });
});

关键要点 #

• 异常传递规则：只有当Execute方法抛出的异常未被内部try-catch或状态机的Catch拦截时，Faulted方法才会被触发。
• Faulted方法的作用：用于在Activity层面实现统一异常处理，比如更新Saga状态、记录错误上下文，避免重复编写异常处理代码。
• 后续流程控制：调用next.Faulted(context)可以让异常继续传递到状态机的后续错误处理环节（比如转换到OrderFailed状态），如果不需要继续传递，可省略该调用。

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