【kafka实战】05 Kafka消费者消费消息过程源码剖析
1. 概述
Kafka消费者（Consumer）是Kafka系统中负责从Kafka集群中拉取消息的客户端组件。消费者消费消息的过程涉及多个步骤，包括消费者组的协调、分区分配、消息拉取、消息处理等。本文将深入剖析Kafka消费者消费消息的源码，并结合相关原理图进行讲解。
以下是一个使用 Java 编写的 KafkaConsumer 的示例。在这个示例中，我们将创建一个简单的 Kafka 消费者，连接到 Kafka 集群，订阅一个主题，并消费该主题中的消息。
1.1 消费者代码使用示例
- 已经安装并启动了 Kafka 集群。
- 你已经添加了 Kafka 客户端依赖到你的项目中。如果你使用 Maven，可以在
pom.xml中添加以下依赖：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.6.0</version>
</dependency>
示例代码
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
private static final String TOPIC_NAME = "test-topic";
private static final String BOOTSTRAP_SERVERS = "localhost:9092";
private static final String GROUP_ID = "test-group";
public static void main(String[] args) {
// 配置消费者属性
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
// 自动提交偏移量
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
// 键和值的反序列化器
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
// 创建 Kafka 消费者实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
try {
// 订阅主题
consumer.subscribe(Collections.singletonList(TOPIC_NAME));
// 持续消费消息
while (true) {
// 从 Kafka 拉取消息
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());
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭消费者
consumer.close();
}
}
}
代码解释
-
配置消费者属性：
BOOTSTRAP_SERVERS_CONFIG：指定 Kafka 集群的地址。GROUP_ID_CONFIG：指定消费者所属的消费组。ENABLE_AUTO_COMMIT_CONFIG：设置是否自动提交偏移量。AUTO_COMMIT_INTERVAL_MS_CONFIG：设置自动提交偏移量的时间间隔。KEY_DESERIALIZER_CLASS_CONFIG和VALUE_DESERIALIZER_CLASS_CONFIG：指定键和值的反序列化器。
-
创建 Kafka 消费者实例：使用配置好的属性创建
KafkaConsumer实例。 -
订阅主题：使用
subscribe方法订阅指定的主题。 -
持续消费消息：使用
poll方法从 Kafka 拉取消息，并遍历消费记录，打印消息的偏移量、键和值。 -
关闭消费者：在消费完成后，使用
close方法关闭消费者。
注意事项
- 确保 Kafka 集群的地址和主题名称正确。
- 如果需要手动提交偏移量，可以将
ENABLE_AUTO_COMMIT_CONFIG设置为false，并使用commitSync()或commitAsync()方法手动提交偏移量。
2. Kafka消费者消费消息的核心流程
Kafka消费者消费消息的核心流程可以分为以下几个步骤：
- 消费者组协调：消费者加入消费者组，并与组协调器（GroupCoordinator）进行协调。
- 分区分配：组协调器为消费者分配分区。
- 消息拉取：消费者从分配的分区中拉取消息。
- 消息处理：消费者处理拉取到的消息。
- 提交偏移量：消费者提交已处理消息的偏移量。
下面我们将结合源码详细分析每个步骤。
3. 源码剖析
关键组件说明
-
ConsumerCoordinator：
- 负责消费者组的协调和分区分配。
- 管理消费者的心跳和重平衡。
-
Fetcher：
- 负责从 Kafka Broker 拉取消息。
- 管理拉取请求和响应的处理。
-
SubscriptionState：
- 管理消费者订阅的主题和分区。
- 记录消费者的消费偏移量。
-
PartitionAssignor：
- 负责分区分配策略的实现。
-
OffsetCommitCallback：
- 处理偏移量提交的回调逻辑。
3.1 消费者组协调
消费者在启动时，首先需要加入消费者组，并与组协调器进行协调。组协调器负责管理消费者组的成员和分区分配。
// org.apache.kafka.clients.consumer.KafkaConsumer#subscribe
public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
// 1. 订阅主题
this.subscriptions.subscribe(new HashSet<>(topics), listener);
// 2. 加入消费者组
coordinator.subscribe(subscriptions);
}
在subscribe方法中，消费者首先订阅指定的主题，然后通过coordinator.subscribe方法加入消费者组。组协调器会为消费者分配一个唯一的memberId，并将其加入到消费者组中。
3.2 分区分配
组协调器在消费者加入消费者组后，会为消费者分配分区。分区分配策略由PartitionAssignor决定，Kafka提供了多种内置的分区分配策略，如RangeAssignor、RoundRobinAssignor等。
// org.apache.kafka.clients.consumer.internals.AbstractCoordinator#ensurePartitionAssignment
private void ensurePartitionAssignment() {
// 1. 获取分区分配结果
Map<String, List<TopicPartition>> assignments = partitionAssignor.assign(metadata.fetch(), subscriptions.subscription());
// 2. 更新消费者的分区分配
subscriptions.assignFromSubscribed(assignments.get(consumerId));
}
在ensurePartitionAssignment方法中，组协调器通过partitionAssignor.assign方法为消费者分配分区，并将分配结果更新到消费者的订阅信息中。
3.3 消息拉取
消费者在分配到分区后，会从分配的分区中拉取消息。Kafka消费者采用拉取模式（Pull Model），即消费者主动从Kafka集群中拉取消息。
// org.apache.kafka.clients.consumer.KafkaConsumer#poll
public ConsumerRecords<K, V> poll(Duration timeout) {
// 1. 拉取消息
Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchRecords(timeout);
// 2. 返回拉取到的消息
return new ConsumerRecords<>(records);
}
在poll方法中，消费者通过fetcher.fetchRecords方法从Kafka集群中拉取消息。fetcher是Kafka消费者中的一个重要组件，负责管理消息的拉取和偏移量的提交。
3.4 消息处理
消费者在拉取到消息后，会对消息进行处理。消息处理的具体逻辑由用户自定义，通常包括消息的反序列化、业务逻辑处理等。
// org.apache.kafka.clients.consumer.KafkaConsumer#poll
public ConsumerRecords<K, V> poll(Duration timeout) {
// 1. 拉取消息
Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchRecords(timeout);
// 2. 处理消息
for (Map.Entry<TopicPartition, List<ConsumerRecord<K, V>>> entry : records.entrySet()) {
for (ConsumerRecord<K, V> record : entry.getValue()) {
// 用户自定义的消息处理逻辑
processRecord(record);
}
}
// 3. 返回拉取到的消息
return new ConsumerRecords<>(records);
}
在poll方法中，消费者通过processRecord方法处理每条消息。processRecord方法的具体实现由用户自定义。
3.5 提交偏移量
消费者在处理完消息后，需要提交已处理消息的偏移量。偏移量的提交可以手动或自动进行，Kafka提供了多种偏移量提交策略，如自动提交、同步提交、异步提交等。
// org.apache.kafka.clients.consumer.KafkaConsumer#commitSync
public void commitSync() {
// 1. 提交偏移量
coordinator.commitOffsetsSync(subscriptions.allConsumed());
}
在commitSync方法中，消费者通过coordinator.commitOffsetsSync方法同步提交偏移量。同步提交会阻塞当前线程，直到偏移量提交成功。
4. 原理图
以下是Kafka消费者消费消息的核心流程示意图：
+-------------------+ +-------------------+ +-------------------+
| | | | | |
| 消费者组协调 | ----> | 分区分配 | ----> | 消息拉取 |
| | | | | |
+-------------------+ +-------------------+ +-------------------+
|
v
+-------------------+ +-------------------+ +-------------------+
| | | | | |
| 消息处理 | <---- | 提交偏移量 | <---- | 网络传输 |
| | | | | |
+-------------------+ +-------------------+ +-------------------+
5. 总结
Kafka消费者消费消息的过程涉及多个步骤，包括消费者组的协调、分区分配、消息拉取、消息处理和偏移量提交。通过源码剖析，我们可以更深入地理解Kafka消费者的工作原理。希望本文能够帮助你更好地理解Kafka消费者的内部机制。
6. 参考
- Kafka官方文档
- Kafka源码