如何在ASP.NET Core中实现Kafka死信队列(DLQ)并查看失败消息

一、基于现有代码改造实现DLQ逻辑 #

1. 准备DLQ生产者实例 #

首先创建DLQ的Kafka生产者，和重试队列生产者逻辑一致，确保能把失败消息写入指定的死信主题。先在配置文件（如appsettings.json）中添加DLQ相关配置：

{
  "KafkaServer": "localhost:9092",
  "ConsumerGroup": "main-consumer-group",
  "Topic": "main-topic",
  "RetryTopic": "retry-topic",
  "DeadLetterTopic": "dlq-topic",
  "MaxRetryCount": "3",
  "DLQConsumerGroup": "dlq-consumer-group"
}

然后初始化DLQ生产者：

// 配置并初始化DLQ生产者
var dlqProducerConfig = new ProducerConfig
{
    BootstrapServers = _configuration["KafkaServer"],
    Acks = Acks.All // 确保消息可靠写入DLQ
};
var _dlqProducer = new ProducerBuilder<int, string>(dlqProducerConfig).Build();

2. 修改消费逻辑，加入重试次数判断与DLQ跳转 #

修改原有消费循环，通过消息头跟踪重试次数，超过阈值后将消息转发到DLQ：

while (!stoppingToken.IsCancellationRequested)
{
    try 
    {
        var consumeResult = consumer.Consume(stoppingToken);
        Console.WriteLine($"Consumed message '{consumeResult.Value}' at: '{consumeResult.TopicPartitionOffset}'.");

        // 从消息头读取当前重试次数，默认0
        var retryCount = 0;
        if (consumeResult.Message.Headers.TryGetLastBytes("RetryCount", out var retryBytes))
        {
            retryCount = BitConverter.ToInt32(retryBytes);
        }

        if (!TryConsume(consumeResult, stoppingToken)) 
        {
            retryCount++;
            var maxRetries = int.Parse(_configuration["MaxRetryCount"] ?? "3");
            
            if (retryCount <= maxRetries)
            {
                // 更新重试次数到消息头，发送到重试队列
                var retryMessage = new Message<int, string>
                {
                    Key = consumeResult.Message.Key,
                    Value = consumeResult.Message.Value,
                    Headers = consumeResult.Message.Headers
                };
                retryMessage.Headers.Remove("RetryCount");
                retryMessage.Headers.Add("RetryCount", BitConverter.GetBytes(retryCount));
                
                await _retryQueueProducer.ProduceAsync(_configuration["RetryTopic"], retryMessage, stoppingToken);
                Console.WriteLine($"Message sent to retry queue, retry count: {retryCount}");
            }
            else
            {
                // 超过最大重试次数，发送到DLQ
                await _dlqProducer.ProduceAsync(_configuration["DeadLetterTopic"], consumeResult.Message, stoppingToken);
                Console.WriteLine($"Message failed after {maxRetries} retries, sent to DLQ: {consumeResult.Value}");
                
                // 提交偏移量，避免重复消费该消息
                consumer.Commit(consumeResult);
            }
        }
        else
        {
            // 消费成功，提交偏移量
            consumer.Commit(consumeResult);
        }
    }
    catch (ConsumeException e)
    {
        Console.WriteLine($"Consume error occurred: {e.Error.Reason}");
        
        // 致命错误直接发送到DLQ
        if (e.Error.IsFatal && e.ConsumeResult?.Message != null)
        {
            await _dlqProducer.ProduceAsync(_configuration["DeadLetterTopic"], e.ConsumeResult.Message, stoppingToken);
            Console.WriteLine($"Fatal error, message sent to DLQ: {e.ConsumeResult.Message.Value}");
            consumer.Commit(e.ConsumeResult);
        }
    }
}

关键逻辑说明 #

• 重试次数跟踪：通过消息头RetryCount记录重试次数，服务重启后不会丢失计数。
• 偏移量提交：无论消费成功还是转入DLQ，都要提交偏移量，避免消息被重复处理。
• 致命错误处理：遇到无法恢复的消费异常时，直接将消息转入DLQ，不再重试。

二、查看DLQ中的失败消息 #

1. 使用Kafka命令行工具 #

如果服务器上安装了Kafka，可通过控制台消费者直接读取DLQ主题内容：

# Linux/macOS
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dlq-topic --from-beginning --property print.key=true --property print.headers=true

# Windows
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic dlq-topic --from-beginning --property print.key=true --property print.headers=true

参数说明：

• --from-beginning：从主题起始位置读取所有历史死信消息。
• print.key=true：打印消息Key，方便关联业务。
• print.headers=true：打印消息头，可查看RetryCount等自定义字段。

2. 编写DLQ消费者服务 #

在ASP.NET Core中编写后台服务，实时监控并处理DLQ消息：

public class DlqConsumerService : BackgroundService
{
    private readonly IConfiguration _configuration;

    public DlqConsumerService(IConfiguration configuration)
    {
        _configuration = configuration;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var kafkaConfig = new ConsumerConfig
        {
            GroupId = _configuration["DLQConsumerGroup"],
            BootstrapServers = _configuration["KafkaServer"],
            AutoOffsetReset = AutoOffsetReset.Earliest // 从最开始读取消息
        };

        using (var consumer = new ConsumerBuilder<int, string>(kafkaConfig).Build())
        {
            consumer.Subscribe(_configuration["DeadLetterTopic"]);

            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    var consumeResult = consumer.Consume(stoppingToken);
                    
                    // 解析重试次数
                    var retryCount = 0;
                    if (consumeResult.Message.Headers.TryGetLastBytes("RetryCount", out var retryBytes))
                    {
                        retryCount = BitConverter.ToInt32(retryBytes);
                    }
                    
                    Console.WriteLine($"DLQ消息 - Key: {consumeResult.Message.Key}, 内容: {consumeResult.Message.Value}, 重试次数: {retryCount}, 偏移量: {consumeResult.TopicPartitionOffset}");
                    
                    // 可扩展逻辑：比如将消息存入数据库、人工审核后重新入队等
                    // ProcessDlqMessage(consumeResult);
                    
                    consumer.Commit(consumeResult);
                }
            }
            finally
            {
                consumer.Close();
            }
        }
    }
}

在Program.cs中注册该服务：

builder.Services.AddHostedService<DlqConsumerService>();

启动服务后即可实时查看DLQ消息，还能根据需求扩展后续处理逻辑。

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