这是《Kafka 零基础实战指南》的独立章节版。本章从概念、实操和生产排查三个视角展开，代码块保留了原书可直接运行的版本。 消费者是业务代码里最容易出问题的一环：丢消息、重复消费、堆积、频繁再平衡，几乎都源于对 poll 模型与位移提交理解不深。本章把消费者彻底讲透。
6.1 消费者与消费组回顾
消费者必须属于某个消费组（group.id）：
- 同一组内：一个分区只分配给一个消费者，实现负载均衡；
- 不同组之间：各自独立消费全量数据，实现广播。
消费者数量与分区数的关系：
分区 6 个：
消费者 3 个 -> 每人 2 个分区（理想情况）
消费者 6 个 -> 每人 1 个分区
消费者 8 个 -> 6 个干活，2 个空闲
想提高消费并行度，先加消费者实例；超过分区数后必须加分区（有代价，见第 7 章）。
6.2 第一个消费者
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class SimpleConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "demo-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("order-events"));
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> r : records) {
System.out.printf("partition=%d offset=%d key=%s value=%s%n",
r.partition(), r.offset(), r.key(), r.value());
}
}
}
}
}
这个“单线程 poll 循环”是 Kafka 消费者的标准形态。所有框架（Spring Kafka、Flink 等）内部都是这个循环。
6.3 poll 循环的心跳模型
消费者靠两个后台线程维持“活着”的状态：
- 心跳线程：定期向协调器发
Heartbeat，证明进程还在。超过session.timeout.ms没心跳，被认为死亡，触发再平衡； - poll 线程（业务线程）：两次
poll()的间隔如果超过max.poll.interval.ms，即使心跳正常，也被认为“处理能力异常”，同样踢出组。
session.timeout.ms = 45000 进程级存活判定
heartbeat.interval.ms = 3000 心跳频率，建议约 1/3 session
max.poll.interval.ms = 300000 业务处理耗时上限
max.poll.records = 500 单次 poll 最大条数
处理慢导致被踢组的典型表现：日志里出现 MaxPollIntervalExceededException，随后再平衡，周而复始形成“再平衡风暴”。解法是调小 max.poll.records、调大 max.poll.interval.ms、优化业务逻辑或增加消费者。
6.4 位移提交
自动提交
enable.auto.commit=true（默认）时，客户端每隔 auto.commit.interval.ms（默认 5 秒）在 poll 时自动提交当前位移。
方便，但有两个问题：
- 先提交后处理的可能：自动提交发生在 poll 返回新数据时，若提交后还没处理完就宕机，重启后会跳过这些消息——丢消息；
- 处理完才提交的时序不可控：也可能处理完但还没到提交时间点就宕机，重启后重复消费。
生产环境推荐关闭自动提交，改为手动。
手动提交
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
while (true) {
var records = consumer.poll(Duration.ofMillis(500));
for (var r : records) {
process(r); // 业务处理
}
consumer.commitSync(); // 处理完成后提交
}
commitSync() 阻塞直到成功，失败会自动重试（无退休额限制内）；commitAsync() 异步且不重试，吞吐更高。常用的折中方案：
try {
while (true) {
var records = consumer.poll(Duration.ofMillis(500));
process(records);
consumer.commitAsync(); // 平时异步，高吞吐
}
} finally {
consumer.commitSync(); // 退出前同步兜底
consumer.close();
}
精确到分区的提交
当一批 records 来自多个分区且处理时间差异大时，可以按分区粒度提交：
for (var partition : records.partitions()) {
var partRecords = records.records(partition);
process(partRecords);
long lastOffset = partRecords.get(partRecords.size() - 1).offset();
consumer.commitSync(Map.of(partition, new OffsetAndMetadata(lastOffset + 1)));
}
提交的是“下一条要消费的 offset”，所以是 lastOffset + 1。
至少一次的处理原则
“先处理后提交”天然是至少一次：宕机时可能重复，但不会丢。重复交给业务幂等解决（数据库唯一键、Redis setnx、状态机判断等）。这是绝大多数系统的正确选择。
6.5 再平衡监听器
分区被收回或新分配时，可以在钩子里做清理与提交：
consumer.subscribe(List.of("order-events"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// 分区即将被收回：同步提交位移、提交本地事务、释放资源
consumer.commitSync();
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
// 新分区到位：初始化缓存、拉取本地状态、打印日志
}
});
onPartitionsRevoked 里的同步提交非常重要：再平衡后新拥有者会从已提交位置开始，这里不提交就可能导致整批重复。
6.6 优雅退出
poll() 会阻塞，从外部线程唤醒它的标准方式是 wakeup()：
final var main = Thread.currentThread();
Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));
try {
while (running) {
var records = consumer.poll(Duration.ofMillis(500));
process(records);
consumer.commitAsync();
}
} catch (WakeupException e) {
// 收到退出信号，正常跳出循环
} finally {
try {
consumer.commitSync();
} finally {
consumer.close();
}
}
K8s/容器环境里优雅退出尤其重要：直接 SIGKILL 会触发再平衡，增加消费中断时间。
6.7 从任意位置开始消费
auto.offset.reset
没有已提交位移（新组或位移过期）时的策略：
earliest：从最早消息开始；latest：只消费启动之后的新消息（默认）；none：直接抛异常。
seek 精确定位
consumer.subscribe(List.of("order-events"));
// 先 poll 一次加入组并获得分区
consumer.poll(Duration.ofMillis(1000));
for (var tp : consumer.assignment()) {
consumer.seek(tp, 100); // 从该分区 offset=100 开始
}
按时间定位
Map<TopicPartition, Long> timestamps = new HashMap<>();
for (var tp : consumer.assignment()) {
timestamps.put(tp, System.currentTimeMillis() - 3600_000L); // 1 小时前
}
var offsets = consumer.offsetsForTimes(timestamps);
offsets.forEach((tp, om) -> {
if (om != null) consumer.seek(tp, om.offset());
});
独立消费者
不需要消费组语义、只想固定读某些分区时，用 assign 代替 subscribe：
TopicPartition tp = new TopicPartition("order-events", 0);
consumer.assign(List.of(tp));
consumer.seekToBeginning(List.of(tp));
注意：assign 的消费者不会参与再平衡，也没有组协调，但提交位移仍然需要 group.id。
6.8 消费者核心参数
| 参数 | 默认值 | 说明 |
|---|---|---|
group.id |
无 | 消费组标识，同一组的消费者共同分摊分区 |
enable.auto.commit |
true | 自动提交；生产建议 false |
auto.commit.interval.ms |
5000 | 自动提交间隔 |
auto.offset.reset |
latest | 无位移时的起点策略 |
session.timeout.ms |
45000 | 心跳超时，超时踢出组 |
heartbeat.interval.ms |
3000 | 心跳频率 |
max.poll.interval.ms |
300000 | 两次 poll 最大间隔 |
max.poll.records |
500 | 单次 poll 最大记录数 |
fetch.min.bytes |
1 | Broker 返回的最小数据量，调大可降低请求频率 |
fetch.max.wait.ms |
500 | 不够 fetch.min.bytes 时的最长等待 |
max.partition.fetch.bytes |
1048576 | 每个分区单次 fetch 上限 |
isolation.level |
read_uncommitted | 事务场景用 read_committed（第 14 章） |
6.9 提高消费吞吐的思路
- 增加消费者实例，直到等于分区数；
- 单实例内多线程处理，但位移管理与再平衡要自己做（提交粒度、线程池关闭）；
- 调大
max.poll.records并批量处理（攒批写库比逐条快得多）； - 消费逻辑尽量轻：重逻辑下沉到专门服务，消费者只做分发；
- 数据库/外部 IO 常是瓶颈，先优化下游再怪 Kafka。
本章小结
- 消费者的标准形态是单线程 poll 循环，心跳与 poll 间隔共同决定存活判定；
- 手动提交 + 先处理后提交 = 至少一次，重复用业务幂等兜底；
- 再平衡监听器里做同步提交与资源清理，优雅退出用
wakeup()； seek/offsetsForTimes能从任意位置重放，这是 Kafka 作为“可回放日志”的核心能力；- 堆积治理优先考虑下游性能、批处理与消费者扩容。
思考题
- 自动提交在什么时序下会丢消息？什么时序下会重复消费？
- 为什么
commitSync提交的是lastOffset + 1而不是lastOffset？ - 消费者处理一条消息要 2 秒，
max.poll.records=500，会发生什么？给出至少两种修复方案。