1. 什么是kafka的Rebalance

kafka集群模式下，一个topic有多个partition，对于消费端，可以有多个consumer同时消费这些partition。为了保证大体上partition和consumer的均衡性，提升topic的并发消费能力，所以会有Rebalance。Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 consumer 如何达成一致，来分配订阅 Topic 的每个分区。

2. 什么时机触发Rebalance

0.10kafka的rebalance条件

条件1：有新的consumer加入
条件2：旧的consumer挂了
条件3：coordinator挂了，集群选举出新的coordinator（0.10 特有的）
条件4：topic的partition新加
条件5：consumer调用unsubscrible()，取消topic的订阅

当一个group中,有consumer加入或者离开时,会触发partitions均衡。Kafka的Consumer Rebalance方案是基于Zookeeper的Watcher来实现的。consumer启动的时候，在zk下都维护一个”/consumers/[group_name]/ids”路径，在此路径下，使用临时节点记录属于此cg的消费者的Id，该Id信息由对应的consumer在启动时创建。每个consumer都会在此路径下简历一个watcher，当有节点发生变化时，就会触发watcher，然后触发Rebalance过程。

3. 0.9之前kafka的Rebalance算法

Consumer rebalacne算法：

将目标 topic 下的所有 partirtion 排序，存于PT
对某 consumer group 下所有 consumer 排序，存于 CG，第 i 个consumer 记为 Ci
N=size(PT)/size(CG)，向上取整
解除 Ci 对原来分配的 partition 的消费权（i从0开始）
将第i*N到（i+1）*N-1个 partition 分配给 Ci

在Rebalance期间，consumer不能正常消费，并且这种Rebalance过程强依赖zk，存在以下问题：

herd effect（羊群效应）：一个被Watch的zk节点变化，导致大量的watcher通知需要被发送给客户端，这会导致在通知期间其他操作的延迟。
split brain：每个Consumer都是通过zk中保存的元数据来判断group中各其他成员的状态，以及broker的状态，进而分别进入各自的Rebalance，执行各自的Rebalance逻辑。不同的Consumer在同一时刻可能连接在不同的zk服务器上，看到的元数据就可能不一样，基于不一样的元数据，执行Rebalance就会产生不一致（冲突）的Rebalance结果，Rebalance的冲突，会到导致consumer的rebalance失败。
重复消费问题：因为Rebalance时，很有可能导致offset commit不成功，所以可能造成重复消费问题。

解决办法：

加大Rebalance的重试时间:"rebalance.backoff.ms=5000"
加大Rebalance失败的retry次数: "rebalance.max.retries=10"
捕获"ConsumerRebalanceFailedException"，退出程序。

4. 0.9后kafka对Rebalance过程进行了改进

Group Coordinator是一个服务，每个Broker在启动的时候都会启动一个该服务。Group Coordinator的作用是用来存储Group的相关Meta信息，并将对应Partition的Offset信息记录到Kafka内置Topic(__consumer_offsets)中。Kafka在0.9之前是基于Zookeeper来存储Partition的Offset信息(consumers/{group}/offsets/{topic}/{partition})，因为ZK并不适用于频繁的写操作，所以在0.9之后通过内置Topic的方式来记录对应Partition的Offset。

每个Group都会选择一个Coordinator来完成自己组内各Partition的Offset信息。那么consumer group如何确定自己的coordinator是谁呢？简单来说分为两步：

确定consumer group位移信息写入__consumers_offsets的哪个分区。具体计算公式：__consumers_offsets partition# = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount)
该分区leader所在的broker就是被选定的coordinator。

注意：groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定，默认是50个分区。前面说过， rebalance本质上是一组协议。group与coordinator共同使用它来完成group的rebalance。目前kafka提供了5个协议来处理与consumer group coordination相关的问题：

Heartbeat请求：consumer需要定期给coordinator发送心跳来表明自己还活着
LeaveGroup请求：主动告诉coordinator我要离开consumer group
SyncGroup请求：group leader把分配方案告诉组内所有成员
JoinGroup请求：成员请求加入组
DescribeGroup请求：显示组的所有信息，包括成员信息，协议名称，分配方案，订阅信息等

rebalance过程分为2步：Join和Sync

Join，顾名思义就是加入组。这一步中，所有成员都向coordinator发送JoinGroup请求，请求入组。一旦所有成员都发送了JoinGroup请求，coordinator会从中选择一个consumer担任leader的角色，并把组成员信息以及订阅信息发给leader——注意leader和coordinator不是一个概念。leader负责消费分配方案的制定。
Sync，这一步leader开始分配消费方案，即哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案封装进SyncGroup请求中发给coordinator，非leader也会发SyncGroup请求，只是内容为空。coordinator接收到分配方案之后会把方案塞进SyncGroup的response中发给各个consumer。这样组内的所有成员就都知道自己应该消费哪些分区了。

5. 如何避免不必要的Rebalance

除开consumer正常的添加和停掉导致rebalance外，在某些情况下，Consumer 实例会被 Coordinator 错误地认为 “已停止” 从而被“踢出”Group，导致rebalance，这种情况应该避免。

第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被 “踢出”Group 而引发的。这种情况下我们可以设置 session.timeout.ms 和 heartbeat.interval.ms 的值，来尽量避免rebalance的出现。（以下的配置是在网上找到的最佳实践，暂时还没测试过）

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

将 session.timeout.ms 设置成 6s 主要是为了让 Coordinator 能够更快地定位已经挂掉的 Consumer，早日把它们踢出 Group。

第二类非必要 Rebalance 是 Consumer 消费时间过长导致的。此时，max.poll.interval.ms 参数值的设置显得尤为关键。如果要避免非预期的 Rebalance，你最好将该参数值设置得大一点，比你的下游最大处理时间稍长一点。

kafka的Rebalance问题分析(续)相关推荐

【kafka】Kafka 之 Group 状态变化分析及 Rebalance 过程
文章目录 1.概述 2. Group 状态机 3.offset 那些事 4.Topic __consumer_offsets 5.GroupCoordinator 6.状态转移图 7.Consumer ...
Kafka触发Rebalance的场景分析
文章目录前言触发Rebalance的原因 1. 消费者成员发生变化 2. 分区数发生变化 3. 订阅Topic发生变化 Rebalance全流程介绍场景一:新成员入组场景二:成员主动离组场景 ...
跟我学Kafka源码Producer分析
2019独角兽企业重金招聘Python工程师标准>>> 跟我学Kafka源码Producer分析博客分类: MQ 本章主要讲解分析Kafka的Producer的业务逻辑,分发逻辑和 ...
【Kafka】Kafka的Rebalance机制可能造成的影响及解决方案
一.kafka的rebalance机制在Kafka中,当有新消费者加入或者订阅的Topic数发生变化时,会触发Rebalance(再均衡:在同一个消费者组当中,分区的所有权从一个消费者转移到另外一个 ...
elk + kafka 简单搭建日志分析系统
elk + kafka 简单搭建日志分析系统文介绍使用ELK(elasticsearch.logstash.kibana) + kafka来搭建一个日志系统. 想象一下淘宝,它可以对不同的用户实现推 ...
Kafka Consumer Rebalance详解
全网最全大数据面试提升手册! 文章目录 Kafka版本 rebalance rebalance策略 rebalance generation rebalance协议 rebalance流程 rebal ...
kafka一直rebalance故障，重复消费
今天我司线上kafka消息代理出现错误日志,异常rebalance,而且平均间隔2到3分钟就会rebalance一次,分析日志发现比较严重.错误日志如下 08-09 11:01:11 131 pool ...
【kafka】记一次线上kafka一直rebalance故障消费慢数据积压
文章目录 1.背景 2. 分析问题 3.分析原因 4.拉取偏移量与提交偏移量 5.解决方案 5.1.增加max.poll.interval.ms处理时长 5.2设置分区拉取阈值 5.3.poll到的消 ...
kafka原理和性能分析测试
1.Kafka写数据流程: producer先从zookeeper的broker-list的节点找到partition(分区)的leader: producer将消息发送给该leader的partit ...

kafka的Rebalance问题分析(续)