如何通过Azure Service Bus Mass Transit重新处理死信消息

要重新处理Azure Service Bus中因异常产生的死信消息，同时保留原异常状态，可按以下步骤配置和实现：

1. 修改MassTransit服务配置 #

在原有服务配置基础上，添加死信队列（DLQ）的消费者配置，指定处理对应订阅死信消息的端点：

services.AddMassTransit(massTransitConfigurator =>
{
    // 注册普通消费者和死信消费者
    massTransitConfigurator.AddConsumer<TestConsumer>();
    massTransitConfigurator.AddConsumer<TestDeadLetterConsumer>();

    massTransitConfigurator.UsingAzureServiceBus((ctx, cfg) =>
    {
        cfg.SubscriptionEndpoint<TestModel>("test-topic", e =>
        {
            e.ConfigureConsumer<TestConsumer>(ctx);

            // 配置当前订阅的死信队列消费者
            e.DeadLetterQueueEndpoint(dlqCfg =>
            {
                dlqCfg.ConfigureConsumer<TestDeadLetterConsumer>(ctx);
            });
        });
    });
});

2. 实现死信消息消费者 #

创建专门的死信消费者，处理被死信的TestModel消息。通过DeadLetterMessage<TestModel>可以获取原消息内容，同时从Service Bus的消息负载中提取原异常的死信原因和错误描述：

public class TestDeadLetterConsumer : IConsumer<DeadLetterMessage<TestModel>>
{
    public async Task Consume(ConsumeContext<DeadLetterMessage<TestModel>> context)
    {
        // 获取原始消息内容
        TestModel originalMessage = context.Message.Message;
        
        // 获取死信的原因和原异常信息
        var sbMessage = context.GetPayload<Azure.Messaging.ServiceBus.ServiceBusReceivedMessage>();
        string deadLetterReason = sbMessage.DeadLetterReason;
        string deadLetterError = sbMessage.DeadLetterErrorDescription;

        // 执行重新处理逻辑（示例：记录日志、重试业务操作等）
        Console.WriteLine($"重新处理死信消息: {JsonSerializer.Serialize(originalMessage)}");
        Console.WriteLine($"死信原因: {deadLetterReason}, 原异常信息: {deadLetterError}");

        // 若重新处理失败，直接抛出异常，消息会再次被死信（保留异常状态）
        // throw new Exception("死信消息重新处理失败，再次进入死信队列");
    }
}

3. 保留未处理异常状态的关键要点 #

• 重新处理死信时，若处理逻辑仍抛出异常，MassTransit会自动将该消息再次移入死信队列，保留异常未处理的状态。
• 原异常的核心信息（死信原因、错误描述）已由Azure Service Bus自动保存，可通过ServiceBusReceivedMessage的属性获取，用于排查问题或记录日志。

内容的提问来源于stack exchange，提问作者Samuel Marvin Aguilos