.NET Core消费Kafka Topic陷入无限循环，如何解决？

你的代码里while (true)是无条件的无限循环，再加上Consume()方法本身是阻塞式调用（没有新消息时会一直等待），导致程序会一直卡在这个循环里无法正常退出，这就是问题根源。

解决办法 #

1. 添加可控制的退出条件（推荐）

使用CancellationToken监听外部中断信号（比如控制台的Ctrl+C），让程序可以优雅终止循环并释放资源：

var cts = new CancellationTokenSource();
// 监听控制台中断事件
Console.CancelKeyPress += (sender, e) =>
{
    e.Cancel = true; // 阻止系统直接终止进程
    cts.Cancel(); // 触发取消信号
};

consumer.Subscribe("my topic name");
try
{
    // 用取消信号控制循环
    while (!cts.Token.IsCancellationRequested)
    {
        var kfResult = consumer.Consume(cts.Token);
        // 这里添加你的消息处理逻辑
        Console.WriteLine($"处理消息：{kfResult.Message.Value}");
        
        // 如果是手动提交偏移量（需配置EnableAutoCommit=false），记得提交
        consumer.Commit(kfResult);
    }
}
catch (OperationCanceledException)
{
    // 捕获取消异常，可做清理操作
}
finally
{
    // 务必关闭并释放消费者资源
    consumer.Close();
    consumer.Dispose();
}

2. 设置消费超时退出

如果不需要持续监听，可以给Consume()设置超时时间，超时后返回null，以此判断是否退出循环：

consumer.Subscribe("my topic name");
while (true)
{
    // 设置5秒超时，超时后返回null
    var kfResult = consumer.Consume(TimeSpan.FromSeconds(5));
    if (kfResult == null)
    {
        // 无消息超时，退出循环
        break;
    }
    // 处理消息
    Console.WriteLine($"处理消息：{kfResult.Message.Value}");
    consumer.Commit(kfResult);
}
// 释放资源
consumer.Close();
consumer.Dispose();

额外注意事项 #

• 若配置了EnableAutoCommit=false，必须在处理完消息后手动调用Commit()提交偏移量，否则重启消费者会重复消费未提交的消息。
• 无论哪种方式，都要在退出时调用Close()和Dispose()释放消费者资源，避免资源泄漏。

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