Masstransit saga 故障处理答案

【问题标题】：Masstransit saga fault handlingMasstransit saga 故障处理
【发布时间】：2019-01-18 01:14:32
【问题描述】：

我正在尝试使用 masstransit 和 azure sb 创建一个简单的 saga 错误处理场景，并为接收器提供了以下工作。

传奇当前在其中一个步骤中引发错误，并在重试几次后按预期发送到 _error 队列。

根据在 saga 中导致错误的步骤，消息将由 Step1FaultConsumer 或 Step2FaultConsumer 处理。我想知道是否有更好的方法可以从单个位置处理整个 saga 中的错误，而不管导致错误的步骤如何。

如果可能，我正在尝试不使用状态机来执行此操作。

程序.cs

 var bus = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
        {
            var host = cfg.Host(new Uri("HostAddress"),
                h =>
                {
                    h.TransportType = TransportType.AmqpWebSockets;
                    h.OperationTimeout = TimeSpan.FromSeconds(5);
                    h.RetryLimit = 5;
                    h.TokenProvider = TokenProvider.CreateSharedAccessSignatureTokenProvider("RootManageSharedAccessKey", "SharedAccessKey");
                });

            cfg.RequiresSession = true;
            cfg.UseRetry(x => x.Exponential(5, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(2)));
            cfg.ReceiveEndpoint(host, "QueueName", e =>
            {
                e.RequiresSession = true;
                e.Saga<SagaConsumer>(new MessageSessionSagaRepository<SagaConsumer>());
                e.Consumer<Step1FaultConsumer>();
                e.Consumer<Step2FaultConsumer>();
            });
        });

saga.cs

public class SagaConsumer:
    ISaga,
    InitiatedBy<ISagaStep1>,
    Orchestrates<ISagaStep2>
{
    public Guid CorrelationId { get; set; }

    public async Task Consume(ConsumeContext<ISagaStep1> context)
    {
        if(context.Message.Input1 == "000")
             throw new Exception("Saga Step 1 Exception");
        await context.Send<ISagaStep2>(context.DestinationAddress, new { Input2 = $"2-{context.Message.Input1}" });
    }

    public async Task Consume(ConsumeContext<ISagaStep2> context)
    {
        throw new Exception("Saga Step 2 Exception");
    }
}

public class Step1FaultConsumer:
    IConsumer<Fault<ISagaStep1>>
{
    public async Task Consume(ConsumeContext<Fault<ISagaStep1>> context)
    {
        // Handle Error
    }
}

public class Step2FaultConsumer:
    IConsumer<Fault<ISagaStep2>>
{
    public async Task Consume(ConsumeContext<Fault<ISagaStep2>> context)
    {
        // Handle Error
    }
}

【问题讨论】：

标签： .net .net-core masstransit azure-servicebus-queues

【解决方案1】：

您可以通过两个IConsumer<T> 接口和两个Consume 方法更新您的消费者以处理这两种故障类型。

我还建议为错误使用者使用单独的队列/接收端点，以避免阻塞 saga 端点上的吞吐量。

【讨论】：