为何我的基础Kafka Consumer示例无法正常运行？

我来帮你排查下这个问题哈，你遇到的Java消费者读不到消息的情况，主要有几个关键问题需要调整：

1. 丢失了持续轮询消息的循环逻辑 #

你代码里把while(true)循环给注释掉了！这就导致消费者只执行了一次poll(100)操作，100毫秒的等待时间非常短，大概率还没完成和Kafka集群的连接、拉取到消息，程序就直接结束了，自然看不到消息输出。

修复方案：取消while(true)的注释，让消费者持续轮询Kafka集群：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

2. 未配置偏移量重置策略，新消费组无法读取历史消息 #

你的消费者配置里没有设置auto.offset.reset参数，Kafka的默认值是latest。这意味着当你的消费组test-consumer-group第一次消费hellotopic时，它会从主题的最新偏移量开始拉取消息——而你之前已经用控制台消费者把历史消息消费过了，新的消费组自然拿不到已有的旧消息。

修复方案：在Properties中添加偏移量重置配置，让新消费组从主题最开始的位置拉取消息：

props.put("auto.offset.reset", "earliest");

3. 可选优化：延长轮询等待时间 #

poll(100)的100ms等待时间偏短，尤其是本地测试环境中，集群响应可能有延迟，你可以适当延长这个时间，比如改成1000毫秒（1秒），增加拉取到消息的概率：

ConsumerRecords<String, String> records = consumer.poll(1000);

调整后的完整代码 #

import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class SampleConsumer {
    public static void main(String[] args) {
        System.out.println("Start consumer code");
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-consumer-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        // 添加偏移量重置配置
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("hellotopic"));
        
        // 恢复循环，持续轮询消息
        while (true) {
            // 延长poll等待时间到1秒
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
        // 注：实际生产环境中建议添加关闭钩子来优雅关闭consumer，这里为了简化暂时省略
        // System.out.println("End consumer code"); 循环执行的话这句不会被执行到
    }
}

测试步骤 #

调整代码后，重新运行SampleConsumer，然后用控制台生产者发送一条新消息（比如key2:value2），或者重新生产之前的key1:value1，你就能看到消费者控制台输出消息的偏移量、key和value了。

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