Mass Transit Azure Service Bus：将达最大投递次数的消息转至错误队列而非死信队列

方案1：队列级别配置死信消息转发 #

Mass Transit的ConnectReceiveEndpoint中无法直接设置ForwardDeadLetteredMessagesTo，但可以通过ConfigureQueue方法访问Azure Service Bus的原生队列配置来实现单队列级别的死信转发：

IReceiveEndpointConnector handle = connector.ConnectReceiveEndpoint(subscriberName, (context, configurator) => 
{
    // 配置原生Azure队列属性
    configurator.ConfigureQueue(queue =>
    {
        queue.ForwardDeadLetteredMessagesTo = "your-deadletter-forward-queue";
    });

    // 其他端点配置逻辑...
});

这段代码会直接修改目标队列的原生Azure Service Bus属性，仅对当前配置的队列生效，而非总线全局范围。

方案2：消费死信子队列并重发 #

Azure Service Bus的死信子队列命名规则为{主队列名称}/$DeadLetterQueue，可直接通过该名称创建接收端点消费死信消息：

1. 绑定特定消息类型的死信消费 #

// 从主队列名称推导死信子队列名称
var deadLetterQueueName = $"{subscriberName}/$DeadLetterQueue";

connector.ConnectReceiveEndpoint(deadLetterQueueName, (context, configurator) => 
{
    configurator.Consumer<DeadLetterMessageConsumer>(context);
});

// 死信消息消费者实现
public class DeadLetterMessageConsumer : IConsumer<Fault<YourMessageType>>
{
    private readonly ISendEndpointProvider _sendEndpointProvider;

    public DeadLetterMessageConsumer(ISendEndpointProvider sendEndpointProvider)
    {
        _sendEndpointProvider = sendEndpointProvider;
    }

    public async Task Consume(ConsumeContext<Fault<YourMessageType>> context)
    {
        // 获取原始消息内容
        var originalMessage = context.Message.Message;

        // 推回主队列重试
        var sendEndpoint = await _sendEndpointProvider.GetSendEndpoint(new Uri($"queue:{subscriberName}"));
        await sendEndpoint.Send(originalMessage);

        // 可根据业务需求添加日志记录、消息修正等逻辑
    }
}

2. 通用死信消息消费（兼容任意消息类型） #

如果需要处理无特定类型的死信消息，可通过RawJsonMessage直接读取原始消息体：

public class GenericDeadLetterConsumer : IConsumer<RawJsonMessage>
{
    private readonly ISendEndpointProvider _sendEndpointProvider;

    public GenericDeadLetterConsumer(ISendEndpointProvider sendEndpointProvider)
    {
        _sendEndpointProvider = sendEndpointProvider;
    }

    public async Task Consume(ConsumeContext<RawJsonMessage> context)
    {
        // 从死信队列地址解析主队列名称
        var mainQueueName = context.ReceiveEndpoint.Address.AbsolutePath.Trim('/').Split('/')[0];
        
        // 直接转发原始消息体到主队列
        var sendEndpoint = await _sendEndpointProvider.GetSendEndpoint(new Uri($"queue:{mainQueueName}"));
        await sendEndpoint.SendRaw(context.Body, context.ContentType);
    }
}

其他优化方案 #

• 前置重试策略减少死信：在接收端点配置阶段，通过UseMessageRetry设置灵活的重试规则，从源头减少死信产生：

  configurator.UseMessageRetry(r => r.Interval(3, TimeSpan.FromSeconds(5)));
  

• SDK手动批量处理：使用Azure Service Bus SDK直接读取死信队列，转换application/vnd.masstransit+json格式后重发：

  // 示例：用Service Bus SDK操作死信队列
  var receiver = new ServiceBusClient(connectionString).CreateReceiver(subscriberName, new ServiceBusReceiverOptions 
  {
      SubQueue = SubQueue.DeadLetter
  });
  
  var messages = await receiver.ReceiveMessagesAsync(maxMessageCount: 10);
  foreach (var message in messages)
  {
      var sender = new ServiceBusClient(connectionString).CreateSender(subscriberName);
      await sender.SendMessageAsync(new ServiceBusMessage(message.Body)
      {
          ContentType = message.ContentType,
          // 复制必要的消息属性（如CorrelationId等）
      });
      await receiver.CompleteMessageAsync(message);
  }
  

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