作者 ｜ vivo 互联网服务器团队-You Shuo

副本迁移是Kafka最高频的操作，对于一个拥有几十万个副本的集群，通过人工去完成副本迁移是一件很困难的事情。Cruise Control作为Kafka的运维工具，它包含了Kafka 服务上下线、集群内负载均衡、副本扩缩容、副本缺失修复以及节点降级等功能。显然，Cruise Control的出现，使得我们能够更容易的运维大规模Kafka集群。

备注：本文基于 Kafka 2.1.1开展。

一、  Kafka 负载均衡

1.1 生产者负载均衡

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

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

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

1.2 消费者负载均衡

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

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

• range: 在保证均衡的前提下，将连续的分区分配给消费者，对应的实现是RangeAssignor；
• round-robin：在保证均衡的前提下，轮询分配，对应的实现是RoundRobinAssignor；
• 0.11.0.0版本引入了一种新的分区分配策略StickyAssignor，其优势在于能够保证分区均衡的前提下尽量保持原有的分区分配结果，从而避免许多冗余的分区分配操作，减少分区再分配的执行时间。

无论是生产者还是消费者，Kafka 客户端内部已经帮我们做了负载均衡了，那我们还有讨论负载均衡的必要吗？答案是肯定的，因为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负载依然得不到减轻。

三、如何对 Kafka 做负载均衡

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切换的命令参考如下：

Kafka 副本迁移脚本

# 副本迁移脚本：kafka-reassign-partitions.sh
# 1. 配置迁移文件
$ vi topic-reassignment.json
{"version":1,"partitions":[
{"topic":"T0","partition":2,"replicas":[broker3,broker1]},
{"topic":"T0","partition":3,"replicas":[broker0,broker3]},
{"topic":"T0","partition":4,"replicas":[broker3,broker1]},
{"topic":"T1","partition":0,"replicas":[broker2,broker3]},
{"topic":"T1","partition":2,"replicas":[broker2,broker0]}
]}
# 2. 执行迁移命令
bin/kafka-reassign-partitions.sh --throttle 73400320 --zookeeper zkurl --execute --reassignment-json-file topic-reassignment.json
# 3. 查看迁移状态/清除限速配置
bin/kafka-reassign-partitions.sh --zookeeper zkurl --verify --reassignment-json-file topic-reassignment.json

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：

• Metrics Reporter实现了Kafka的指标上报接口MetricsReporter，以特定的格式将原生的Kafka指标上报到topic __CruiseControlMetrics中。

• Metrics Sampler从__CruiseControlMetrics中获取原生指标后按照broker和分区级指标分别进行聚合，聚合后的指标包含了broker、分区负载的均值、最大值等统计值，这些中间结果将被发送topic __KafkaCruiseControlModelTrainingSamples和__KafkaCruiseControlPartitionMetricSamples中；

（2）Analyzer

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

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

Analyzer可能需要改进的地方：

1. 由于Monitor生成的是整个集群的负载模型，我们的Kafka平台将Kafka集群划分为多个资源组，不同资源组的资源利用率存在很大差别，所以原生的集群负载模型不再适用于我们的应用场景。
2. 大多数业务没有指定key进行生产，所以各分区的负载偏差不大。如果topic分区副本均匀分布在资源组内，则资源组也随之变得均衡。
3. 原生的cruise control会从集群维度来展开均衡工作，指定资源组后可以从资源组维度展开均衡工作，但无法满足跨资源组迁移的场景。

（3）Executor

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

Executor可能需要改进的地方：

cruise control 在执行副本迁移类的功能时，不能触发集群首选leader切换：有时在集群均衡过程中出现了宕机重启，以问题机器作为首选leader的分区，其leader不能自动切换回来，造成集群内其他节点压力陡增，此时往往会产生连锁反应。

(4)Anomaly Detector

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

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

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

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

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

服务端在接收到均衡请求后，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可以做到如下几点：

1. 通过均衡参数，我们可以只均衡某个或多个资源组的broker。
2. 更改topic配置，比如增加topic副本时，新扩的副本需要和topic原先的副本在同一个资源组内。
3. 在资源组内分析broker上的资源是迁入还是迁出。对于每一类资源目标，cruise control是计算资源组范围内的统计指标，然后结合阈值和离散因子来分析broker是迁出资源还是迁入资源。

如下图所示，我们将集群、资源组、以及资源组下的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上迁移

考虑这些场景：

1. 一个项目下会有几个资源组，由于业务变更，业务想要把A资源组下的topic迁移到B资源组。
2. 业务想要把公共资源组的topic迁移到C资源组。
3. 均衡完成之后，发现总有几个topic/分区分布不是很均匀。

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

1. 只均衡指定的topic或topic分区；
2. 均衡的topic或topic分区只往指定的broker上迁移。

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

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

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

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

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

1. 新增一类资源分析：topic分区leader副本分散性。
2. 首先保证每个topic的leader副本和follower副本尽可能的均匀分布在资源组的broker上。
3. 在2的基础上，副本会尽可能的往负载较低的broker上分布。

如下图所示，针对每一个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，增加如下配置：

在Kafka服务上配置采集程序

# 修改kafka的server.properties
metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
cruise.control.metrics.reporter.bootstrap.servers=域名:9092
cruise.control.metrics.reporter.security.protocol=SASL_PLAINTEXT
cruise.control.metrics.reporter.sasl.mechanism=SCRAM-SHA-256
cruise.control.metrics.reporter.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"ys\" password=\"ys\";

【步骤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_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"ys\" password=\"ys\";
bootstrap.servers=域名:9092
zookeeper.connect=zkURL

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

# 集群id
cluster_id=xxx
db_url=jdbc:mysql://hostxxxx:3306/databasexxx
db_user=xxx
db_pwd=xxx

四、总结

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

1. Kafka每个partition replica与机器的磁盘绑定，partition replica由一系列的Segment组成，所以往往单分区存储会占用比较大的磁盘空间，对于磁盘会有很大压力。
2. 在集群扩容broker时必须做Rebalance，需要broker有良好的执行流程，保证没有任何故障的情况下使得各broker负载均衡。

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/