如何通过MassTransit操作Azure Service Bus死信队列：发送与消费

一、消息进入死信队列的条件 #

消息进入死信队列（DLQ）的触发场景有多种，常见的包括：

• 重试次数耗尽：像你配置的UseMessageRetry(r => r.Interval(5, 500))，当消息连续消费失败5次后，会被自动移入死信队列
• 消息过期：发送消息时设置了TimeToLive，到期后未被消费的消息会进入死信队列
• 手动标记死信：在消费逻辑中主动调用context.DeadLetter()方法，直接将消息移入死信队列
• 队列配额超限：Azure Service Bus队列达到最大长度或大小限制时，新消息可能被移入死信
• 消息处理失败：比如消息格式错误无法反序列化、消费过程中抛出未处理的异常，经过重试后仍失败的消息

二、是否只有无消费者时消息才会进入死信队列？ #

不是。即使有消费者存在，只要满足上面的任一条件（比如重试耗尽、手动标记、消息过期），消息都会进入死信队列。没有消费者的情况下，消息只会先停留在普通队列，直到过期后才会移入死信，这只是其中一种场景。

三、如何创建死信消费者？ #

Azure Service Bus中，普通队列的死信队列命名规则是原队列名/$DeadLetterQueue，主题订阅的死信队列是主题名/订阅名/$DeadLetterQueue。你可以通过以下方式创建消费者处理死信消息：

1. 配置死信队列的接收端点 #

在MassTransit的总线配置中，直接为死信队列添加接收端点并绑定消费者：

busFactory.ConfigureEndpoints(serviceProvider);

// 替换为你的原始队列名
busFactory.ReceiveEndpoint("your-original-queue-name/$DeadLetterQueue", endpoint =>
{
    endpoint.Consumer<DeadLetterEventCreatedConsumer>(serviceProvider);
});

2. 定义死信消费者类 #

针对IEventCreated类型的死信消息，可以创建专门的消费者：

public class DeadLetterEventCreatedConsumer : IConsumer<Fault<IEventCreated>>
{
    private readonly ILogger<DeadLetterEventCreatedConsumer> _logger;

    public DeadLetterEventCreatedConsumer(ILogger<DeadLetterEventCreatedConsumer> logger)
    {
        _logger = logger;
    }

    public Task Consume(ConsumeContext<Fault<IEventCreated>> context)
    {
        var firstException = context.Message.Exceptions.FirstOrDefault();
        _logger.LogError("处理死信消息，ID: {MessageId}，错误原因: {Reason}",
            context.MessageId,
            firstException?.Message ?? "未知错误");

        // 这里可以添加自定义处理逻辑，比如写入错误日志、触发告警等
        return Task.CompletedTask;
    }
}

如果需要处理所有类型的死信消息，也可以使用通用消息类型：

public class GenericDeadLetterConsumer : IConsumer<Message>
{
    private readonly ILogger<GenericDeadLetterConsumer> _logger;

    public GenericDeadLetterConsumer(ILogger<GenericDeadLetterConsumer> logger)
    {
        _logger = logger;
    }

    public Task Consume(ConsumeContext<Message> context)
    {
        var messageBody = Encoding.UTF8.GetString(context.Message.Body.ToArray());
        _logger.LogInformation("收到通用死信消息，ID: {MessageId}，内容: {Body}",
            context.MessageId,
            messageBody);

        return Task.CompletedTask;
    }
}

关于你现有配置的说明 #

你在EventCreatedConsumerDefinition中配置的ConfigureDeadLetterQueueDeadLetterTransport()和ConfigureDeadLetterQueueErrorTransport()已经正确开启了死信转发，重试失败的消息会自动进入死信队列，无需修改这部分配置。

四、测试死信消息的快速方法 #

如果要快速验证死信流程，可以修改你的消费者代码，主动触发死信：

public Task Consume(ConsumeContext<IEventCreated> context)
{
    try
    {
        var serializedMessage = JsonSerializer.Serialize(context.Message, new JsonSerializerOptions());
        logger.LogInformation("Event consumed: {serializedMessage}", serializedMessage);
        
        // 手动标记消息进入死信队列（测试用）
        return context.DeadLetter("测试触发", "手动将消息移入死信队列");
    }
    catch (Exception ex)
    {
        logger.LogError(ex, "处理消息失败，ID: {messageId}，重试次数: {retryAttempt}", context.MessageId, context.GetRetryAttempt());
        throw;
    }
}

内容的提问来源于stack exchange，提问作者Tom van den Bogaart