Kafka消费者组

__consumer_offsets

说消费组之前先说一下，__consumer_offsets；

__consumer_offsets是一个特殊的内部主题，用于存储消费者组的偏移量信息。这个主题是由Kafka自动管理和维护的，用于跟踪消费者组内每个分区的最后读取位置，这样当消费者重启或者发生故障时，可以从之前的位置继续消费消息。以下是关于__consumer_offsets的一些关键点：

        存储偏移量：每当消费者提交其偏移量时，这些信息就被存储在__consumer_offsets主题中。偏移量代表了消费者在特定分区的读取位置。

        消费者组：偏移量与消费者组关联。这意味着即使组内的单个消费者实例失败，只要组存在，偏移量信息就不会丢失，新的消费者实例可以接管并从上次提交的位置继续消费。

        高可用性：由于偏移量存储在Kafka集群中，即使Zookeeper或某个Broker不可用，偏移量信息仍然可恢复，这提高了系统的容错性和可用性。

        自动创建：__consumer_offsets主题是自动创建的，不需要用户显式定义。它通常有三个副本以确保数据的持久性和可用性。

        数据结构：__consumer_offsets主题的消息键通常是{group_id}-{topic_name}-{partition_id}的格式，而消息值则是偏移量的数值。

        管理工具：Kafka提供了一些管理工具，如kafka_consumer_groups.sh脚本，允许用户检查和管理__consumer_offsets中的信息。

在较老的Kafka版本中，消费者偏移量是存储在Zookeeper中的，但在新版本中，推荐使用__consumer_offsets主题来存储这些信息，以减少对外部系统的依赖，提高性能和可靠性。

消费者组（Consumer Group）

Consumer Group是Kafka提供的可扩展且具有容错性的消费者机制。

消费者组内可以有多个消费者或消费者实例（Consumer Instance），它们共享一个公共的ID，这个ID被称为Group ID。组内的所有消费者协调在一起来消费订阅主题（Subscribed Topics）的所有分区 （Partition）。当然，每个分区只能由同一个消费者组内的一个Consumer实例来消费。

当Consumer Group订阅了多个主题后，组内的每个实例不要求一定要订阅主题的所有分区，它只会消费部分分区中的消息。

Consumer Group之间彼此独立，互不影响，它们能够订阅相同的一组主题而互不干涉。如果所有实例都属于同一个Group，那么它实现的就是消息队列模型；如果所有实例分别属于不同的 Group，那么它实现的就是发布/订阅模型。

建议：Consumer实例的数量应该等于该 Group订阅主题的分区总数。

Coordinator

在Kafka中有个Coordinator，它专门为Consumer Group服务，负责为Group执行Rebalance以及提供位移管理和组成员管理等，所有Broker都有各自的Coordinator组件。

Kafka为某个Consumer Group确定Coordinator所在的Broker的算法有2个步骤。

        第1步：确定由位移主题的哪个分区来保存该Group数据：

partitionId=Math.abs(groupId.hashCode() %offsetsTopicPartitionCount)。

        第2步：找出该分区Leader副本所在的Broker，该Broker即为对应的Coordinator。

首先，Kafka会计算该Group的group.id参数的哈希值。比如你有个Group的group.id设置成了“test-group”，那么它的hashCode值就应该是627841412。其次，Kafka会计算 __consumer_offsets的分区数，通常是50个分区，之后将刚才那个哈希值对分区数进行取模加求绝对值计算，即abs(627841412 %50) = 12。此时，我们就知道了位移主题的分区12负责保存这个Group的数据。有了分区号，算法的第2步就变得很简单了，我们只需要找出位移主题分区12的Leader副本在哪个Broker上就可以了。这个Broker，就是我们要找的Coordinator。

重平衡（Rebalance）

Rebalance本质上是一种协议，规定了一个Consumer Group下的所有Consumer如何达成一致，来分配订阅Topic的每个分区。比如某个Group下有20个Consumer实例，它订阅了一个具有100个分区的 Topic。正常情况下，Kafka平均会为每个Consumer分配5个分区。这个分配的过程就叫Rebalance。

Rebalance的触发条件有3个：

        1. 组成员数发生变更。比如有新的Consumer实例加入组或者离开组，抑或是有Consumer实例崩溃被“踢出”组。

        2. 订阅主题数发生变更。Consumer Group可以使用正则表达式的方式订阅主题，比如consumer.subscribe(Pattern.compile(“t.*c”))就表明该Group订阅所有以字母t开头、字母c结尾的主题。在Consumer Group的 运行过程中，你新创建了一个满足这样条件的主题，那么该Group就会发生Rebalance。

        3. 订阅主题的分区数发生变更。Kafka当前只能允许增加一个主题的分区数。当分区数增加时，就会触发订阅该主题的所有Group开启Rebalance。

假设目前某个Consumer Group下有两个Consumer，比如A和B，当第三个成员C加入时，Kafka会触发Rebalance，并根据默认的分配策略重新为A、B和C分配分区，如下图所示：

缺点：在Rebalance过程中所有Consumer实例都会停止消费，等待Rebalance完成。 类似于著名的stop the world，简称STW。 最好避免Rebalance的发生。

重平衡的通知机制

重平衡的通知机制是靠消费者端的心跳线程（Heartbeat Thread）完成的。

当协调者决定开启新一轮重平衡后，它会将“REBALANCE_IN_PROGRESS”封装进心跳请求的响应中，发还给消费者实例。当消费者实例发现心跳响应中包含了“REBALANCE_IN_PROGRESS”，就能立马知道重平衡又开始了，这就是重平衡的通知机制。

消费者端参数heartbeat.interval.ms，设置了心跳的间隔时间，作用是控制重平衡通知的频率。如果想要消费者实例更迅速地得到通知，那么就可以给这个参数设置一个非常小的值。

消费者组状态机

Kafka设计了一套消费者组状态机（State Machine），来帮助协调者完成整个重平衡流程。它们分别是：Empty、Dead、PreparingRebalance、CompletingRebalance和Stable。

消费者组启动时的状态流转过程：

一个消费者组最开始是Empty状态，当重平衡过程开启后，它会被置于PreparingRebalance状态等待成员加入，之后变更到CompletingRebalance状态等待分配方 案，最后流转到Stable状态完成重平衡。 当有新成员加入或已有成员退出时，消费者组的状态从Stable直接跳到PreparingRebalance状态，此时，所有现存成员就必须重新申请加入组。当所有成员都退出组后，消费者组状态变更为Empty。Kafka定 期自动删除过期位移的条件就是，组要处于Empty状态。因此，如果你的消费者组停掉了很长时间（超过7天），那么Kafka很可能就把该组的位移数据删除了。 

消费者端重平衡流程

1. 加入组 - JoinGroup请求

        当组内成员加入组，向协调者发送JoinGroup请求，将自己订阅的主题上报，协调者就能收集到所有成员的订阅信息，从中选择一个消费者组的领导者。选出领导者之后，协调者会把消费者组订阅信息封装进JoinGroup请求的响应体中，然后发给领导者，由领导者统一做出分配方案后，进入到下一步发送SyncGroup请求。

2. 等待领导者消费者（Leader Consumer）- SyncGroup请求

        SyncGroup请求中，领导者向协调者发送SyncGroup请求，将刚刚做出的分配方案发给协调者。值得注意的是，其他成员也会向协调者发送SyncGroup请求，只不过请求体中并没有实际的内容。这一步的主要目的是让协调者接收分配方案，然后统一以SyncGroup响应的方式分发给所有成员，这样组内所有成员就都知道自己该消费哪些分区了。

3. 组成员主动离组 - LeaveGroup请求

        消费者实例所在线程或进程调用close()方法主动通知协调者它要退出。发送LeaveGroup请求，协调者收到LeaveGroup请求后，依然会以心跳响应的方式通知其他成员。

4. 组成员崩溃离组

        崩溃离组是指消费者实例出现严重故障，突然宕机导致的离组，不是必然发生，一般是由消费者端参数session.timeout.ms控制的。也就是说，Kafka一般不会超过session.timeout.ms就能感知到这个崩溃。当然，后面处理崩溃离组的流程也是以心跳响应的方式通知其他成员。

5. 重平衡时协调者对组内成员提交位移的处理。

        正常情况下，每个组内成员都会定期汇报位移给协调者。当重平衡开启时，协调者会给予成员一段缓冲时间，要求每个成员必须在这段时间内快速地上报自己的位移信息，然后再开启正常的 JoinGroup/SyncGroup请求发送。

避免Rebalance

Consumer Group下的Consumer实例数量发生变化，是Rebalance发生的最常见的原因。

Coordinator会在什么情况下认为某个Consumer实例已挂从而要退组呢？

        1. Consumer端参数，session.timeout.ms，就是被用来表征此事的。该参数的默认值是10秒，即如果Coordinator在10秒之内没有收到Group 下某Consumer实例的心跳，它就会认为这个Consumer实例已经挂了。

        2. Consumer还提供了一个允许你控制发送心跳请求频率的参数，就是heartbeat.interval.ms。这个值设置得越小，Consumer实例发送心跳请求的频率就越高。频繁地发送心跳请求会额外消耗带宽资源，但好处是能够更加快速地知晓当前是否开启Rebalance，因为，目前Coordinator通知各个Consumer实例开启Rebalance的方法，就是将REBALANCE_NEEDED标志封装进心跳请求的响应体中。

        3. Consumer端还有一个参数，用于控制Consumer实际消费能力对Rebalance的影响，即max.poll.interval.ms参数。它限定了Consumer端应用程序两次调用poll方法的最大时间间隔。它的默认值是5分钟，表示你的Consumer程序如果在5分钟之内无法消费完poll方法返回的消息，那么Consumer会主动发起“离开组”的请求，Coordinator也会开启新一轮Rebalance。

哪些Rebalance是“不必要的”？

第一类非必要Rebalance是因为未能及时发送心跳，导致Consumer被“踢出”Group而引发的。因此，需要仔细地设置session.timeout.ms和heartbeat.interval.ms的值。生产环境中推荐设置：

       * session.timeout.ms = 6s。

       * heartbeat.interval.ms = 2s。

       * 要保证Consumer实例在被判定为“dead”之前，能够发送至少3轮的心跳请求，即session.timeout.ms >= 3 * heartbeat.interval.ms。

将session.timeout.ms设置成6s主要是为了让Coordinator能够更快地定位已经挂掉的Consumer。毕竟，我们还是希望能尽快揪出那些“尸位素餐”的Consumer，早日把它们踢出Group。

第二类非必要Rebalance是Consumer消费时间过长导致的。设置max.poll.interval.ms参数，最好将该参数值设置得大一点，避免非预期的Rebalance。

第三类 以上俩类都设置了，还有问题，排查一下Consumer端的GC表现，多因为GC设置不合理导致程序频发FullGC而引发的非预期Rebalance了。

尽量地降低生产环境中的Rebalance数量，从而整体提升Consumer端TPS。

参考：Kafka 核心技术与实战 (geekbang.org)