java中使用kafkaconsumer需正确配置bootstrap.servers、group.id、key/value.deserializer、auto.offset.reset和enable.auto.commit；订阅主题后循环调用poll()拉取消息，手动提交偏移量并确保close()释放资源。
在 Java 中使用 KafkaConsumer 订阅主题并拉取消息，核心是正确配置消费者、订阅主题、循环调用 poll() 获取消息批次，并及时提交偏移量。关键点在于线程安全（KafkaConsumer 不是线程安全的）、合理设置 poll() 超时、以及避免重复消费或丢失数据。
初始化 KafkaConsumer 并配置必要参数
必须设置以下基础属性，否则会抛异常或行为异常：
-
bootstrap.servers：Kafka 集群地址，如
"localhost:9092" - group.id：消费者组 ID，决定消息如何分配和偏移量如何提交
-
key.deserializer 和 value.deserializer：指定反序列化器，常用
StringDeserializer或ByteArrayDeserializer -
auto.offset.reset：指定无初始偏移量或偏移量失效时的行为，如
"earliest"（从头读）或"latest"（只读新消息） -
enable.auto.commit：建议设为
false，手动控制提交时机，避免自动提交导致的数据重复或丢失
示例配置：
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "false"); // 关闭自动提交
KafkaConsumer<string string> consumer = new KafkaConsumer(props);
</string>
订阅主题并启动拉取循环
调用 subscribe() 方法传入主题名列表（支持正则表达式），然后进入 while(true) 循环调用 poll(Duration)：
-
poll()是阻塞式拉取，参数为最大等待时间（如Duration.ofMillis(100)），超时后返回当前已拉到的消息（可能为空） - 每次
poll()返回一个ConsumerRecords，它是按分区组织的ConsumerRecord集合，可遍历处理 - 不要在循环中创建新消费者实例；应复用单个实例
基础循环结构：
consumer.subscribe(Arrays.asList("my-topic"));
try {
while (true) {
ConsumerRecords<string string> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<string string> record : records) {
System.out.printf("topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
}
// 处理完一批后手动提交偏移量（见下节）
consumer.commitSync();
}
} finally {
consumer.close();
}
</string></string>
安全提交偏移量以保证精确一次语义
若启用手动提交（enable.auto.commit=false），应在确认消息**成功处理完毕后**再提交。推荐方式：
- commitSync()：同步提交，阻塞直到提交成功或抛出异常，适合对一致性要求高的场景
- commitAsync()：异步提交，不阻塞，但失败时不重试，需配合回调处理（如记录日志、告警）
- 可在每次
poll()后提交，也可累积多批次后统一提交（需自己管理偏移量对象）
简单安全做法：每批处理完立即 commitSync()；若想提升吞吐，可用 commitAsync((offsets, exception) -> { ... }) 并在回调中处理异常。
注意关闭资源与异常恢复
KafkaConsumer 必须显式调用 close()，否则可能泄露网络连接和线程资源。建议用 try-with-resources（Kafka 3.0+ 支持 AutoCloseable）或 finally 块确保关闭：
- 捕获
WakeupException可用于优雅退出（如收到 shutdown 信号时调用consumer.wakeup()） - 遇到
CommitFailedException表明提交失败（如消费者已退出组），通常需重启消费逻辑 - 网络抖动可能导致
TimeoutException或InterruptException，应按需重试或记录
不复杂但容易忽略。
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
