Kafka 负载均衡在 vivo 的落地实践

1.1 生产者负载均衡

Kafka 客户端可以使用分区器依据消息的key计算分区，如果在发送消息时未指定key，则默认分区器会基于round robin算法为每条消息分配分区；

否则会基于murmur2哈希算法计算key的哈希值，并与分区数取模的到最后的分区编号。

很显然，这并不是我们要讨论的Kafka负载均衡，因为生产者负载均衡看起来并不是那么的复杂。

1.2 消费者负载均衡

考虑到消费者上下线、topic分区数变更等情况，KafkaConsumer还需要负责与服务端交互执行分区再分配操作，以保证消费者能够更加均衡的消费topic分区，从而提升消费的性能；

Kafka目前主流的分区分配策略有2种(默认是range，可以通过partition.assignment.strategy参数指定)：

无论是生产者还是消费者，Kafka 客户端内部已经帮我们做了负载均衡了，那我们还有讨论负载均衡的必要吗？答案是肯定的，因为Kafka负载不均的主要问题存在于服务端而不是客户端。

我们先来看一下Kafka集群的流量分布（图1）以及新上线机器后集群的流量分布（图2）：

图1

图2

从图1可以看出资源组内各broker的流量分布并不是很均衡，而且由于部分topic分区集中分布在某几个broker上，当topic流量突增的时候，会出现只有部分broker流量突增。

这种情况下，我们就需要扩容topic分区或手动执行迁移动操作。

图2是我们Kafka集群的一个资源组扩容后的流量分布情况，流量无法自动的分摊到新扩容的节点上。此时，就需要我们手动的触发数据迁移，从而才能把流量引到新扩容的节点上。

2.1  Kafka 存储结构

为什么会出现上述的问题呢？这个就需要从Kafka的存储机制说起。

下图是Kafka topic的存储结构，其具体层级结构描述如下：

1. 每个broker节点可以通过logDirs配置项指定多个log目录，我们线上机器共有12块盘，每块盘都对应一个log目录。

   
   

2. 每个log目录下会有若干个[topic]-[x]字样的目录，该目录用于存储指定topic指定分区的数据，对应的如果该topic是3副本，那在集群的其他broker节点上会有两个和该目录同名的目录。

   
   

3. 客户端写入kafka的数据最终会按照时间顺序成对的生成.index、.timeindex、.snapshot以及.log文件，这些文件保存在对应的topic分区目录下。

   
   

4. 为了实现高可用目的，我们线上的topic一般都是2副本/3副本，topic分区的每个副本都分布在不同的broker节点上，有时为了降低机架故障带来的风险，topic分区的不同副本也会被要求分配在不同机架的broker节点上。

了解完Kafka存储机制之后，我们可以清晰的了解到，客户端写入Kafka的数据会按照topic分区被路由到broker的不同log目录下，只要我们不人工干预，那每次路由的结果都不会改变。因为每次路由结果都不会改变，那么问题来了：

随着topic数量不断增多，每个topic的分区数量又不一致，最终就会出现topic分区在Kafka集群内分配不均的情况。

比如：topic1是10个分区、topic2是15个分区、topic3是3个分区，我们集群有6台机器。那6台broker上总会有4台broker有两个topic1的分区，有3台broke上有3个topic3分区等等。

这样的问题就会导致分区多的broker上的出入流量可能要比其他broker上要高，如果要考虑同一topic不同分区流量不一致、不同topic流量又不一致，再加上我们线上有7000个topic、13万个分区、27万个副本等等这些。

这么复杂的情况下，集群内总会有broker负载特别高，有的broker负载特别低，当broker负载高到一定的时候，此时就需要我们的运维同学介入进来了，我们需要帮这些broker减减压，从而间接的提升集群总体的负载能力。

当集群整体负载都很高，业务流量会持续增长的时候，我们会往集群内扩机器。有些同学想扩机器是好事呀，这会有什么问题呢？问题和上面是一样的，因为发往topic分区的数据，其路由结果不会改变，如果没有人工干预的话，那新扩进来机器的流量就始终是0，集群内原来的broker负载依然得不到减轻。

3.1 人工生成迁移计划和迁移

如下图所示，我们模拟一个简单的场景，其中的T0-P0-R0表示topic-分区-副本，假设topic各分区流量相同，假设每个分区R0副本是leader。

我们可以看到，有两个topic T0和T1，T0是5分区2副本（出入流量为10和5），T1是3分区2副本（出入流量为5和1），如果严格考虑机架的话，那topic副本的分布可能如下：

假设我们现在新扩入一台broker3(Rack2)，如下图所示：由于之前考虑了topic在机架上的分布，所以从整体上看，broker2的负载要高一些。

我们现在想把broker2上的一些分区迁移到新扩进来的broker3上，综合考虑机架、流量、副本个数等因素，我们将T0-P2-R0、T0-P3-R1、T0-P4-R0、T1-P0-R1四个分区迁移到broker3上。

看起来还不是很均衡，我们再将T1-P2分区切换一下leader：

经历一番折腾后，整个集群就均衡许多了，关于上面迁移副本和leader切换的命令参考如下：

3.2 使用负载均衡工具-cruise control

经过对Kafka存储结构、人工干预topic分区分布等的了解后，我们可以看到Kafka运维起来是非常繁琐的，那有没有一些工具可以帮助我们解决这些问题呢？

答案是肯定的。

cruise control是LinkedIn针对Kafka集群运维困难问题而开发的一个项目，cruise control能够对Kafka集群各种资源进行动态负载均衡，这些资源包括：CPU、磁盘使用率、入流量、出流量、副本分布等，同时cruise control也具有首选leader切换和topic配置变更等功能。

3.2.1 cruise cotnrol 架构

我们先简单介绍下cruise control的架构。

如下图所示，其主要由Monitor、Analyzer、Executor和Anomaly Detector 四部分组成：

（来源：cruise control 官网）

（1）Monitor

Monitor分为客户端Metrics Reporter和服务端Metrics Sampler：

（2）Analyzer

Analyzer作为cruise control的核心部分，它根据用户提供的优化目标和基于Monitor生成的集群负载模型生成迁移计划。

在cruise control中，“用户提供的优化目标”包括硬性目标和软性目标两大类，硬性目标是Analyzer在做预迁移的时候必须满足的一类目标（例如：副本在迁移后必须满足机架分散性原则），软性目标则是尽可能要达到的目标，如果某一副本在迁移后只能同时满足硬性目标和软性目标中的一类，则以硬性目标为主，如果存在硬性目标无法满足的情况则本次分析失败。

Analyzer可能需要改进的地方：

（3）Executor

Executor作为一个执行者，它执行Analyzer分析得到的迁移计划。它会将迁移计划以接口的形式分批提交到Kafka集群上，后续Kafka会按照提交上来的迁移脚本执行副本迁移。

Executor可能需要改进的地方：

（4）Anomaly Detector

Anomaly Detector是一个定时任务，它会定期检测Kafka集群是否不均衡或者是否有副本缺失这些异常情况，当Kafka集群出现这些情况后，Anomaly Detector会自动触发一次集群内的负载均衡。

在后面的主要功能描述中，我会主要介绍Monitor和Analyzer的处理逻辑。

3.2.2 均衡 broker 出入流量 / 机器上下线均衡

对于Kafka集群内各broker之间流量负载不均的原因、示意图以及解决方案，我们在上面已经介绍过了，那么cruise control是如何解决这个问题的。

其实cruise control均衡集群的思路和我们手动去均衡集群的思路大体一致，只不过它需要Kafka集群详细的指标数据，以这些指标为基础，去计算各broker之间的负载差距，并根据我们关注的资源去做分析，从而得出最终的迁移计划。

以topic分区leader副本这类资源为例：

服务端在接收到均衡请求后，Monitor会先根据缓存的集群指标数据构建一个能够描述整个集群负载分布的模型。

下图简单描述了整个集群负载信息的生成过程，smaple fetcher线程会将获取到的原生指标加载成可读性更好的Metric Sample，并对其进行进一步的加工，得到带有brokerid、partition分区等信息的统计指标，这些指标保存在对应broker、replica的load属性中，所以broker和repilca会包含流量负载、存储大小、当前副本是否是leader等信息。

Analyzer 会遍历我们指定的broker（默认是集群所有的broker），由于每台broker及其下面的topic分区副本都有详细的指标信息，分析算法直接根据这些指标和指定资源对broker进行排序。

本例子的资源就是topic分区leader副本数量，接着Analyzer会根据我们提前设置的最大/最小阈值、离散因子等来判断当前broker上某topic的leader副本数量是否需要增加或缩减，如果是增加，则变更clustermodel将负载比较高的broker上对应的topic leader副本迁移到当前broker上，反之亦然，在后面的改造点中，我们会对Analyzer的工作过程做简单的描述。

遍历过所有broker，并且针对我们指定的所有资源都进行分析之后，就得出了最终版的clustermodel，再与我们最初生成的clustermodel对比，就生成了topic迁移计划。

cruise control会根据我们指定的迁移策略分批次的将topic迁移计划提交给kafka集群执行。

迁移计划示意图如下：

3.2.3 首选 leader 切换

切换非首选leader副本，迁移计划示意图如下：

3.2.4 topic配置变更

改变topic副本个数，迁移计划示意图如下：

3.3  改造 cruise control

3.3.1 指定资源组进行均衡

当集群规模非常庞大的时候，我们想要均衡整个集群就变得非常困难，往往均衡一次就需要半个月甚至更长时间，这在无形之中也加大了我们运维同学的压力。

针对这个场景，我们对cruise control也进行了改造，我们从逻辑上将Kafka集群划分成多个资源组，使得业务拥有自己的资源组，当某个业务出现流量波动的时候，不会影响到其他的业务。

通过指定资源组，我们每次只需要对集群的一小部分或多个部分进行均衡即可，大大缩短了均衡的时间，使得均衡的过程更加可控。

改造后的cruise control可以做到如下几点：

如下图所示，我们将集群、资源组、以及资源组下的topic这些元数据保存在数据库中，Analyzer得以在指定的资源组范围内，对每个broker按照资源分布目标做均衡分析。

例如：当对broker-0做均衡分析的时候，Analyzer会遍历goals列表，每个goals负责一类资源负载目标（cpu、入流量等），当均衡分析到达goal-0的时候，goal-0会判断broker-0的负载是否超出上限阈值，如果超出，则需要将broker-0的一些topic副本迁移到负载较低的broker上；反之，则需要将其他broker上的副本迁移到broker-0上。

其中，下面的recheck goals是排在后面的goal在做均衡分析的时候，在更新cluster model之前会判断本次迁移会不会与之前的goal冲突，如果冲突，那就不更新cluster model，当前的goal会继续尝试往其他broker上迁移，直到找到适合的迁移目标，然后更新cluster model。

3.3.2 topic/topic分区往指定broker上迁移

考虑这些场景：

面对这些场景，我们上面指定资源组进行均衡的功能就满足不了我们的需求了。所以，我们针对上述场景改造后的cruise control可以做到如下几点：

3.3.3 新增目标分析——topic分区leader副本分散性

业务方大多都是没有指定key进行发送数据的，所以同一topic每个分区的流量、存储都是接近的，即每一个topic的各个分区的leader副本尽可能均匀的分布在集群的broker上时，那集群的负载就会很均匀。

有同学会问了，topic分区数并不总是能够整除broker数量，那最后各broker的负载不还是不一致嘛？

答案是肯定的，只通过分区的leader副本还不能做到最终的均衡。

针对上述场景改造后的cruise control可以做到如下几点：

如下图所示，针对每一个topic的副本，Analyzer会依次计算当前broker的topic leader数是否超过阈值上限，如果超过，则Analyzer会按照topic的leader副本数量、topic的follower副本数量、broker的出流量负载等来选出AR中的follower副本作为新的leader进行切换，如果AR副本中也没有符合要求的broker，则会选择AR列表以外的broker。

3.3.4 最终均衡效果

下图是某个资源组均衡后的流量分布，各节点间流量偏差非常小，这种情况下，既可以增强集群扛住流量异常突增的能力又可以提升集群整体资源利用率和服务稳定性，降低成本。

3.4 安装/部署cruise control

3.4.1 客户端部署：指标采集

【步骤1】：创建Kafka账号，用于后面生产和消费指标数据

【步骤2】：创建3个Kafka内部topic：a是用来存储Kafka服务原生jmx指标；b和c分别是用来存储cruise control处理过后的分区和模型指标；

【步骤3】：给步骤1创建的账号授予读/写以及集群的操作权限，用于读/写步骤2创建的topic；

【步骤4】：修改kafka的server.properties，增加如下配置：

【步骤5】：添加cruise-control-metrics-reporter的jar包到Kafka的lib目录下：mv cruise-control-metrics-reporter-2.0.104-SNAPSHOT.jar kafka_dir/lib/；

【步骤6】：重启Kafka服务。

3.4.2 服务端部署：指标聚合/均衡分析

（1）到https://github.com/linkedin/cruise-control 下载zip文件并解压；

（2）将自己本地cruise control子模块下生成的jar包替换cruise control的：

mv cruise-control-2.0.xxx-SNAPSHOT.jar 

cruise-control/build/libs；

（3）修改cruise control配置文件，主要关注如下配置：

# 修改cruise control配置文件security.protocol=SASL_PLAINTEXTsasl.mechanism=SCRAM-SHA-256sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"ys\" password=\"ys\";bootstrap.servers=域名:9092zookeeper.connect=zkURL

（4）修改数据库连接配置：

# 集群idcluster_id=xxx  db_url=jdbc:mysql://hostxxxx:3306/databasexxxdb_user=xxxdb_pwd=xxx

通过以上的介绍，我们可以看出Kafka存在比较明显的两个缺陷：

cruise control就是针对Kafka集群运维困难问题而诞生的，它能够很好的解决kafka运维困难的问题。

参考文章：

1. linkedIn/cruise-control
2. Introduction to Kafka Cruise Control
3. Cloudera Cruise Control REST API Reference

4. http://dockone.io/article/2434664

5. https://www.zhenchao.org/2019/06/22/kafka/kafka-log-manage/