| 订阅 | subscribe 走组协调；assign 手动指定分区 |
|---|---|
| 提交 | 处理成功后 commitSync/commitAsync |
| 退出 | wakeup() 打断阻塞的 poll 后 close() |
最小示例
最小示例
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092");
props.put("group.id", "order-consumer");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> r : records) {
process(r);
}
consumer.commitSync();
}poll 循环里做同步外部调用（长 SQL、慢接口）会阻塞心跳之外的处理进度，容易触发再均衡。