如何在.NET Core控制台应用中消费完Kafka所有消息后退出while循环？

你的问题出在consumer.Consume()是阻塞式调用：当没有新消息时，它会一直等待，不会主动返回，导致程序卡在这一行无法执行后续逻辑。以下是两种可行的解决方式：

方法1：使用带超时的Consume重载 #

直接调用带TimeSpan参数的Consume方法，设置一个超时时间。当超时后没有收到新消息，方法会返回null，此时即可退出循环。

修改核心代码片段：

// 替换原有的无参Consume调用，设置5秒超时
var consumeResult = consumer.Consume(TimeSpan.FromSeconds(5));

// 超时后consumeResult为null，直接退出循环
if (consumeResult == null)
{
    logger.LogInformation("超时未收到新消息，退出消费循环");
    break;
}

方法2：结合分区EOF状态+超时（更精准） #

如果需要确保所有分区都已消费到末尾且无新消息时才退出，可以维护一个所有订阅分区的EOF状态，结合超时判断：

完整修改后的代码 #

var config = new ConsumerConfig
{
    GroupId = groupId,
    BootstrapServers = brokerList,
    SaslMechanism = SaslMechanism.Plain,
    SaslUsername = saslUsername,
    SaslPassword = saslPassword,
    SecurityProtocol = SecurityProtocol.SaslSsl,
    AutoOffsetReset = AutoOffsetReset.Earliest
};

// 记录每个分区是否已到达EOF
var partitionEofStatus = new Dictionary<TopicPartition, bool>();

using (var consumer = new ConsumerBuilder<Ignore, string>(config)
  .SetErrorHandler((_, e) => logger.LogInformation($"Error: {e.Reason}"))
  .SetStatisticsHandler((_, json) => logger.LogInformation($"Statistic{json}"))
  .Build())
{
    consumer.Subscribe(topic);
    
    // 初始化所有订阅分区的EOF状态为false
    foreach (var partition in consumer.Assignment)
    {
        partitionEofStatus[partition] = false;
    }

    try
    {
        while (true)
        {
            try
            {
                // 设置3秒超时
                var consumeResult = consumer.Consume(TimeSpan.FromSeconds(3));

                if (consumeResult == null)
                {
                    // 检查所有分区是否都已到达EOF
                    if (partitionEofStatus.Values.All(isEof => isEof))
                    {
                        logger.LogInformation("所有分区已消费完成，且无新消息，退出循环");
                        break;
                    }
                    // 还有分区未到EOF，继续等待
                    continue;
                }

                if (consumeResult.IsPartitionEOF)
                {
                    logger.LogInformation($"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                    // 更新该分区的EOF状态
                    partitionEofStatus[consumeResult.TopicPartition] = true;
                    continue;
                }

                if (consumeResult?.Message == null) { break; }
                
                var mess = consumeResult.Message.Value;
                var vesselScoreFleetData = JsonConvert.DeserializeObject<VesselScoreFleet>(mess);
                vesselScoreFleets.Add(vesselScoreFleetData);
                logger.LogInformation($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");

                try
                {
                    consumer.StoreOffset(consumeResult);
                }
                catch (KafkaException e)
                {
                    logger.LogError($"Store Offset error: {e.Error.Reason}");
                }
            }
            catch (ConsumeException e)
            {
                logger.LogError($"Consume error: {e.Error.Reason}");
            }
        }
    }
    catch (OperationCanceledException)
    {
        logger.LogError("Closing consumer.");
        consumer.Close();
    }
}

// 后续执行逻辑写在这里
logger.LogInformation("消费循环已退出，开始执行后续步骤");

说明 #

• 初始化时记录所有订阅的分区，标记初始EOF状态为false
• 每次收到IsPartitionEOF时，更新对应分区的状态为true
• 超时后检查所有分区是否都已到达EOF，是则退出循环，否则继续等待

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