Kafka 消费者启动后与服务器的交互流程涉及多个关键步骤，主要包括初始化、查找组协调器、加入消费者组、分区分配、心跳维持、拉取数据和提交偏移量等。以下是详细的流程说明：

1. 初始化消费者

• 创建消费者实例：应用程序通过调用KafkaConsumer的构造函数，传入配置参数创建消费者实例。
• 配置参数：包括bootstrap.servers（Kafka集群地址）、group.id（消费者组ID）、key.deserializer、value.deserializer等。

2. 订阅主题

• 调用subscribe方法：消费者通过调用subscribe方法订阅一个或多个主题，也可以使用正则表达式来匹配多个主题。

3. 查找组协调器

• 消费者发送FindCoordinator请求：消费者向Kafka集群中的任意Broker发送FindCoordinator请求，请求中包含消费者组ID。
• Broker服务器接收请求：Broker根据消费者组ID计算出组协调器所在的Broker节点，并返回该节点的地址信息。
  - 计算组协调器算法：

/**      * 表示 内部主题 __consumer_offsets 的分区数量，默认初始化值是50（顺带一提__consumer_offsets 副本因子默认值是3）      * 初始值为 -1，表示尚未设置。      * 使用 volatile 关键字确保在多线程环境中对该变量的修改是可见的。      */      private volatile int numPartitions = -1;     /**      * 内部主题 __consumer_offsets 的各个分区分布在各个Broker服务器上，算出当前消费者组的协调器在哪个服务器      * 消费者组协调器所在brokerId = 消费者组id的哈希值 % 50       */ 	 coordinator_broker_id = Utils.abs(groupId.hashCode()) % numPartitions 	  	 public static int abs(int n) {         return (n == Integer.MIN_VALUE) ? 0 : Math.abs(n);     } 

• 消费者连接组协调器：
  • 消费者根据FindCoordinator响应中的地址信息，连接到组协调器。

4. 加入消费者组

• Kafka消费者加入消费者组的过程主要涉及JoinGroup和SyncGroup两个关键步骤。这个过程确保消费者能够以协调的方式加入消费者组，并且分区能够被合理地分配给消费者组内的消费者。以下是JoinGroup和SyncGroup的具体流程：
• 阶段一：JoinGroup阶段

  • （1）发送JoinGroup请求：当消费者启动并调用poll方法时，如果它尚未加入消费者组，或者需要重新加入（例如，因为再平衡），它会向组协调器发送JoinGroup请求。这个请求包含消费者的group.id、订阅的主题列表以及消费者使用的分区分配策略。

  • （2）等待响应：组协调器收到JoinGroup请求后，会等待一段时间，以允许其他消费者也发送他们的JoinGroup请求。这个等待时间是为了收集同一消费者组内所有消费者的信息。

  • （3）选择Leader：对于同一个消费者组的第一次JoinGroup请求，协调器会选择第一个消费者作为Leader。Leader负责为组内的所有消费者分配分区。Leader的选择基于消费者的JoinGroup请求顺序。

  • （4）分区分配策略：Leader消费者收到协调器的响应后，会根据提供的分区分配策略（如Range、RoundRobin等）和所有消费者的订阅信息来决定分区的分配方案。

• 阶段二：SyncGroup阶段

  • （1）发送SyncGroup请求：Leader消费者将分区分配方案通过SyncGroup请求发送给组协调器。随后，组内的其他消费者也发送SyncGroup请求，但不包含分区分配方案。

  • （2）协调器广播分区分配方案：组协调器接收到SyncGroup请求后，将leader消费者的分区分配方案广播给消费者组内的所有消费者。

5. 开始消费

• 消费者接收分区分配：每个消费者接收到SyncGroup响应后，会知道自己被分配到了哪些分区。
• 初始化分区消费：消费者根据分配到的分区，初始化分区消费的相关资源，如设置分区的偏移量。
• 拉取数据并消费：消费者开始从分配给它的分区拉取数据并进行消费。

6. 心跳维持和再平衡

• 发送心跳：消费者会定期向组协调器发送心跳，以表明它仍然活跃。
• 处理再平衡：如果有新的消费者加入或现有消费者离开消费者组，协调器会触发再平衡过程，重新分配分区。

7. 拉取数据

• 发送Fetch请求：消费者向分配给它的分区的Leader Broker发送Fetch请求，请求包含拉取数据的偏移量。
• 接收数据：Broker返回包含消息的响应，消费者处理这些消息。

8. 提交偏移量

• 自动提交：如果启用了自动提交（enable.auto.commit=true），消费者会定期自动提交消费的偏移量。
• 手动提交：如果使用手动提交，消费者需要调用commitSync或commitAsync方法提交偏移量。

9. 处理再平衡：

• 再平衡触发：当消费者组成员发生变化，协调器会触发再平衡。
  • 1. 消费者加入消费者组
       当新的消费者加入现有的消费者组时，会触发再平衡。新消费者可能是新启动的实例，或者是之前失败后重新加入的实例。
  • 2. 消费者离开消费者组
    • 主动离开：消费者调用close方法或者主动离开消费者组时，会触发再平衡。
    • 被动离开：如果消费者因为网络问题、崩溃或者长时间未发送心跳而被组协调器认为已经离开，也会触发再平衡。
  • 3. 订阅主题的分区数变化
       如果消费者组订阅的主题新增了分区，那么为了将新增的分区分配给消费者，也会触发再平衡。
  • 4. 消费者订阅模式变化
       如果消费者组内的任何消费者更改了其订阅模式（例如，通过subscribe方法订阅了新的主题或者取消订阅了某些主题），这也会触发再平衡。
  • 5. 组协调器变更
       如果负责管理消费者组的组协调器（Group Coordinator）发生变化（例如，因为原协调器所在的Broker宕机），新的协调器在接管消费者组管理职责时，会触发再平衡。
  • 6. 主题元数据变化
       消费者定期从Broker获取订阅主题的元数据（如分区信息）。如果检测到元数据变化，可能会触发再平衡，尽管这种情况较少见。
• 暂停拉取：在再平衡期间，消费者会暂停拉取数据。
• 重新分配分区：协调器重新分配分区，并通知消费者新的分区分配情况。
• 恢复拉取：再平衡完成后，消费者恢复拉取数据。

示例代码

Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "test-group"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  KafkaConsumer consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("test-topic")); // 订阅主题  // 消费者加入消费者组并开始消费的过程是在第一次调用poll方法时触发的 ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : records) {     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); }  // 提交偏移量 consumer.commitSync(); 

在上述代码中，消费者通过调用subscribe方法订阅了主题test-topic，然后通过调用poll方法触发了加入消费者组的完整流程，包括查找组协调器、加入消费者组、分区分配、拉取数据和提交偏移量等步骤。

总结

Kafka消费者启动后与服务器的交互流程是一个复杂的过程，涉及与组协调器的多次交互。这个流程确保了消费者能够正确地加入消费者组，分区能够被合理地分配给消费者组内的消费者，并且在消费者组成员变化时能够进行适当的再平衡，同时保证了消费者能够从正确的位置继续消费数据。