Kafka核心消费者机制详解
消费者模型演进与版本差异
Kafka的消费者实现经历了从基于ZooKeeper的旧架构到独立管理位移的新架构的演变。在0.9之前，消费者分为高层（high-level）与低层（low-level）两种模式，而自0.10.1.0版本起，官方推荐使用全新的Java API消费者。
旧版消费者：依赖ZooKeeper
- 低层消费者（SimpleConsumer）：直接与Broker通信，需手动管理分区、副本选举、偏移量提交及异常恢复，灵活性高但复杂度大。适用于需要精确控制消费位置或监控场景。
- 高层消费者（ConsumerConnector）：通过ZooKeeper协调组内成员、分区分配和偏移量存储，支持自动负载均衡。其核心元数据保存在：
/consumers/{group}/offsets/{topic}/{partition}，由ZooKeeper维护。
新版消费者：去中心化设计
新版本消费者不再依赖ZooKeeper，所有状态（如偏移量、组成员信息）由Kafka内部主题__consumer_offsets统一管理。该主题默认50个分区，用于持久化每个消费者组对各分区的消费进度。
核心概念解析
- 消费者组（Consumer Group）：同一组内的消费者共同消费一个或多个主题的数据，每条消息仅被组内一个实例处理。
- 偏移量（Offset）：记录消费者当前读取的位置。新版本中，偏移量由Kafka集群内部维护，而非客户端或ZooKeeper。
- 再平衡（Rebalance）：当组成员变化或订阅变更时，系统重新分配分区给消费者。此过程由特定Broker作为协调者（Coordinator）驱动。
单线程消费者示例
Properties props = new Properties();
props.put("bootstrap.servers", "kafka01:9092,kafka02:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
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");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic-a", "topic-b"));
try {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
record.partition(), record.offset(), record.key(), record.value());
}
}
} finally {
consumer.close();
}
关键配置说明
bootstrap.servers：Kafka集群地址列表，建议使用主机名。group.id：标识消费者所属组。auto.offset.reset：无初始偏移或偏移越界时的行为：earliest：从最早可用位置开始。latest：从最新消息开始。none：若任一分区无有效偏移，则抛出异常。
enable.auto.commit：是否启用自动提交偏移量，默认为真。max.poll.records：每次轮询最多返回的消息数量，默认500。session.timeout.ms：会话超时时间，影响rebalance触发速度。max.poll.interval.ms：两次poll之间最大允许处理时间，超过将视为消费者失效。
偏移量提交策略
偏移量提交方式决定了消息处理的语义：
- 最多一次（At Most Once）：先提交后处理，可能丢失消息。
- 最少一次（At Least Once）：先处理后提交，可能重复。
- 精确一次（Exactly Once）：通过事务+幂等性实现，需配合生产者启用。
推荐手动提交以确保准确性，可通过以下方法：
commitSync()：同步提交，阻塞等待。commitAsync()：异步提交，不阻塞主线程。
再平衡机制与分区分配策略
再平衡触发条件包括：
- 消费者加入或退出组。
- 订阅的主题或正则表达式发生变化。
- 分区数量动态调整。
默认分区分配策略为RangeAssignor，可替换为：
RoundRobinAssignor：轮询分配，适合单一主题。StickyAssignor：保持历史分配状态，减少重分配次数，0.11.0.0引入。
多线程消费实现
由于KafkaConsumer非线程安全，应为每个线程创建独立实例。典型做法是启动多个消费者线程，分别绑定不同分区。
public class ConsumerRunnable implements Runnable {
private final KafkaConsumer<String, String> consumer;
public ConsumerRunnable(String bootstrapServers, String groupId, String topic) {
Properties props = new Properties();
props.put("bootstrap.servers", bootstrapServers);
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
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");
this.consumer = new KafkaConsumer<>(props);
this.consumer.subscribe(Arrays.asList(topic));
}
@Override
public void run() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
}
}
}
}
独立消费者（Standalone Consumer）
某些场景下需指定消费者只消费特定分区，不受组管理影响。使用assign()方法显式指定分区集合。
List<TopicPartition> partitions = new ArrayList<>();
List<PartitionInfo> partitionInfos = consumer.partitionsFor("test-topic");
for (PartitionInfo info : partitionInfos) {
partitions.add(new TopicPartition(info.topic(), info.partition()));
}
consumer.assign(partitions);
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Partition: %d, Offset: %d, Value: %s%n", record.partition(), record.offset(), record.value());
}
}
上述实现适用于需要精确控制消费路径、避免组级再平衡干扰的场景。
