码迷,mamicode.com
首页 > 其他好文 > 详细

Kafka中的再均衡

时间:2020-10-27 11:55:04      阅读:42      评论:0      收藏:0      [点我收藏+]

标签:res   alt   broker   事务   cep   散列   不必要   ali   分配   

在《Kafka消费者的使用和原理》中已经提到过“再均衡”的概念,我们先回顾下,一个主题可以有多个分区,而订阅该主题的消费组中可以有多个消费者。每一个分区只能被消费组中的一个消费者消费,可认为每个分区的消费权只属于消费组中的一个消费者。但是世界是变化的,例如消费者会宕机,还有新的消费者会加入,而为了应对这些变化,让分区所属权的分配合理,这都需要对分区所属权进行调整,也就是所谓的“再均衡”。本文将对再均衡的相关知识进行详细叙述。
技术图片

触发时机

首先,我们需要了解什么情况下会触发再均衡,在前文已经提到了消费者数量的变化,是需要再均衡的。在使用Kafka时,除了消费者数量可能会变化,分区数量也同样可能变化,我们可以人为的对分区数量进行修改,但是Kafka只允许增加分区,所以我们只能把分区数量调大,不能调小,否则会收到InvalidPartitionException异常。关于为什么不能减少分区,可参考下面的回答:

按Kafka现有的代码逻辑,此功能是完全可以实现的,不过也会使得代码的复杂度急剧增大。实现此功能需要考虑的因素很多,比如删除掉的分区中的消息该作何处理?如果随着分区一起消失则消息的可靠性得不到保障;如果需要保留则又需要考虑如何保留。直接存储到现有分区的尾部,消息的时间戳就不会递增,如此对于Spark、Flink这类需要消息时间戳(事件时间)的组件将会受到影响;如果分散插入到现有的分区中,那么在消息量很大的时候,内部的数据复制会占用很大的资源,而且在复制期间,此主题的可用性又如何得到保障?与此同时,顺序性问题、事务性问题、以及分区和副本的状态机切换问题都是不得不面对的。反观这个功能的收益点却是很低,如果真的需要实现此类的功能,完全可以重新创建一个分区数较小的主题,然后将现有主题中的消息按照既定的逻辑复制过去即可。

简单来说,就是做这个功能需要考虑很多因素,这样会把代码弄的很复杂,而收益却很低,而且存在替代方案来实现该效果,创建一个分区数小的主题,再把当前主题迁移过去。

除了消费者、分区数量的变化,还有一种情况,也需要进行再均衡。当消费者订阅主题时使用的是正则表达式,例如“test.*”,表示订阅所有以test开头的主题,当有新的以test开头的主题被创建时,则需要通过再均衡将该主题的分区分配给消费者。

再均衡的三种触发时机,我们已经清楚了,下面我们看下再均衡是如何实现的。

协调者

再均衡,将分区所属权分配给消费者。因此需要和所有消费者通信,这就需要引进一个协调者的概念,由协调者为消费组服务,为消费者们做好协调工作。在Kafka中,每一台Broker上都有一个协调者组件,负责组成员管理、再均衡和提交位移管理等工作。如果有N台Broker,那就有N个协调者组件,而一个消费组只需一个协调者进行服务,那该由哪个Broker为其服务?确定Broker需要两步:

  1. 计算分区号
    partition = Math.abs(groupID.hashCode() % offsetsTopicPartitionCount)

根据groupID的哈希值,取余offsetsTopicPartitionCount(内部主题consumer_offsets的分区数,默认50)的绝对值,其意思就是把消费组哈希散列到内部主题__consumer_offsets的一个分区上。确定协调者为什么要和内部主题扯上关系。这就跟协调者的作用有关了。协调者不仅是负责组成员管理和再均衡,在协调者中还需要负责处理消费者的偏移量提交,而偏移量提交则正是提交到consumer_offsets的一个分区上。所以这里需要取余offsetsTopicPartitionCount来确定偏移量提交的分区。

  1. 找出分区Leader副本所在的Broker

确定了分区就简单了,分区Leader副本所在的Broker上的协调者,就是我们要找的。

这个算法通常用于帮助定位问题。当一个消费组出现问题时,我们可以先确定协调者的Broker,然后查看Broker端的日志来定位问题。

交互方式

协调者,我们确定了。那协调者和消费者之间是如何交互的?协调者如何掌握消费者的状态,又如何通知再均衡。这里使用了心跳机制。在消费者端有一个专门的心跳线程负责以heartbeat.interval.ms的间隔频率发送心跳给协调者,告诉协调者自己还活着。同时协调者会返回一个响应。而当需要开始再均衡时,协调者则会在响应中加入REBALANCE_IN_PROGRESS,当消费者收到响应时,便能知道再均衡要开始了。

由于再平衡的开始依赖于心跳的响应,所以heartbeat.interval.ms除了决定心跳的频率,也决定了再均衡的通知频率。
现在我们再重新看下,触发再均衡的时机,前面说到有三种情况触发再均衡,分别是消费者数量的增加或减少、分区数的增加和新创建主题,其中消费者数量增加、分区数增加和新创建主题,这都是必须是人为操作,算是计划内的再均衡。而消费者数量减少则除了是人为操作,也可能因为其他原因导致,属于计划之外的再均衡,这是我们需要关心的,毕竟再均衡的开销还是很大的,所有消费者都会停止工作,所以我们应尽量避免不必要的再均衡。下面我们看下影响消费者数量减少的参数有哪些:

  1. session.timeout.ms:Broker端参数,消费者的存活时间,默认10秒,如果在这段时间内,协调者没收到任何心跳,则认为该消费者已崩溃离组;
  2. heartbeat.interval.ms:消费者端参数,发送心跳的频率,默认3秒;
  3. max.poll.interval.ms:消费者端参数,两次调用poll的最大时间间隔,默认5分钟,如果5分钟内无法消费完,则会主动离组。

可以看出session.timeout.ms和heartbeat.interval.ms是相关的,这里给出一个建议参考的公式:

session.timeout.ms ≥ 3 * heartbeat.interval.ms

为尽量避免因为偶发的网络原因,心跳无法到达协调者,在超时之前,应至少能发送3轮心跳。再给出一个经验值的设置:session.timeout.ms=6s,heartbeat.interval.ms=2s。

max.poll.interval.ms的设置,则主要和下游处理时间有关,例如下游处理时间需要6分钟,那按默认值是不合理的,消费者会频繁主动离组。所以需要把值设置的比下游处理时间大一点,避免不必要的再均衡。

这一小节主要讲了协调者如何通知消费者开始再均衡,以及如何设置参数避免不必要的再均衡,下面我们看下再均衡的流程是怎么样的。

流程

  1. 当消费者收到协调者的再均衡开始通知时,需要立即提交偏移量;
  2. 消费者在收到提交偏移量成功的响应后,再发送JoinGroup请求,重新申请加入组,请求中会含有订阅的主题信息;
  3. 当协调者收到第一个JoinGroup请求时,会把发出请求的消费者指定为Leader消费者,同时等待rebalance.timeout.ms,在收集其他消费者的JoinGroup请求中的订阅信息后,将订阅信息放在JoinGroup响应中发送给Leader消费者,并告知他成为了Leader,同时也会发送成功入组的JoinGroup响应给其他消费者;
  4. Leader消费者收到JoinGroup响应后,根据消费者的订阅信息制定分配方案,把方案放在SyncGroup请求中,发送给协调者。普通消费者在收到响应后,则直接发送SyncGroup请求,等待Leader的分配方案;
  5. 协调者收到分配方案后,再通过SyncGroup响应把分配方案发给所有消费组。
  6. 当所有消费者收到分配方案后,就意味着再均衡的结束,可以正常开始消费工作了。
    技术图片

Kafka中的再均衡

标签:res   alt   broker   事务   cep   散列   不必要   ali   分配   

原文地址:https://blog.51cto.com/9167833/2544236

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!