Kafka消费者组

发布于:2024-06-29 ⋅ 阅读:(36) ⋅ 点赞:(0)

消费者组(Consumer Group)

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

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

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

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

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

重平衡(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的发生。

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会计算该Groupgroup.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

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)


网站公告

今日签到

点亮在社区的每一天
去签到