本文主要介紹云音樂對kafka的優(yōu)化，給生產(chǎn)集群帶來了顯著的收益，初步統(tǒng)計每年為云音樂節(jié)省幾百萬的成本。本文主要分為兩部分，第一部分詳細介紹優(yōu)化kafka性能時遇到的問題及解決過程，第二部分介紹kafka producer的寫入原理及batch優(yōu)化。如果對我們定位問題的過程不太感興趣的同學(xué)，可以直接閱讀第二部分。

一、背景

隨著云音樂主站流量的增長以及曙光埋點的放量，平臺的流量逐漸增長，kafka平臺壓力越來越大。當(dāng)前云音樂平臺分3個kafka集群：老集群、mirror集群、新集群。老集群和mirror集群歷史比較久遠，還在使用比較古老的kafka版本。新集群是最近搭建的，使用最新kafka版本，拆分為兩個region：default和dawn，其中，default集群用于新流量接入，dawn集群專屬用于曙光埋點流量使用。

kafka集群穩(wěn)定性的兩個重要指標(biāo)為：NetworkProcessorIdle和RequestHandlerIdle，PE同學(xué)反饋新kafka集群的這兩個指標(biāo)很低，高峰期在10%以下，集群流量卻與老集群相差較大，新kafka集群性能存在嚴(yán)重問題，同時，老集群也不容樂觀，idle高峰期15%左右，kafka集群的優(yōu)化迫在眉睫。

二、問題定位

構(gòu)建完善監(jiān)控體系

兵馬未動糧草先行，優(yōu)化要做監(jiān)控先行，監(jiān)控指標(biāo)一來可以指明方向，二來可以驗證優(yōu)化效果，所以，必須要有比較完善的監(jiān)控才能去做相關(guān)優(yōu)化。由于歷史原因，云音樂的kafka集群缺少必要的體系化的監(jiān)控告警，只有一些比較宏觀的監(jiān)控指標(biāo)，無法對問題進行多維度深入分析。首先我們需要做的就是構(gòu)建完善的監(jiān)控體系。

基于Prometheus構(gòu)建的監(jiān)控平臺已經(jīng)是當(dāng)前業(yè)界的標(biāo)配，使用prometheus有2個最大優(yōu)點：

1. kafka社區(qū)提供一套完善的基于prometheus+grafana的監(jiān)控系統(tǒng)構(gòu)建方法，包括指標(biāo)采集上報和grafana模板，可以很快的構(gòu)建出kafka的監(jiān)控平臺，詳細可參考此文章

2. 利用prometheus提供的PQL語言，可以很靈活的做到多監(jiān)控維度的下鉆和上卷分析，比較容易發(fā)現(xiàn)問題

另外，當(dāng)前網(wǎng)易內(nèi)部已經(jīng)有現(xiàn)成的prometheus平臺可以直接使用，因此，此方案可謂是不二之選。

基于此，PE同學(xué)僅花了一天時間就構(gòu)建出kafka集群完善的監(jiān)控系統(tǒng)，在grafana上可以看到集群的各種維度詳細指標(biāo)信息，包括：大盤流量、topic流量、broker流量、副本復(fù)制流量、線程idle情況、ISR信息、Consumer指標(biāo)、Producer指標(biāo)、Replica指標(biāo)、Request指標(biāo)等，這些詳細的指標(biāo)為我們分析問題帶來很大的便利。有了監(jiān)控后，我們開始著手集群的分析和優(yōu)化工作。

監(jiān)控指標(biāo)分析及優(yōu)化過程

1. 第一次分析及優(yōu)化：流量均衡

分析

我們對新集群的核心指標(biāo)進行了分析，并與老集群的指標(biāo)進行對比，發(fā)現(xiàn)新集群存在幾個現(xiàn)象：

1. bytesout分布不均，90MB/s ~ 180MB/s之間

2. request handler idle分布不均，20% ~ 80%之間

3. network processor idle都比較高，80%以上

4. msgin分布不均，18K op/s ~

5. Request Queue Size較高，有時會打滿到500，Response Queue Size較低

6. 磁盤IOUtil較低

7. 集群單臺機器出流量最高在1Gbps左右

基于上述現(xiàn)象我們基本可以得出以下幾個結(jié)論：

1. 請求處理線程不足

2. 流量分布不均是當(dāng)前主要矛盾

3. 當(dāng)前流量并未達到機器硬件的上限

木桶效應(yīng)很明顯，當(dāng)前集群吞吐量受限于壓力比較大的broker，因此，首先我們需要對集群進行流量均衡。

具體指標(biāo)如下：

優(yōu)化

PE同學(xué)均衡了kafka的bytesOut流量，idle稍微提升，但是效果不太明顯。

隨后對msgIn流量進行均衡，操作后，idle有了較明顯的變化，最低的idle從10%左右漲到了20%左右。

雖然idle上升了不少，但是從絕對值來看，kafka集群還是比較危險，一般idle在30%以下就說明集群已經(jīng)處于比較高負荷狀態(tài)，很難應(yīng)對流量突發(fā)情況。所以，優(yōu)化工作還需繼續(xù)。

2. 第二次分析及優(yōu)化：減少請求數(shù)

Network Processor Idle較高，Request Handler Idle較低，根據(jù)以前經(jīng)驗，當(dāng)大量meta請求時，會出現(xiàn)此類現(xiàn)象，查看metadata request指標(biāo)，每秒140左右，量級不大，應(yīng)該不是這個問題導(dǎo)致。關(guān)于這兩個idle的相關(guān)原理，可以參考kafka網(wǎng)絡(luò)模型總結(jié)

根據(jù)第一次優(yōu)化經(jīng)驗，我們有了一些優(yōu)化方向，bytesOut影響不大，msgIn影響較大，Request Handler是用來處理所有kafka的請求的（并不只是處理磁盤相關(guān)請求），也就是如果能降低request的請求次數(shù)，應(yīng)該就能提升idle值。分析request請求分布情況，發(fā)現(xiàn)絕大部分是在produce和fetch請求，因此，接下來的優(yōu)化方向是如何減輕讀寫的請求量。

kafka服務(wù)端使用zero copy的優(yōu)化技術(shù)，減輕服務(wù)端開銷，客戶端發(fā)送的數(shù)據(jù)塊，kafka服務(wù)端默認(rèn)不會做解包（compression.type=producer），也就是通過提升producer端的batch，是可以減輕request請求數(shù)的。

分析MsgIn較大的幾個topic發(fā)現(xiàn)，這些topic的batch size都很低，分布在1-2之間（對應(yīng)指標(biāo)messagesinpersec/totalproducerequestspersec），單條記錄大小不到1KB，對應(yīng)寫入任務(wù)的batch.size使用了默認(rèn)的16KB，按理來說，batch應(yīng)該是可以生效的。通過查看相關(guān)文章，發(fā)現(xiàn)影響producer batch的還有一個參數(shù)linger.ms（默認(rèn)為0），此參數(shù)原理可參考文章kafka producer linger.ms和batch.size參數(shù)說明。

我們將topic對應(yīng)的寫入任務(wù)，配置linger.ms=50，上線后產(chǎn)生了一定的效果，idle從12%左右提升到19%，如下：

查看對應(yīng)topic的batch size從1提升到4，bytesIn從14MB/s降低到4MB/s，如下：

從上述優(yōu)化效果來看，基本可以明確優(yōu)化Producer的batchSize，可以給集群帶來很好的效果。隨后，我們優(yōu)化了流量最大的UA日志（占據(jù)集群大部分流量）的寫入任務(wù)配置，集群的整體指標(biāo)得到了極大的改善，kafka集群整體壓力降低60%+，具體指標(biāo)如下：

總體topic的batchSize從1.67提升到9.71

bytesOut從3.5GB/s降低到1.7GB/s，bytesIn從392MB/s降低到218MB/s

idle從20%增長到90%

3. 優(yōu)化經(jīng)驗遷移到老集群經(jīng)歷

優(yōu)化不生效

有了上述優(yōu)化經(jīng)驗，我們繼續(xù)將此優(yōu)化配置應(yīng)用到老集群。但是，在變更老集群的部分寫入任務(wù)后，發(fā)現(xiàn)調(diào)整linger.ms參數(shù)后，雖然參數(shù)已生效，但是集群指標(biāo)沒有明顯變化，對應(yīng)topic的batchSize指標(biāo)也沒有明顯變化。

分析原因并調(diào)整生效

搜索了相關(guān)資料后，未發(fā)現(xiàn)特別有用的相關(guān)說明。只能去看kafka producer的源碼了，通過分析kafka的源碼，發(fā)現(xiàn)kafka producer寫入record時，如果record中沒有指定partitionID，則會隨機分配partitionID，放入對應(yīng)的topic-partition級別的本地緩存batch。同時。另外一個異步Sender線程從緩存batch中拿數(shù)據(jù)發(fā)送到對應(yīng)broker，此過程會在下面原理章節(jié)詳細介紹。

弄清楚相關(guān)原理后，經(jīng)過分析發(fā)現(xiàn)，我們優(yōu)化的寫入任務(wù)并行度120，對應(yīng)topic有150個分區(qū)，寫入流量大概40K/s，這樣均攤到每個topic-partition的流量大概是：2.2條/s，而我們設(shè)置的linger.ms為50，也就是Sender線程在從topic-partition的batch拿數(shù)據(jù)的時候，只能拿到1條數(shù)據(jù)，與監(jiān)控數(shù)據(jù)也比較吻合。

為了驗證分析，我們將任務(wù)的并行度從120調(diào)整到1，這樣每個topic-partition的流量大概是：266條/s。50ms一個batch，大概有13條左右，調(diào)整后觀察對應(yīng)topic的batchSize指標(biāo)，也與我們分析的相符合。

問題原因找到了，接下來需要考慮如何進行優(yōu)化。只需要將每條記錄對應(yīng)的partition的策略修改即可，起初我們想通過flink的接口FlinkKafkaPartitioner或者kafka自帶的接口Partitioner來自定義record對應(yīng)partition的計算策略（增加計數(shù)器，每N條記錄變更一次對應(yīng)的partition或者每N ms時間間隔變更一次）。但是更深入的看了kafka producer源碼后，發(fā)現(xiàn)2.4版本中，kafka新引入了一種Sticky Partitioner的策略，詳情可參考KIP-480，我們將寫入任務(wù)的kafka client版本升級到2.4使用sticky的partition策略進行測試后發(fā)現(xiàn)，batchSize有很大的提升。

將相關(guān)原始流量的分發(fā)寫入任務(wù)調(diào)優(yōu)寫入batch后，老kafka集群的壓力降低了40%+，具體指標(biāo)如下：

bytesIn從1.04GB/s降低到705MB/s，bytesOut從15.4GB/s降低到9.56GB/s

idle從8%提升到50%+

batchSize從1提升到16

三、原理

通過上述優(yōu)化過程，我們梳理清楚kafka producer的寫入原理，batch的寫入不止與linger.ms、batch.size的配置有關(guān)，還和partitioner計算record的分區(qū)規(guī)則有關(guān)。

首先，簡單介紹下，producer的寫入過程，如下圖：

kafka producer寫入主要分兩部分：寫入線程和異步sender發(fā)送線程。

1. 寫入線程

當(dāng)客戶端調(diào)用KafkaProducer.send(ProducerRecord)方法，主要有以下四個步驟：

1. 等待topicPartition相關(guān)元信息獲?。ㄓ芯彺妫?/p>

2. 序列化ProducerRecord中key和value（如果有）

3. 如果當(dāng)前reocord中沒有指定partition信息，則調(diào)用Partitioner.partition()獲取對應(yīng)的partition

4. 預(yù)估當(dāng)前記錄大小并將其追加到相應(yīng)的topic-partition緩存隊列的ProducerBatch中

2. Sender線程

發(fā)送線程會一直循環(huán)遍歷緩存隊列，并對數(shù)據(jù)做相關(guān)處理，發(fā)送你ProduceRequest請求到對應(yīng)broker，主要分以下六個步驟：

1. 遍歷所有topic-partition隊列的隊首ProducerBatch

2. 拿到該ProducerBatch做如下判斷（圖中為了簡便并未全部寫明）

• 第一個元素的追加時間與當(dāng)前時間相比，是否超過linger.ms

• ProducerBatch是否滿

• 內(nèi)存Buffer是否使用完

• 是否close或flush中

3. 如果滿足上述的一個條件，則將當(dāng)前ProducerBatch對應(yīng)的TopicPartition的leader所在的brokerId放入ready節(jié)點隊列中

4. 得到可以發(fā)送請求的broker的ready節(jié)點后，遍歷每個brokerId并做如下操作：

• 盡可能多的拿到該brokerId所負責(zé)的topicPartition對應(yīng)的ProducerBatch（當(dāng)record大小超過max.request.size時break，為了防止餓死，每次從不同的topicPartition獲?。?/p>

• 將這些ProducerBatch包裝成一個ProducerRequest

5. 更新每個batch相關(guān)的metric，如topic.{name}.records-per-batch/bytes/compression-rate等

6. 遍歷發(fā)送每個ProducerRequest到對應(yīng)的brokerId

3. 服務(wù)端

關(guān)于服務(wù)端的邏輯，此處只做簡單介紹，不做深入分析。

服務(wù)端收到ProducerRequest后，會將其中的每個ProducerBatch拆出來，使用zero-copy的方式將各topic-partition數(shù)據(jù)直接append到相應(yīng)的log-segment文件中，可極大提升服務(wù)端的性能。

4. 場景分析

現(xiàn)在我們來詳細看下為什么我們將老集群的寫入任務(wù)linger.ms設(shè)置為50時沒有效果。

我們寫入任務(wù)的流量大概是40K/s，寫入的topic分區(qū)數(shù)為150，任務(wù)并發(fā)為120，如下圖所示：

由于老版本（2.4之前）kafkaClient的每個record對應(yīng)的partition的分配邏輯是隨機的，從圖中可以很明顯看出，落到每個topicParitionQueue寫入緩存隊列的流量才不過40000/120/150=2.2，而我們設(shè)置的linger.ms為40，也就是在40ms內(nèi)，每個topicPartitionQueue中的數(shù)據(jù)最多也就一條，發(fā)送給每個的batch也就1左右，經(jīng)過測試也驗證了這個結(jié)論。

起初我們計劃通過flink或者kafka暴露出來的Partitioner接口自定義record對應(yīng)的partition分配邏輯。但是，還有一個問題比較奇怪，為什么在新集群上卻有效果呢？我們深入比較新老集群對應(yīng)的寫入任務(wù)并看了kafka相關(guān)源碼發(fā)現(xiàn)，kafka 2.4版本中對record的partition分配策略做了優(yōu)化。簡單來說sticky的分區(qū)策略是保證一個topic在一個producerBatch內(nèi)的所有rocords對應(yīng)的分區(qū)ID保證一樣，這樣就可以保證一個batch內(nèi)的數(shù)據(jù)都放到一個分區(qū)里，詳情可查看KIP-480。我們將老機器的寫入任務(wù)使用的kafka client版本也升級到2.4后，發(fā)現(xiàn)寫入的batch大小從1提升到12左右，也與我們預(yù)期相符。

5. 重要監(jiān)控指標(biāo)匯總

在此，總結(jié)下kafka broker端比較重要的一些監(jiān)控指標(biāo)以供后續(xù)完善相關(guān)告警做參考，具體信息可參考kafka監(jiān)控指標(biāo)官方文檔。

服務(wù)穩(wěn)定性相關(guān)

• active controller個數(shù)：kafka_controller_kafkacontroller_activecontrollercount，kafka中控節(jié)點，正常為1

• UR分區(qū)數(shù)：kafka_server_replicamanager_underreplicatedpartitions，追不上的分區(qū)數(shù)，正常為0

• 離線分區(qū)數(shù)：kafka_controller_kafkacontroller_offlinepartitionscount，未追的分區(qū)數(shù)（有broker宕機），正常為0

• 網(wǎng)絡(luò)處理線程idle：kafka_network_socketserver_networkprocessoravgidlepercent，處理socket請求的線程池空閑值，需保持在至少30%之上

• 請求處理線程idle：kafka_server_kafkarequesthandlerpool_requesthandleravgidlepercent_total，真正處理request請求的線程池空閑值，需保持在至少30%以上

• 請求的隊列大?。?code>kafka_network_requestchannel_requestqueuesize，默認(rèn)500，如果經(jīng)常滿說明當(dāng)前broker的處理慢或壓力大

• ISR shrink頻率：kafka_server_replicamanager_isrshrinkspersec，如果經(jīng)常出現(xiàn)，則說明經(jīng)常有副本同步不上leader被移除isr，預(yù)示集群存在穩(wěn)定性問題

• ISR expand頻率：kafka_server_replicamanager_isrexpandspersec，和shrink相反的指標(biāo)，表名副本追上leader放進isr的頻率

流量相關(guān)

• topic消息client寫入條數(shù)：kafka_server_brokertopicmetrics_messagesinpersec

• topic消息client寫入大小：kafka_server_brokertopicmetrics_bytesinpersec

• topic消息client讀大?。ú话琭ollower復(fù)制）：kafka_server_brokertopicmetrics_bytesoutpersec

• follower的replication副本復(fù)制bytesIn和bytesOut：ReplicationBytesInPerSec/ReplicationBytesOutPerSec

上述流量指標(biāo)均可細化到每個topic每個broker級別（無法到partition級別），通過PQL計算幾種組合信息，可查看broker/topic/總體級別流量信息。

請求相關(guān)

• producer/client/副本復(fù)制請求數(shù)：kafka_network_requestmetrics_requestspersec

• 每個topic的producer請求數(shù)：kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec

• 每個topic的拉取請求數(shù)：kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec

上述指標(biāo)，可細化到請求類型（FetchConsumer/FetchFollower/Produce/ListOffset等）、broker和topic級別，也可通過PQL進行各維度組合查看相應(yīng)指標(biāo)。

四、結(jié)尾

至此，本文完成對kafka的整個優(yōu)化過程和相關(guān)原理描述。后續(xù)也可將此優(yōu)化推廣到其他此類場景。優(yōu)化前需要注意以下四項：

1. 通過監(jiān)控指標(biāo)確認(rèn)寫入的topic的batch大小是否為1左右，在kafka-overview中可查看相應(yīng)指標(biāo)

2. 確保程序中，寫入kafka的records中并未指定key（如果指定key，則默認(rèn)不會使用sticky分配策略，以確保相同key在一個partition，除非強行指定sticky partitioner）

3. 當(dāng)linger.ms設(shè)置為0時，batch也可能生效，只要數(shù)據(jù)產(chǎn)生速度大于發(fā)送速度即可

4. 當(dāng)設(shè)置linger.ms時，允許數(shù)據(jù)存在指定的延遲時間

另外，kafka服務(wù)還存在一定的優(yōu)化空間，包括磁盤、網(wǎng)絡(luò)、內(nèi)存等場景，這里就做不詳細介紹了。

在此，也感謝PE同學(xué)：馬宏展、童罕，感謝他們提供的幫助，尤其是kafka的監(jiān)控系統(tǒng)和promethus平臺，在優(yōu)化過程中發(fā)揮了重要的作用。

五、參考文章

1. 使用promethus+grafana監(jiān)控kafka

2. kafka網(wǎng)絡(luò)模型總結(jié)

3. kafka producer linger.ms和batch.size參數(shù)說明

4. KIP-480

5. kafka監(jiān)控指標(biāo)官方文檔