.NET 6 Kafka消费者：重复拉取消息与退出死循环问题

原始代码 #

public List<KafkaCdrModel> Consume(string topic)
{
    List<KafkaCdrModel> result = new();
    var config = new ConsumerConfig
    {
        GroupId = consumerSettings.GroupId,
        BootstrapServers = consumerSettings.BootstrapServers,
        AutoOffsetReset = AutoOffsetReset.Earliest,
    };

    using var consumer = new ConsumerBuilder<string, string>(config).Build();
    consumer.Subscribe(topic);
    try
    {
        while (true)
        {
            try
            {
                var consumeResult = consumer.Consume();
                var serializedResult = JsonConvert.DeserializeObject<KafkaCdrModel>(consumeResult.Message.Value);
                result.Add(serializedResult);
            }
            catch (ConsumeException ex)
            {
                Console.WriteLine($"Error occured: {ex.Error.Reason}");
            }
        }

    }
    catch (OperationCanceledException)
    {
        consumer.Close();
    }
    return result;
}

问题1：测试时如何重复拉取相同的消息？ #

有三种实用方案：

• 更换全新GroupId：Kafka为每个消费组独立维护消息偏移量，只要使用未用过的GroupId，同时保持AutoOffsetReset = AutoOffsetReset.Earliest，消费者就会从topic起始位置重新拉取所有消息。
• 手动重置偏移量：不想换GroupId时，可通过代码或命令行重置：
  • 代码方式：订阅topic后，将所有分区偏移量定位到起始位置：

    consumer.Subscribe(topic);
    foreach (var partition in consumer.Assignment)
    {
        consumer.Seek(new TopicPartitionOffset(partition, Offset.Beginning));
    }
    

  • 命令行方式（本地测试用）：用Kafka自带工具重置：

    kafka-consumer-groups.sh --bootstrap-server <你的Kafka地址> --group <你的GroupId> --reset-offsets --to-earliest --topic <你的Topic> --execute
    

• 禁用自动提交偏移量：在ConsumerConfig中添加EnableAutoCommit = false，消费消息后偏移量不会自动同步到Kafka，每次启动消费者都会从上次未提交的位置（首次启动则从Earliest位置）重新消费。

问题2：无消息时如何让程序退出while循环？ #

核心是给Consume方法设置超时时间，超时未获取到消息就主动退出循环，修改后的代码示例：

public List<KafkaCdrModel> Consume(string topic)
{
    List<KafkaCdrModel> result = new();
    var config = new ConsumerConfig
    {
        GroupId = consumerSettings.GroupId,
        BootstrapServers = consumerSettings.BootstrapServers,
        AutoOffsetReset = AutoOffsetReset.Earliest,
    };

    using var consumer = new ConsumerBuilder<string, string>(config).Build();
    consumer.Subscribe(topic);
    try
    {
        // 设置无消息超时时间，这里设为3秒
        var timeout = TimeSpan.FromSeconds(3);
        while (true)
        {
            try
            {
                var consumeResult = consumer.Consume(timeout);
                // 超时后consumeResult返回null，退出循环
                if (consumeResult == null)
                {
                    Console.WriteLine("超时未获取到消息，退出消费循环");
                    break;
                }
                var serializedResult = JsonConvert.DeserializeObject<KafkaCdrModel>(consumeResult.Message.Value);
                result.Add(serializedResult);
            }
            catch (ConsumeException ex)
            {
                Console.WriteLine($"Error occured: {ex.Error.Reason}");
                // 若为超时异常，直接退出
                if (ex.Error.Code == ErrorCode.Local_Timeout)
                {
                    break;
                }
            }
        }
        // 提交已消费的偏移量（按需选择）
        consumer.Commit();
    }
    catch (OperationCanceledException)
    {
        consumer.Close();
    }
    return result;
}

也可以用CancellationToken实现更灵活的退出控制，比如设置超时自动取消：

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
while (!cts.Token.IsCancellationRequested)
{
    try
    {
        var consumeResult = consumer.Consume(cts.Token);
        // 处理消息逻辑
    }
    catch (OperationCanceledException)
    {
        // 超时或外部触发取消，退出循环
        break;
    }
}

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