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可以看出資源組內各broker的流量分佈並不是很均衡，而且由於部分topic分割區集中分佈在某幾個broker上，當topic流量突增的時候，會出現只有部分broker流量突增。

這種情況下，我們就需要擴容topic分割區或手動執行遷移動操作。

圖2是我們Kafka叢集的一個資源組擴容後的流量分佈情況，流量無法自動的分攤到新擴容的節點上。此時，就需要我們手動的觸發資料遷移，從而才能把流量引到新擴容的節點上。

2.1 Kafka 儲存結構

為什麼會出現上述的問題呢？這個就需要從Kafka的儲存機制說起。

下圖是Kafka topic的儲存結構，其具體層級結構描述如下：

每個broker節點可以通過logDirs設定項指定多個log目錄，我們線上機器共有12塊盤，每塊盤都對應一個log目錄。
每個log目錄下會有若干個[topic]-[x]字樣的目錄，該目錄用於儲存指定topic指定分割區的資料，對應的如果該topic是3副本，那在叢集的其他broker節點上會有兩個和該目錄同名的目錄。
使用者端寫入kafka的資料最終會按照時間順序成對的生成.index、.timeindex、.snapshot以及.log檔案，這些檔案儲存在對應的topic分割區目錄下。
為了實現高可用目的，我們線上的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 四部分組成：

（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可能需要改進的地方：

由於Monitor生成的是整個叢集的負載模型，我們的Kafka平臺將Kafka叢集劃分為多個資源組，不同資源組的資源利用率存在很大差別，所以原生的叢集負載模型不再適用於我們的應用場景。
大多數業務沒有指定key進行生產，所以各分割區的負載偏差不大。如果topic分割區副本均勻分佈在資源組內，則資源組也隨之變得均衡。
原生的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之間的負載差距，並根據我們關注的資源去做分析，從而得出最終的遷移計劃。

以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可以做到如下幾點：

通過均衡引數，我們可以只均衡某個或多個資源組的broker。
更改topic設定，比如增加topic副本時，新擴的副本需要和topic原先的副本在同一個資源組內。
在資源組內分析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上遷移

考慮這些場景：

一個專案下會有幾個資源組，由於業務變更，業務想要把A資源組下的topic遷移到B資源組。
業務想要把公共資源組的topic遷移到C資源組。
均衡完成之後，發現總有幾個topic/分割區分佈不是很均勻。

面對這些場景，我們上面指定資源組進行均衡的功能就滿足不了我們的需求了。所以，我們針對上述場景改造後的cruise control可以做到如下幾點：

只均衡指定的topic或topic分割區；
均衡的topic或topic分割區只往指定的broker上遷移。

3.3.3 新增目標分析——topic分割區leader副本分散性

業務方大多都是沒有指定key進行傳送資料的，所以同一topic每個分割區的流量、儲存都是接近的，即每一個topic的各個分割區的leader副本儘可能均勻的分佈在叢集的broker上時，那叢集的負載就會很均勻。

有同學會問了，topic分割區數並不總是能夠整除broker數量，那最後各broker的負載不還是不一致嘛？

答案是肯定的，只通過分割區的leader副本還不能做到最終的均衡。

針對上述場景改造後的cruise control可以做到如下幾點：

新增一類資源分析：topic分割區leader副本分散性。
首先保證每個topic的leader副本和follower副本儘可能的均勻分佈在資源組的broker上。
在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存在比較明顯的兩個缺陷：

Kafka每個partition replica與機器的磁碟繫結，partition replica由一系列的Segment組成，所以往往單分割區儲存會佔用比較大的磁碟空間，對於磁碟會有很大壓力。
在叢集擴容broker時必須做Rebalance，需要broker有良好的執行流程，保證沒有任何故障的情況下使得各broker負載均衡。

cruise control就是針對Kafka叢集運維困難問題而誕生的，它能夠很好的解決kafka運維困難的問題。

參考文章：

Kafka 負載均衡在 vivo 的落地實踐