上一篇【第27篇】SubscriptionState源码解析——消费者是怎么记住自己订阅了什么下一篇【第29篇】Kafka心跳机制源码解析——消费者如何向Broker报平安摘要在Kafka消费者中ConsumerCoordinator扮演着外交官的角色——它负责与Broker端的GroupCoordinator进行一系列复杂的协议交互。从查找Coordinator位置、发送JoinGroup请求、接收分区分配策略、到发送SyncGroup请求确认分配结果、再到周期性地发送心跳维护成员身份这一切都由ConsumerCoordinator统筹调度。它继承了AbstractCoordinator抽象类在其中定义了心跳管理、状态转换、成员身份维护的通用逻辑。本文从ConsumerCoordinator的继承体系入手逐层剖析其核心职责、心跳线程的调度机制、JoinGroup/SyncGroup的完整发送与处理流程以及分区分配结果如何被应用到SubscriptionState中。一、ConsumerCoordinator的家族谱系【ConsumerCoordinator 类继承关系】 AbstractCoordinator (抽象基类) │ ├─ heartbeat: Heartbeat ← 心跳辅助类 ├─ heartbeatTask: HeartbeatTask ← 心跳定时任务 ├─ groupId: String ├─ client: ConsumerNetworkClient ├─ coordinator: Node ← GroupCoordinator所在Broker ├─ memberId: String ← 服务端分配的唯一成员ID ├─ generation: int ← 年代号(每次Rebalance递增) ├─ needsJoinPrepare: boolean ← 是否需要在Join前做准备 ├─ rejoinNeeded: boolean ← 是否需要重新加入Group │ ├─ ensureActiveGroup() ← 确保Consumer在Group中活跃 ├─ pollHeartbeat() ← 发送心跳 └─ coordinatorUnknown() ← 检查Coordinator是否已知 │ ▼ (继承) ConsumerCoordinator │ ├─ assignors: ListPartitionAssignor ← 分区分配策略列表 ├─ metadata: Metadata ├─ subscriptions: SubscriptionState ├─ autoCommitEnabled: boolean ← 是否自动提交offset ├─ autoCommitTask: AutoCommitTask ← 自动提交定时任务 ├─ interceptors: ConsumerInterceptors ← 拦截器链 │ ├─ poll() ← 主协调方法 ├─ onJoinPrepare() ← JoinGroup前的准备钩子 ├─ onJoinComplete() ← JoinGroup完成后的处理钩子 ├─ commitOffsetsSync/Async() ← 提交offset └─ maybeAutoCommitOffsets() ← 自动提交offset二、AbstractCoordinator——协调器的公共底座2.1 核心字段publicabstractclassAbstractCoordinator{// 心跳管理protectedfinalHeartbeatheartbeat;// 心跳计时器privateHeartbeatTaskheartbeatTask;// 心跳定时任务// Group信息protectedfinalStringgroupId;// Consumer Group IDprotectedNodecoordinator;// GroupCoordinator的位置protectedStringmemberId;// 分配给当前消费者的成员IDprotectedintgeneration;// 年代号(Rebalance递增)// 网络通信protectedfinalConsumerNetworkClientclient;// 状态标志protectedbooleanneedsJoinPrepare;// 是否需要在Join前运行准备钩子protectedbooleanrejoinNeeded;// 是否需要重新Join}2.2 ensureActiveGroup()——协调核心入口这是ConsumerCoordinator中最重要的方法之一它在每次poll()中被调用确保消费者处于Group的活跃状态// AbstractCoordinator.ensureActiveGroup() 核心流程publicvoidensureActiveGroup(){// 1. 如果没有指定Coordinator先查找if(coordinatorUnknown()){lookupCoordinator();}// 2. 如果needsJoinPreparetrue执行准备操作if(needsJoinPrepare){onJoinPrepare(generation,memberId);needsJoinPreparefalse;}// 3. 发送JoinGroupRequestRequestFutureByteBufferfuturesendJoinGroupRequest();// 4. 等待JoinGroupResponse处理响应// (在onJoinComplete中处理)}状态流转【Consumer端Rebalance状态机】 ┌──────────┐ │ UNJOINED │ ← 初始状态 / 被踢出Group └────┬─────┘ │ needsJoinPreparetrue ▼ ┌──────────────────────┐ │ PREPARING_REBALANCE │ ← 准备阶段 │ (onJoinPrepare执行) │ ① 提交offset防止重复消费 └──────┬───────────────┘ ② 刷新本地缓存 │ │ sendJoinGroupRequest ▼ ┌──────────────────────┐ │ COMPLETING_REBALANCE │ ← 等待分配完成 │ (Leader计算分配结果) │ ① Leader: 执行PartitionAssignor.assign() └──────┬───────────────┘ ② 发送SyncGroupRequest │ │ SyncGroupResponse成功 ▼ ┌──────────┐ │ STABLE │ ← 正常消费中 └────┬─────┘ 定期发送心跳 │ │ 心跳失败/ILLEGAL_GENERATION ▼ 重新进入 UNJOINED 或 PREPARING_REBALANCE2.3 generation——年代号防乱序generation是防止过期请求干扰的关键设计【generation的作用示意】 Rebalance #1 (generation1): Consumer发送HeartbeatRequest(generation1) Rebalance #2 (generation2): ← 新一轮Rebalance ┌── generation递增为2 ──┐ │ │ │ 此时收到了Network延迟的请求... │ HeartbeatResponse(generation1) ← 来自上一轮的过期响应 │ │ Broker端检测: generation1 ≠ 当前generation2 │ 返回 ILLEGAL_GENERATION 错误 │ │ Consumer端处理: rejoinNeeded true │ Consumer重新发送JoinGroupRequest └──────────────────────────┘三、查找GroupCoordinator——“谁的场子”消费者在加入Group之前需要先找到管理本Consumer Group的GroupCoordinator。这个查找过程也叫FindCoordinator。【GroupCoordinator查找流程】 Consumer 任意Broker │ │ ├─FindCoordinatorRequest──────────────►│ │ groupId order-consumers │ │ ├─ 根据groupId计算hash │ │ hash(order-consumers) % 50 │ │ → 找到 __consumer_offsets 分区N │ │ → 分区N的Leader所在的Broker │ │ 就是GroupCoordinator │◄──FindCoordinatorResponse────────────┤ │ coordinator: Broker-3:9092 │ │ │ │ consumer.coordinator Broker-3 │ │ │ ├─后续所有请求都发给Broker-3──────────────►│ GroupCoordinator源码实现// AbstractCoordinator.lookupCoordinator()protectedsynchronizedvoidlookupCoordinator(){// 发送FindCoordinator请求到负载最低的BrokerNodenodeclient.leastLoadedNode();RequestFutureVoidfuturesendFindCoordinatorRequest(node);// 等待响应client.poll(future);// 响应中的coordinator已经被设置到 this.coordinator 字段}GroupCoordinator分配规则Kafka根据Consumer Group的groupId计算hash然后映射到__consumer_offsets内部Topic的某个分区该分区的Leader所在的Broker就是这个Group的GroupCoordinator。一个Broker可以同时担任多个Consumer Group的Coordinator。四、JoinGroup/SyncGroup请求的发送与处理4.1 sendJoinGroupRequest()——“我要加入”// ConsumerCoordinator.sendJoinGroupRequest()privateRequestFutureByteBuffersendJoinGroupRequest(){// 构建 JoinGroupRequestJoinGroupRequestrequestnewJoinGroupRequest(groupId,// Consumer Group IDsessionTimeoutMs,// session超时generation.memberId,// 当前成员IDprotocolType(),// consumermetadata()// 订阅信息 支持的分配策略);// protocolMetadata 包含的信息:// {// version: 1,// subscription: [topic-A, topic-B], // 订阅的Topic列表// userData: // 自定义数据// }// 发送并处理响应returnclient.send(coordinator,ApiKeys.JOIN_GROUP,request).compose(newJoinGroupResponseHandler());}4.2 JoinGroupResponseHandler——处理JoinGroup响应// 处理JoinGroupResponseprivateclassJoinGroupResponseHandlerextendsRequestFutureAdapterClientResponse,ByteBuffer{OverridepublicvoidonSuccess(ClientResponseresp,RequestFutureByteBufferfuture){JoinGroupResponsejoinResponsenewJoinGroupResponse(resp.responseBody());// 处理错误handleJoinGroupResponseError(resp,future);// 保存Leader信息// Leader的响应中包含所有Member的订阅信息// Follower的响应只包含Leader的IDif(joinResponse.isLeader()){// 保存全量订阅信息用于后续分区分配subscriptions.groupSubscription(joinResponse.members().stream().flatMap(m-parseSubscriptions(m).stream()).collect(Collectors.toSet()));// 执行分区分配 ← 只有Leader做这件事MapString,AssignmentassignmentsperformAssignment(joinResponse.leaderId(),joinResponse.groupProtocol(),// 选定的分配策略joinResponse.members());// 将分配结果封装到SyncGroupRequest中// ...}future.complete(joinResponse.memberAssignment());}}4.3 performAssignment()——分区分配计算// ConsumerCoordinator.performAssignment()// 只有Leader Consumer执行privateMapString,AssignmentperformAssignment(StringleaderId,StringgroupProtocol,// 选定的分区分配策略名ListJoinGroupResponseMemberallMembers){// 1. 按Topic收集分区数MapString,IntegerpartitionsPerTopicnewHashMap();for(Stringtopic:subscriptions.groupSubscription()){partitionsPerTopic.put(topic,metadata.fetch().partitionCountForTopic(topic));}// 2. 收集每个Member的订阅信息MapString,ListStringmemberSubscriptionsnewHashMap();for(JoinGroupResponseMembermember:allMembers){memberSubscriptions.put(member.memberId(),parseSubscriptions(member));}// 3. 找到对应名称的PartitionAssignorPartitionAssignorassignorfindAssignor(groupProtocol);// 4. 执行分区分配MapString,ListTopicPartitionrawResultassignor.assign(partitionsPerTopic,memberSubscriptions);// 5. 封装结果MapString,AssignmentresultnewHashMap();rawResult.forEach((memberId,partitions)-result.put(memberId,newAssignment(partitions.stream().map(tp-newTopicPartition(tp.topic(),tp.partition())).collect(Collectors.toList()))));returnresult;}4.4 onJoinComplete()——分配结果应用// ConsumerCoordinator.onJoinComplete()protectedvoidonJoinComplete(intgeneration,StringmemberId,StringassignmentStrategy,ByteBufferassignmentBuffer){// 1. 解析SyncGroupResponse中的分区分配结果AssignmentassignmentConsumerProtocol.deserializeAssignment(assignmentBuffer);// 2. 将分配结果应用到SubscriptionStatesubscriptions.assignFromSubscribed(assignment.partitions());// 3. 触发用户定义的ConsumerRebalanceListenerconsumerRebalanceListener.onPartitionsAssigned(assignment.partitions());// 4. 设置标志位this.needsJoinPreparetrue;// 下次Join前先执行准备钩子this.rejoinNeededfalse;// Join完成清标志}五、ConsumerCoordinator.poll()——协调器的主循环// ConsumerCoordinator.poll() 核心流程publicvoidpoll(longnow){// Step 1: 检查是否需要JoinGroupif(rejoinNeeded){ensureActiveGroup();// 触发JoinGroup → SyncGroup流程}// Step 2: 检查状态决定是否需要再次Join// (例如心跳失败、收到ILLEGAL_GENERATION等)checkState();// Step 3: 发送心跳如果在STABLE状态if(needToSendHeartbeat(now)){sendHeartbeatRequest();}// Step 4: 自动提交offsetif(autoCommitEnabled){maybeAutoCommitOffsetsAsync(now);}}本篇小结ConsumerCoordinator是消费者端最复杂的组件承担着与GroupCoordinator的外交职责查找Coordinator通过FindCoordinator请求定位管理本Group的Broker基于groupId的hash映射到__consumer_offsets分区JoinGroup协议向GroupCoordinator发送JoinGroupRequest由GroupCoordinator选举出Group Leader并指定分区分配策略分区分配Group Leader执行PartitionAssignor的计算逻辑只有Leader做然后将结果通过SyncGroupRequest上报结果应用所有Consumer收到SyncGroupResponse后通过onJoinComplete()钩子将分区更新到SubscriptionState并触发用户的自定义RebalanceListenergeneration防乱序通过递增的年代号防止过期请求对当前Rebalance造成干扰下一篇我们将聚焦心跳机制看看消费者是如何通过HeartbeatThread不断地向Broker报平安的。上一篇【第27篇】SubscriptionState源码解析——消费者是怎么记住自己订阅了什么下一篇【第29篇】Kafka心跳机制源码解析——消费者如何向Broker报平安
【Kafka源码解读和使用指南】第28篇：ConsumerCoordinator源码解析——消费者与GroupCoordinator的“谈判桌“
上一篇【第27篇】SubscriptionState源码解析——消费者是怎么记住自己订阅了什么下一篇【第29篇】Kafka心跳机制源码解析——消费者如何向Broker报平安摘要在Kafka消费者中ConsumerCoordinator扮演着外交官的角色——它负责与Broker端的GroupCoordinator进行一系列复杂的协议交互。从查找Coordinator位置、发送JoinGroup请求、接收分区分配策略、到发送SyncGroup请求确认分配结果、再到周期性地发送心跳维护成员身份这一切都由ConsumerCoordinator统筹调度。它继承了AbstractCoordinator抽象类在其中定义了心跳管理、状态转换、成员身份维护的通用逻辑。本文从ConsumerCoordinator的继承体系入手逐层剖析其核心职责、心跳线程的调度机制、JoinGroup/SyncGroup的完整发送与处理流程以及分区分配结果如何被应用到SubscriptionState中。一、ConsumerCoordinator的家族谱系【ConsumerCoordinator 类继承关系】 AbstractCoordinator (抽象基类) │ ├─ heartbeat: Heartbeat ← 心跳辅助类 ├─ heartbeatTask: HeartbeatTask ← 心跳定时任务 ├─ groupId: String ├─ client: ConsumerNetworkClient ├─ coordinator: Node ← GroupCoordinator所在Broker ├─ memberId: String ← 服务端分配的唯一成员ID ├─ generation: int ← 年代号(每次Rebalance递增) ├─ needsJoinPrepare: boolean ← 是否需要在Join前做准备 ├─ rejoinNeeded: boolean ← 是否需要重新加入Group │ ├─ ensureActiveGroup() ← 确保Consumer在Group中活跃 ├─ pollHeartbeat() ← 发送心跳 └─ coordinatorUnknown() ← 检查Coordinator是否已知 │ ▼ (继承) ConsumerCoordinator │ ├─ assignors: ListPartitionAssignor ← 分区分配策略列表 ├─ metadata: Metadata ├─ subscriptions: SubscriptionState ├─ autoCommitEnabled: boolean ← 是否自动提交offset ├─ autoCommitTask: AutoCommitTask ← 自动提交定时任务 ├─ interceptors: ConsumerInterceptors ← 拦截器链 │ ├─ poll() ← 主协调方法 ├─ onJoinPrepare() ← JoinGroup前的准备钩子 ├─ onJoinComplete() ← JoinGroup完成后的处理钩子 ├─ commitOffsetsSync/Async() ← 提交offset └─ maybeAutoCommitOffsets() ← 自动提交offset二、AbstractCoordinator——协调器的公共底座2.1 核心字段publicabstractclassAbstractCoordinator{// 心跳管理protectedfinalHeartbeatheartbeat;// 心跳计时器privateHeartbeatTaskheartbeatTask;// 心跳定时任务// Group信息protectedfinalStringgroupId;// Consumer Group IDprotectedNodecoordinator;// GroupCoordinator的位置protectedStringmemberId;// 分配给当前消费者的成员IDprotectedintgeneration;// 年代号(Rebalance递增)// 网络通信protectedfinalConsumerNetworkClientclient;// 状态标志protectedbooleanneedsJoinPrepare;// 是否需要在Join前运行准备钩子protectedbooleanrejoinNeeded;// 是否需要重新Join}2.2 ensureActiveGroup()——协调核心入口这是ConsumerCoordinator中最重要的方法之一它在每次poll()中被调用确保消费者处于Group的活跃状态// AbstractCoordinator.ensureActiveGroup() 核心流程publicvoidensureActiveGroup(){// 1. 如果没有指定Coordinator先查找if(coordinatorUnknown()){lookupCoordinator();}// 2. 如果needsJoinPreparetrue执行准备操作if(needsJoinPrepare){onJoinPrepare(generation,memberId);needsJoinPreparefalse;}// 3. 发送JoinGroupRequestRequestFutureByteBufferfuturesendJoinGroupRequest();// 4. 等待JoinGroupResponse处理响应// (在onJoinComplete中处理)}状态流转【Consumer端Rebalance状态机】 ┌──────────┐ │ UNJOINED │ ← 初始状态 / 被踢出Group └────┬─────┘ │ needsJoinPreparetrue ▼ ┌──────────────────────┐ │ PREPARING_REBALANCE │ ← 准备阶段 │ (onJoinPrepare执行) │ ① 提交offset防止重复消费 └──────┬───────────────┘ ② 刷新本地缓存 │ │ sendJoinGroupRequest ▼ ┌──────────────────────┐ │ COMPLETING_REBALANCE │ ← 等待分配完成 │ (Leader计算分配结果) │ ① Leader: 执行PartitionAssignor.assign() └──────┬───────────────┘ ② 发送SyncGroupRequest │ │ SyncGroupResponse成功 ▼ ┌──────────┐ │ STABLE │ ← 正常消费中 └────┬─────┘ 定期发送心跳 │ │ 心跳失败/ILLEGAL_GENERATION ▼ 重新进入 UNJOINED 或 PREPARING_REBALANCE2.3 generation——年代号防乱序generation是防止过期请求干扰的关键设计【generation的作用示意】 Rebalance #1 (generation1): Consumer发送HeartbeatRequest(generation1) Rebalance #2 (generation2): ← 新一轮Rebalance ┌── generation递增为2 ──┐ │ │ │ 此时收到了Network延迟的请求... │ HeartbeatResponse(generation1) ← 来自上一轮的过期响应 │ │ Broker端检测: generation1 ≠ 当前generation2 │ 返回 ILLEGAL_GENERATION 错误 │ │ Consumer端处理: rejoinNeeded true │ Consumer重新发送JoinGroupRequest └──────────────────────────┘三、查找GroupCoordinator——“谁的场子”消费者在加入Group之前需要先找到管理本Consumer Group的GroupCoordinator。这个查找过程也叫FindCoordinator。【GroupCoordinator查找流程】 Consumer 任意Broker │ │ ├─FindCoordinatorRequest──────────────►│ │ groupId order-consumers │ │ ├─ 根据groupId计算hash │ │ hash(order-consumers) % 50 │ │ → 找到 __consumer_offsets 分区N │ │ → 分区N的Leader所在的Broker │ │ 就是GroupCoordinator │◄──FindCoordinatorResponse────────────┤ │ coordinator: Broker-3:9092 │ │ │ │ consumer.coordinator Broker-3 │ │ │ ├─后续所有请求都发给Broker-3──────────────►│ GroupCoordinator源码实现// AbstractCoordinator.lookupCoordinator()protectedsynchronizedvoidlookupCoordinator(){// 发送FindCoordinator请求到负载最低的BrokerNodenodeclient.leastLoadedNode();RequestFutureVoidfuturesendFindCoordinatorRequest(node);// 等待响应client.poll(future);// 响应中的coordinator已经被设置到 this.coordinator 字段}GroupCoordinator分配规则Kafka根据Consumer Group的groupId计算hash然后映射到__consumer_offsets内部Topic的某个分区该分区的Leader所在的Broker就是这个Group的GroupCoordinator。一个Broker可以同时担任多个Consumer Group的Coordinator。四、JoinGroup/SyncGroup请求的发送与处理4.1 sendJoinGroupRequest()——“我要加入”// ConsumerCoordinator.sendJoinGroupRequest()privateRequestFutureByteBuffersendJoinGroupRequest(){// 构建 JoinGroupRequestJoinGroupRequestrequestnewJoinGroupRequest(groupId,// Consumer Group IDsessionTimeoutMs,// session超时generation.memberId,// 当前成员IDprotocolType(),// consumermetadata()// 订阅信息 支持的分配策略);// protocolMetadata 包含的信息:// {// version: 1,// subscription: [topic-A, topic-B], // 订阅的Topic列表// userData: // 自定义数据// }// 发送并处理响应returnclient.send(coordinator,ApiKeys.JOIN_GROUP,request).compose(newJoinGroupResponseHandler());}4.2 JoinGroupResponseHandler——处理JoinGroup响应// 处理JoinGroupResponseprivateclassJoinGroupResponseHandlerextendsRequestFutureAdapterClientResponse,ByteBuffer{OverridepublicvoidonSuccess(ClientResponseresp,RequestFutureByteBufferfuture){JoinGroupResponsejoinResponsenewJoinGroupResponse(resp.responseBody());// 处理错误handleJoinGroupResponseError(resp,future);// 保存Leader信息// Leader的响应中包含所有Member的订阅信息// Follower的响应只包含Leader的IDif(joinResponse.isLeader()){// 保存全量订阅信息用于后续分区分配subscriptions.groupSubscription(joinResponse.members().stream().flatMap(m-parseSubscriptions(m).stream()).collect(Collectors.toSet()));// 执行分区分配 ← 只有Leader做这件事MapString,AssignmentassignmentsperformAssignment(joinResponse.leaderId(),joinResponse.groupProtocol(),// 选定的分配策略joinResponse.members());// 将分配结果封装到SyncGroupRequest中// ...}future.complete(joinResponse.memberAssignment());}}4.3 performAssignment()——分区分配计算// ConsumerCoordinator.performAssignment()// 只有Leader Consumer执行privateMapString,AssignmentperformAssignment(StringleaderId,StringgroupProtocol,// 选定的分区分配策略名ListJoinGroupResponseMemberallMembers){// 1. 按Topic收集分区数MapString,IntegerpartitionsPerTopicnewHashMap();for(Stringtopic:subscriptions.groupSubscription()){partitionsPerTopic.put(topic,metadata.fetch().partitionCountForTopic(topic));}// 2. 收集每个Member的订阅信息MapString,ListStringmemberSubscriptionsnewHashMap();for(JoinGroupResponseMembermember:allMembers){memberSubscriptions.put(member.memberId(),parseSubscriptions(member));}// 3. 找到对应名称的PartitionAssignorPartitionAssignorassignorfindAssignor(groupProtocol);// 4. 执行分区分配MapString,ListTopicPartitionrawResultassignor.assign(partitionsPerTopic,memberSubscriptions);// 5. 封装结果MapString,AssignmentresultnewHashMap();rawResult.forEach((memberId,partitions)-result.put(memberId,newAssignment(partitions.stream().map(tp-newTopicPartition(tp.topic(),tp.partition())).collect(Collectors.toList()))));returnresult;}4.4 onJoinComplete()——分配结果应用// ConsumerCoordinator.onJoinComplete()protectedvoidonJoinComplete(intgeneration,StringmemberId,StringassignmentStrategy,ByteBufferassignmentBuffer){// 1. 解析SyncGroupResponse中的分区分配结果AssignmentassignmentConsumerProtocol.deserializeAssignment(assignmentBuffer);// 2. 将分配结果应用到SubscriptionStatesubscriptions.assignFromSubscribed(assignment.partitions());// 3. 触发用户定义的ConsumerRebalanceListenerconsumerRebalanceListener.onPartitionsAssigned(assignment.partitions());// 4. 设置标志位this.needsJoinPreparetrue;// 下次Join前先执行准备钩子this.rejoinNeededfalse;// Join完成清标志}五、ConsumerCoordinator.poll()——协调器的主循环// ConsumerCoordinator.poll() 核心流程publicvoidpoll(longnow){// Step 1: 检查是否需要JoinGroupif(rejoinNeeded){ensureActiveGroup();// 触发JoinGroup → SyncGroup流程}// Step 2: 检查状态决定是否需要再次Join// (例如心跳失败、收到ILLEGAL_GENERATION等)checkState();// Step 3: 发送心跳如果在STABLE状态if(needToSendHeartbeat(now)){sendHeartbeatRequest();}// Step 4: 自动提交offsetif(autoCommitEnabled){maybeAutoCommitOffsetsAsync(now);}}本篇小结ConsumerCoordinator是消费者端最复杂的组件承担着与GroupCoordinator的外交职责查找Coordinator通过FindCoordinator请求定位管理本Group的Broker基于groupId的hash映射到__consumer_offsets分区JoinGroup协议向GroupCoordinator发送JoinGroupRequest由GroupCoordinator选举出Group Leader并指定分区分配策略分区分配Group Leader执行PartitionAssignor的计算逻辑只有Leader做然后将结果通过SyncGroupRequest上报结果应用所有Consumer收到SyncGroupResponse后通过onJoinComplete()钩子将分区更新到SubscriptionState并触发用户的自定义RebalanceListenergeneration防乱序通过递增的年代号防止过期请求对当前Rebalance造成干扰下一篇我们将聚焦心跳机制看看消费者是如何通过HeartbeatThread不断地向Broker报平安的。上一篇【第27篇】SubscriptionState源码解析——消费者是怎么记住自己订阅了什么下一篇【第29篇】Kafka心跳机制源码解析——消费者如何向Broker报平安
