【編者的話】在大規(guī)模的使用Kafka過程中，我們通常會(huì)遇到各種各樣的問題，比如說，通常會(huì)有一些大數(shù)據(jù)集群中的Job發(fā)現(xiàn)總有幾個(gè)Task會(huì)比較慢，導(dǎo)致整體的任務(wù)遲遲不能完成運(yùn)行，這種情況通常問題會(huì)比較復(fù)雜，想要知道具體延遲在哪里，我們需要知道在Kafka集群中哪些點(diǎn)可能會(huì)增加端到端的延遲。

接下來的內(nèi)容翻譯自Confluent官網(wǎng)博客中的一篇文章，希望能夠幫助大家理解Kafka使用過程中端到端的延遲。

欺詐檢測、支付系統(tǒng)和股票交易平臺只是許多Apache Kafka用例中的一小部分，這些用例需要快速且可預(yù)測的數(shù)據(jù)交付。例如，在線銀行交易的欺詐檢測必須實(shí)時(shí)發(fā)生，以交付業(yè)務(wù)價(jià)值，而不需要為每個(gè)交易增加超過50-100毫秒的開銷，以保持良好的客戶體驗(yàn)。

在Kafka術(shù)語中，數(shù)據(jù)交付時(shí)間(data delivery time)是由端到端延遲(end-to-end latency)定義的，即消費(fèi)者獲取一條向Kafka生成的記錄所需的時(shí)間。延遲目標(biāo)表示為目標(biāo)延遲和滿足此目標(biāo)的重要性。例如，您的延遲目標(biāo)可以表示為:我希望99%的情況下從Kafka獲得端到端延遲為50 ms。

這將增加可用性、持久性和吞吐量目標(biāo)。實(shí)現(xiàn)高持久性和高吞吐量兩個(gè)目標(biāo)，我們需要進(jìn)行一定的權(quán)衡，挑戰(zhàn)在于在保持延遲界限的同時(shí)擴(kuò)展應(yīng)用程序的吞吐量，并調(diào)整Kafka集群的大小以使用可接受的Broker延遲來處理客戶端和復(fù)制的請求。延遲也取決于您對硬件或云提供商的選擇，所以您需要能夠監(jiān)視和調(diào)優(yōu)您的客戶端，以在您獨(dú)特的環(huán)境中實(shí)現(xiàn)您的特定延遲目標(biāo)。

注意：通常情況下，Broker所在的網(wǎng)絡(luò)區(qū)域其實(shí)也會(huì)對延遲造成很大影響，當(dāng)然這仍然取決于您對可用性和延遲的權(quán)衡。

之前，我們有寫過白皮書《Optimizing Your Apache Kafka Deployment》，其中列出了配置Kafka部署以優(yōu)化各種目標(biāo)的指導(dǎo)原則。

這篇文章將幫助你進(jìn)一步獲得更好的直覺和對端到端延遲的理解，并配置和擴(kuò)展您的應(yīng)用程序的吞吐量，同時(shí)保持延遲的界限。

理解端到端的延遲(end-to-end latency)

端到端延時(shí)是指應(yīng)用邏輯調(diào)用KafkaProducer.send()生產(chǎn)消息到該消息被應(yīng)用邏輯通過KafkaConsumer.poll()消費(fèi)到之間的時(shí)間。

下圖顯示了一條記錄在系統(tǒng)中的路徑，從Kafka生產(chǎn)者到Kafka的Broker節(jié)點(diǎn)，副本的復(fù)制，以及消費(fèi)者最終在其主體分區(qū)日志中獲取到具體的消息。

因此，端到端的延遲主要會(huì)由以下幾個(gè)部分組成：

• Produce time：內(nèi)部Kafka Producer處理消息并將消息打包的時(shí)間
• Publish time：Producer發(fā)送到Broker并寫入到Leader副本log的時(shí)間
• Commit time：Follower副本備份消息的時(shí)間
• Catch-up time：消費(fèi)者追趕消費(fèi)進(jìn)度，消費(fèi)到該消息位移值前所花費(fèi)的時(shí)間
• Fetch time：從Broker讀取該消息的時(shí)間

在接下來的內(nèi)容中，我們將分別解釋這五個(gè)延遲階段的具體含義，特定的客戶端配置或應(yīng)用邏輯設(shè)計(jì)通常會(huì)極大地影響端到端延時(shí)，因此我們有必要精準(zhǔn)定位哪個(gè)因素對延時(shí)的影響最大。

• Produce time

Produce time指的是從應(yīng)用程序通過KafkaProducer.send()生產(chǎn)一條記錄到包含該消息的生產(chǎn)者請求被發(fā)往leader副本所在的broker之間的時(shí)間。(因此，生產(chǎn)者所處的網(wǎng)絡(luò)環(huán)境以及對應(yīng)topic分區(qū)leader副本所在的broker的網(wǎng)絡(luò)可能會(huì)影響到produce time的延遲)

Kafka producer會(huì)將相同topic分區(qū)下的一組消息打包在一起形成一個(gè)批次(batch)以提升網(wǎng)絡(luò)I/O性能。(在必要情況下，我們可以對生產(chǎn)者的batch size進(jìn)行一定的調(diào)整)

默認(rèn)情況下，producer會(huì)立即發(fā)送batch，這樣一個(gè)batch中通常不會(huì)包含太多的消息。為了提高batch的效率，生產(chǎn)者通常會(huì)對linger.ms來人為設(shè)置一個(gè)較小的延遲來保證有足夠多的消息記錄能封裝在一個(gè)batch中。一旦過了linger.ms設(shè)置的事件，或者batch size已經(jīng)達(dá)到最大值(batch.size的參數(shù)值)，這個(gè)batch將被認(rèn)為已經(jīng)完成。

如果生產(chǎn)者也開啟了壓縮(compression.type)，Kafka的生產(chǎn)者會(huì)將已完成的batch進(jìn)行壓縮。在batch完成之前，它的大小時(shí)根據(jù)生產(chǎn)者指定的壓縮類型和之前觀測到的壓縮比率估算出來的。

如果發(fā)送給leader副本的未確認(rèn)的生產(chǎn)者請求數(shù)量已經(jīng)達(dá)到最大(max.inflight.requests.per.connection=5)，則在生產(chǎn)者的批處理可能需要等待更長的時(shí)間。因此，broker響應(yīng)生產(chǎn)者請求越快，生產(chǎn)者的等待時(shí)間也將會(huì)變得更小。

• Publish time

Publish time是指內(nèi)部Kafka生產(chǎn)者發(fā)送生產(chǎn)者請求到一個(gè)broker節(jié)點(diǎn)，并且對應(yīng)的消息到達(dá)leader副本日志之間的時(shí)間。當(dāng)請求到達(dá)Broker節(jié)點(diǎn)時(shí)，負(fù)責(zé)連接的網(wǎng)絡(luò)線程將獲取該請求并將其放入請求隊(duì)列中。其中一個(gè)請求處理程序線程從隊(duì)列中獲取請求并處理它們。(對應(yīng)broker節(jié)點(diǎn)的num.thread 和num.io.thread兩個(gè)相關(guān)參數(shù))

因此，Publish time包含生產(chǎn)者請求的網(wǎng)絡(luò)時(shí)間，broker上的排隊(duì)時(shí)間，以及將消息追加到日志所消耗的時(shí)間(通常也是page cache 的訪問時(shí)間)。當(dāng)Broker端負(fù)載比較低，網(wǎng)絡(luò)和日志的追加寫入時(shí)間會(huì)影響publish time，隨著broker負(fù)載變高，隊(duì)列延遲的增加 將會(huì)更多的影響publish time。

• Commit time

Commit time是指從leader副本中復(fù)制消息到全部的同步副本(all in-sync replicas)中所消耗的時(shí)間。Kafka只會(huì)將已提交(committed)的消息暴露給consumer，也就是該消息必須在全部的ISR中包含。follower副本中的消息會(huì)從leader副本中并行的拉取，在一個(gè)正常的集群中，我們通常不希望副本處于不同步狀態(tài)(當(dāng)然有的業(yè)務(wù)場景可能會(huì)導(dǎo)致短暫的不同步現(xiàn)象)。這意味著消息被提交的時(shí)間等于ISR中最慢的follower副本所在broker去從ledaer broker節(jié)點(diǎn)獲取記錄并寫入到follower副本日志的時(shí)間。

為了復(fù)制數(shù)據(jù)，follower所在的broker會(huì)想leader節(jié)點(diǎn)發(fā)送fetch請求，準(zhǔn)確的來講消費(fèi)者也是使用fetch請求來獲取消息。但是，官方在副本復(fù)制的fetch請求中，broker端優(yōu)化了默認(rèn)配置: leader副本會(huì)盡早的發(fā)送請求，只要有一個(gè)字節(jié)可用，就會(huì)發(fā)起fetch請求(由replica.fetch.min.bytes參數(shù)控制)或者當(dāng)replica.fetch.wait.max.ms滿足條件。Commit time主要受副本因此配置參數(shù)的影響以及集群的當(dāng)前負(fù)載情況。

• Catch-up time

Kafka中消息是按照其生產(chǎn)的順序被消費(fèi)的，除非顯示的聲明了一個(gè)新的offset或者有一個(gè)新的消費(fèi)者從最新的offset進(jìn)行消費(fèi)。同一個(gè)分區(qū)下，consumer必須要消費(fèi)完之前發(fā)布的消息后才能讀取后面的消息。假設(shè)在提交消息時(shí)，消費(fèi)者的偏移量是提交消息后面的N條消息，那么，Catch-up time就是消費(fèi)者消費(fèi)者N條消息的總時(shí)間。

當(dāng)我們在構(gòu)建實(shí)時(shí)處理應(yīng)用的時(shí)候，最好讓catch-up時(shí)間為0，即一旦消息被提交，消費(fèi)者可以立馬讀取到消息。如果消費(fèi)者總是落后，端到端延遲可能會(huì)變得無限大。因此，catch-up 時(shí)間通常依賴于消費(fèi)者的能力是否能夠追趕上生產(chǎn)者的吞吐量。

• Fetch time

訂閱主題分區(qū)的消費(fèi)者會(huì)不斷輪詢?nèi)膌eader副本中獲取更多的數(shù)據(jù)，F(xiàn)etch time是從leader副本所在broker節(jié)點(diǎn)獲取消息記錄的s時(shí)間，可能需要等待足夠的數(shù)據(jù)來形成對fetch請求的響應(yīng)，并從KafkaConsumer.poll()的響應(yīng)中返回記錄。在默認(rèn)的配置下，已經(jīng)對于消費(fèi)者的fetch延遲做了優(yōu)化(fetch.min.bytes=1)，即及時(shí)只有一個(gè)字節(jié)可用的時(shí)候，fetch請求也會(huì)響應(yīng)數(shù)據(jù)，或者在一個(gè)短暫超時(shí)之后fetch.max.wait.ms。

• End-to-end latency VS producer and consumer latencies

下圖顯示了Kafka客戶端觀察到的延遲(通常稱為生產(chǎn)者延遲和消費(fèi)者延遲)與端到端延遲之間的關(guān)系。

生產(chǎn)者延遲是指KafkaProducer.send()發(fā)送和生產(chǎn)的消息被確認(rèn)間的事件。消息的確認(rèn)依賴于acks的配置，該參數(shù)可以控制消息的持久性(durability)：

• 當(dāng)acks=0，立即確認(rèn)，不等待broker的返回
• 當(dāng)acks=1，消息被追加到leader副本所在分區(qū)后再確認(rèn)
• 當(dāng)acks=all，在所有的ISR(同步副本)都接收到消息時(shí)才確認(rèn)

所以，生產(chǎn)者延遲包含produce time，publich time(如果acks >= 1)，commit time(如果acks=all)以及生產(chǎn)者響應(yīng)從broker返回到生產(chǎn)者的時(shí)間。

上圖清晰的向我們展示了為何改變acks參數(shù)能夠減少生產(chǎn)者延遲(其實(shí)是通過從生產(chǎn)者延遲中移除幾個(gè)延遲概念來減少的publish和commit)。不過，無論我們?nèi)绾闻渲蒙a(chǎn)者的acks參數(shù)，publish和commit時(shí)間總是端到端延遲的一部分。

消費(fèi)者延遲(Consumer latency)是指消費(fèi)者發(fā)起一個(gè)fetch請求到broker節(jié)點(diǎn)，以及broker節(jié)點(diǎn)向consumer返回響應(yīng)的時(shí)間。計(jì)算方法是KafkaConsumer.poll()返回的時(shí)間。Consumer的延遲主要包含了上圖中的fetch time。

控制end-to-end latency

如果我們思考一條消息的生命周期，控制端到端延時(shí)其實(shí)就是控制消息在系統(tǒng)中流轉(zhuǎn)的時(shí)間總和。很多Kafka clients端和broker端參數(shù)的默認(rèn)值已然對延時(shí)做了優(yōu)化：比如減少人為等待時(shí)間來提高批處理的能力(通過linger.ms，fetch.min.bytes，replica.fetch.min.bytes參數(shù)來適當(dāng)調(diào)優(yōu))。其他的延時(shí)可能來自于broker端上的隊(duì)列等候時(shí)間，控制這種延時(shí)就要涉及控制broker的負(fù)載(CPU或吞吐量)，通常情況下我們要時(shí)刻關(guān)注broker節(jié)點(diǎn)的各項(xiàng)基礎(chǔ)監(jiān)控指標(biāo)。

如果我們將系統(tǒng)視為一個(gè)整體，那么整個(gè)端到端的延遲還要求系統(tǒng)中的每一個(gè)部分(生產(chǎn)者，broker，消費(fèi)者)都能夠可靠的維持應(yīng)用程序邏輯所需的吞吐量。

例如，如果你的應(yīng)用程序邏輯以100 MB/s發(fā)送數(shù)據(jù)，但是由于某種原因，你的Kafka消費(fèi)者吞吐量在幾秒鐘內(nèi)下降到10 MB/s，那么在此之前產(chǎn)生的大多數(shù)消息都需要在系統(tǒng)中等待更長的時(shí)間，直到消費(fèi)者趕上了。此時(shí)，你需要一種高效的方式來擴(kuò)展你的Kafka clients程序以提升吞吐量——高效地利用broker端資源來減少隊(duì)列等候時(shí)間和偶發(fā)的網(wǎng)絡(luò)擁塞。

理想情況下，限制延遲意味著確保所有延遲都低于目標(biāo)。但實(shí)際生產(chǎn)環(huán)境中，由于意外故障和峰值負(fù)載，這種嚴(yán)格的保證是不可能的。不過，可以設(shè)計(jì)應(yīng)用程序并對系統(tǒng)進(jìn)行調(diào)優(yōu)，以實(shí)現(xiàn)95%的延遲目標(biāo)，控制所有的消息延遲在95~99%低于目標(biāo)延遲時(shí)間。高百分位延遲也稱為尾部延遲，因?yàn)樗鼈兪茄舆t頻譜的尾部。

目標(biāo)延時(shí)所用的百分位越大，你需要降低或容忍應(yīng)用最差表現(xiàn)所做的努力就越多。比如，偶爾的大請求可能會(huì)阻塞全部的請求，從而增加整體的延遲，這也就是所謂的head-of-line隊(duì)首阻塞。同時(shí)，大量低速率客戶端可能偶爾會(huì)同時(shí)向Kafka發(fā)送生產(chǎn)或消費(fèi)請求，或全部刷新集群元數(shù)據(jù)，也會(huì)導(dǎo)致請求隊(duì)列比平常更長，從而引發(fā)比平時(shí)更嚴(yán)重的尾延遲。這種行為就是所謂的micro-bursting(微型沖擊，可能就是水滴石穿的意思吧)。

不同客戶端配置的延遲測試

在這接下來的內(nèi)容中，我們使用實(shí)驗(yàn)結(jié)果來說明Kafka客戶端配置和吞吐量擴(kuò)展技術(shù)對性能的影響。我們使用Kafka內(nèi)置的Trogdor測試框架以及生產(chǎn)者和消費(fèi)者的基準(zhǔn)測試，ProduceBench和ConsumeBench來進(jìn)行我們的生產(chǎn)者和消費(fèi)者實(shí)驗(yàn)測試。

我們所有的測試都在一個(gè)包含9個(gè)代理的Kafka集群上運(yùn)行，該集群的復(fù)制因子為3，這保證了在出現(xiàn)最多兩個(gè)同時(shí)發(fā)生的節(jié)點(diǎn)故障時(shí)不會(huì)丟失消息。

Kafka集群運(yùn)行在AWS的r5.xlarge實(shí)例上，使用有2T的EBS(彈性塊存儲)。Kafka的broker節(jié)點(diǎn)分布在同一區(qū)域內(nèi)的三個(gè)可用性區(qū)域(AZ)，以獲得更強(qiáng)的容錯(cuò)性，其中每個(gè)主題分區(qū)副本被放置在一個(gè)不同的AZ上，并且Kafka客戶端配置使用SASL認(rèn)證和SSL加密，Broker之間使用PLAINTEXT進(jìn)行通信。

主題：需要注意的是，分布式集群中節(jié)點(diǎn)如果在不同可用區(qū)也可能導(dǎo)致延遲的增加，當(dāng)然這要在延遲和容錯(cuò)性角度進(jìn)行權(quán)衡，也需要考慮到云廠商的可用區(qū)之間本身的延遲。

我們的實(shí)驗(yàn)使用了以下非默認(rèn)客戶端配置和其他規(guī)范：

這個(gè)測試場景會(huì)產(chǎn)生額外的延遲：多可用區(qū)可能增加commit時(shí)間，由于是跨可用區(qū)的副本。無論是clients端還是broker端，SSL加密也是有開銷的。同時(shí)由于SSL無法利用Zero Copy特性進(jìn)行數(shù)據(jù)傳輸，因?yàn)閏onsumer獲取消息時(shí)也會(huì)增加額外的開銷。

雖然這些因素都會(huì)影響延遲，但是通常情況下企業(yè)內(nèi)部可能還是需要這種架構(gòu)上的考慮，因此采用該部署結(jié)構(gòu)進(jìn)行測試。

持久性設(shè)置對延遲的影響

當(dāng)將延遲目標(biāo)與其他需求疊加在一起時(shí)，首先考慮持久性需求是很有用的。由于數(shù)據(jù)的重要性，通常需要一定程度的持久性。

優(yōu)化持久性會(huì)增加端到端延遲，因?yàn)檫@會(huì)增加延遲的復(fù)制開銷(提交時(shí)間)，并向Broker添加復(fù)制負(fù)載，從而增加排隊(duì)延遲。

• Replication factor

Replication factor是Kafka持久化保證的核心，它定義了Kafka集群上保存的topic副本數(shù)。Replication factor = N表示我們最多能夠容忍N(yùn)-1臺broker宕機(jī)而不必?cái)?shù)據(jù)丟失。N=1能夠令端到端延時(shí)最小化，但卻是最低的持久化保證。

增加副本數(shù)會(huì)增加備份開銷并給broker額外增加負(fù)載。如果clients端帶寬在broker端均勻分布，那么每個(gè)broker都會(huì)使用N * w寫帶寬和r + (N - 1) * w讀帶寬，其中w是clients端在broker上的寫入帶寬占用，r是讀帶寬占用。

由此，降低N 對端到端延時(shí)影響的最佳方法就是確保每個(gè)broker上的負(fù)載是均勻的。這會(huì)降低commit time，因?yàn)閏ommit time是由最慢的那個(gè)follower副本決定的。

如果你的Kafka broker使用了過多的磁盤帶寬或CPU，follower就會(huì)開始出現(xiàn)追不上leader的情況從而推高了commit time。(其實(shí)還需要注意的是，當(dāng)最小的ISR默認(rèn)為副本的數(shù)量個(gè)數(shù)時(shí)，在出現(xiàn)follower和leader不同步時(shí)恰巧leader節(jié)點(diǎn)宕機(jī)，會(huì)導(dǎo)致topic本身不可用)

我們建議為副本同步消息流量設(shè)置成使用不同的listener來減少與正常clients流量的干擾。你也可以在follower broker上增加I/O并行度，并增加副本拉取線程數(shù)量number.replica.fetchers來改善備份性能。

• Acks

縱然我們配置了多個(gè)副本，producer還是必須通過Acks參數(shù)來配置可靠性水平。設(shè)置acks=all能夠提供最強(qiáng)的可靠性保證，但同時(shí)也會(huì)增加broker應(yīng)答PRODUCE請求的時(shí)間，就像我們之前討論的那樣。

Broker端應(yīng)答的速度變慢通常會(huì)降低單個(gè)producer的吞吐量，進(jìn)而增加producer的等待時(shí)間。這是因?yàn)閜roducer端會(huì)限制未應(yīng)答請求的數(shù)量(max.inflight.requests.per.connection)。

舉個(gè)例子，在我們的環(huán)境中acks=1，我們啟動(dòng)了9個(gè)producer(同時(shí)也跑了9個(gè)consumer)，吞吐量達(dá)到了195MB/秒。當(dāng)Acks切換成all時(shí)，吞吐量下降到161MB/秒。設(shè)置更高級別的acks通常要求我們擴(kuò)展producer程序才能維持之前的吞吐量水平以及最小化producer內(nèi)部的等待時(shí)間。

• Min.insync.replicas

min.insync.replicas是一個(gè)重要的持久化參數(shù)，因?yàn)樗x了broker端ISR副本中最少要有多少個(gè)副本寫入消息才算PRODUCE請求成功。這個(gè)參數(shù)會(huì)影響可用性，但是不會(huì)影響端到端的延時(shí)。因此，選擇一個(gè)小一點(diǎn)的值并不能減少commit time并減少延遲。

在滿足延遲目標(biāo)的前提下擴(kuò)展吞吐

• 延遲和吞吐的權(quán)衡

優(yōu)化Kafka clients端吞吐量意味著優(yōu)化batching的效果。Kafka producer內(nèi)部會(huì)執(zhí)行一類batching，即收集多條消息到一個(gè)batch中。

每個(gè)batch被統(tǒng)一壓縮然后作為一個(gè)整體被寫入日志或從日志中讀取。這說明消息備份也是以batch為單位進(jìn)行的。

Batching會(huì)減少每條消息的成本，因?yàn)樗鼘⑦@些成本攤還到clients端和broker端。通常來說，batch越大這種開銷降低的效果就越高，減少的網(wǎng)絡(luò)和磁盤I/O就越多。

另一類batching就是在單個(gè)網(wǎng)絡(luò)請求/響應(yīng)中收集多個(gè)batch以減少網(wǎng)絡(luò)數(shù)據(jù)傳輸量。這能降低clients端和broker端的請求處理開銷。這類batching能夠提升吞吐量和降低延時(shí)，因?yàn)閎atch越大，網(wǎng)絡(luò)傳輸I/O量越小，CPU和磁盤使用率越低，故最終能夠優(yōu)化吞吐量。另外batch越大還能減低端到端延時(shí)，因?yàn)槊織l消息的成本降低了，使得系統(tǒng)處理相同數(shù)量消息的總時(shí)間變少了。

這里的延時(shí)-吞吐量權(quán)衡是指通過人為增加等待時(shí)間來提升打包消息的能力。但過了某個(gè)程度，人為等待時(shí)間的增加可能會(huì)抵消或覆蓋你從打包機(jī)制獲得的延時(shí)收益。因此你的延時(shí)目標(biāo)有可能會(huì)限制你能實(shí)施打包化的水平，進(jìn)而減少所能達(dá)到的吞吐量并增加延時(shí)。如果拉低了本能達(dá)到的吞吐量或端到端延時(shí)水平，你可以通過擴(kuò)展集群來換取或“購買”更多的吞吐量或處理能力。

• 配置kafka的生產(chǎn)者和消費(fèi)者以實(shí)現(xiàn)batching

對于producer而言，batching由兩個(gè)參數(shù)進(jìn)行控制: batch.size(16KB)和linger.ms(0)，前者控制batch的大小，后者限制延遲量。如果使用場景中，應(yīng)用會(huì)頻繁的像kafka集群發(fā)送數(shù)據(jù)，及時(shí)設(shè)置了linger.ms=0，整個(gè)batch也會(huì)被盡快填滿。如果應(yīng)用生產(chǎn)數(shù)據(jù)的頻率較低，可以通過增加linger.ms來增加batch。

對于consumer而言，可以調(diào)整fetch.min.bytes(1)來限制每個(gè)消費(fèi)者在每個(gè)fetch響應(yīng)中接收的數(shù)據(jù)量，該參數(shù)指定了broker應(yīng)該在一個(gè)fetch響應(yīng)中返回的最小數(shù)據(jù)，以及fetch.max.wait.ms(500)來設(shè)置等待數(shù)據(jù)的超時(shí)時(shí)間。在fetch響應(yīng)中的數(shù)據(jù)越多，就會(huì)有更少的fetch請求。

在生產(chǎn)者端的batching也會(huì)間接影響produce和fetch的請求數(shù)量，因?yàn)閎atch定義了數(shù)據(jù)能夠被獲取的最小數(shù)據(jù)量。

值得注意的是，默認(rèn)情況下，Kafka producer和consumer設(shè)置的是無人為等待時(shí)間，這么做的目的是為了降低延時(shí)。但是，即使你的目標(biāo)就是了使延時(shí)最小化，我們依然推薦你設(shè)置一個(gè)不為0的linger.ms值，比如5~10ms。當(dāng)然，這么做是有前提的：

• 如果你擴(kuò)展了你的producer程序，平均下來使得每個(gè)producer實(shí)例的發(fā)送速率變得很低，那么你的batch只會(huì)包含很少的幾條消息。如果你整體的吞吐量已然很高了，那么你可能會(huì)直接把你的Kafka集群壓掛，導(dǎo)致超高的隊(duì)列等候時(shí)間從而推高延時(shí)。此時(shí)，設(shè)置一個(gè)較小的linger.ms值確實(shí)能夠改善延時(shí)。
• 如果你在意尾延時(shí)，那么增加linger.ms可能會(huì)降低請求速率以及同時(shí)到達(dá)broker端的瞬時(shí)沖擊流量。這種沖擊越大，請求在尾部的延時(shí)就越高。這些瞬時(shí)沖擊流量決定了你的尾延時(shí)水平。

下面這個(gè)實(shí)驗(yàn)說明了以上兩種場景。我們啟動(dòng)了90個(gè)producer，向一個(gè)有108個(gè)分區(qū)的topic發(fā)送消息。生產(chǎn)端整體的吞吐量峰值在90MB/秒。我們跑了3次測試，每一次對應(yīng)一種不同的producer配置。

因?yàn)樵诮o定的總吞吐下，我們有相對大量的生產(chǎn)者，因此linger.ms = 0導(dǎo)致在生產(chǎn)者端機(jī)會(huì)沒有batch操作。將linger.ms從0調(diào)整到5可以增加batching能力: 向kafka發(fā)起的生產(chǎn)者請求從2800降低到了1100。這減少了50%和99%的生產(chǎn)者延遲。

增加batch.size不會(huì)直接影響生產(chǎn)者的等待時(shí)間，因?yàn)樯a(chǎn)者在填滿batch的時(shí)間不會(huì)超過linger.ms的限制。在我們的實(shí)驗(yàn)中，增加batch.size從16KB到128KB沒有增加bacth的效果，因?yàn)槊總€(gè)生產(chǎn)者的吞吐量非常低。正如預(yù)期的那樣，生產(chǎn)者延遲在兩種配置之間沒有變化。

總之，如果您的目標(biāo)是最小化延遲，我們建議保留默認(rèn)的客戶端批處理配置，并盡可能增加linger.ms。如果你在意尾延時(shí)，最好調(diào)優(yōu)下打包水平來減少請求發(fā)送率以及大請求沖擊的概率。

不增加人為延遲以提高batching效率

batching效果不好的另一個(gè)原因是producer發(fā)送消息給大量分區(qū)。如果消息不是發(fā)往同一個(gè)分區(qū)的，它們就無法聚集在一個(gè)batch下。因此，通常最好設(shè)計(jì)成讓每個(gè)producer都只向有限的幾個(gè)分區(qū)發(fā)送消息。

另外，可以考慮升級到Kafka 2.4 producer。這個(gè)版本引入了一個(gè)全新的Sticky分區(qū)器。該分區(qū)器能夠改善non-keyed topic的打包效果，同時(shí)還無需引入人為等待。

• clients的數(shù)量對尾延遲(tail-latency)的影響

即使整體的生產(chǎn)和消費(fèi)的吞吐量保持不變，通常也是Clients數(shù)越多，broker上負(fù)載越大。這是因?yàn)閏lients數(shù)量多會(huì)導(dǎo)致更多的METADATA請求發(fā)到Kafka，繼而要維護(hù)更多的連接，故給broker帶來更大的開銷。

相對于50%或平均延時(shí)的影響，Clients數(shù)量增加對尾延時(shí)的影響更大。

每個(gè)producer最多發(fā)送max.inflight.requests.per.connection個(gè)PRODUCE請求給單個(gè)broker，而每個(gè)consumer一次最多只會(huì)給一個(gè)broker發(fā)送FETCH請求。Clients越多，同一時(shí)刻發(fā)送到broker的PRODUCE和FETCH請求也就越多，這就增加了形成請求瞬時(shí)沖擊的概率，進(jìn)而推高了尾延時(shí)。

Consumer數(shù)量通常由topic分區(qū)數(shù)量以及期望consumer沒有較大lag的目標(biāo)共同決定。但是，我們卻很容易為了擴(kuò)展吞吐量而引入大量的producer。

基于吞吐量的考量增加producer實(shí)例數(shù)可能有相反的效果，因?yàn)閜roducer會(huì)導(dǎo)致更少的消息被打包，畢竟每個(gè)producer處理了更少的消息，因而發(fā)送速率會(huì)變慢。同時(shí)producer還必須等待更長的時(shí)間來積累相同數(shù)量的消息進(jìn)到batch里面。

在我們的實(shí)驗(yàn)中，我們將producer的數(shù)量從90增加到900，發(fā)現(xiàn)吞吐量沒有他打變化：90MB/秒。

我們使用batch.size=16KB，linger.ms=5，acks=all的生產(chǎn)者配置，實(shí)驗(yàn)結(jié)果如下:

結(jié)果顯示增加producer數(shù)量(90->900)增加了60%的中位數(shù)延時(shí)值，而99%延時(shí)值幾乎增加了3倍。

延時(shí)的增加是因?yàn)閜roducer端打包效果變差導(dǎo)致的。

尾延時(shí)的增加是因?yàn)楦蟮恼埱笏矔r(shí)沖擊，這會(huì)拉升broker端延時(shí)，同時(shí)producer端會(huì)等待更長的時(shí)間來接收應(yīng)答。

在900個(gè)producer的測試中，broker完全被PRODUCE請求壓垮了。用于處理請求的時(shí)間幾乎占到了broker端CPU使用率的100%。另外由于我們使用了SSL，它也會(huì)進(jìn)一步引入請求級的開銷。

如果你通過添加producer來提升吞吐量，那么可以考慮增加單個(gè)proudcer的吞吐量，即改善batching的效果。不管怎樣，你最終可能會(huì)有很多producer實(shí)例。比如，大公司收集設(shè)備上的統(tǒng)計(jì)指標(biāo)，而設(shè)備數(shù)可能有成千上萬。此時(shí)，你可以考慮使用一個(gè)Broker收集來自多個(gè)clieints的請求，然后把它們轉(zhuǎn)換成更高效的PRODUCE請求再發(fā)給Kafka。你也可以增加broker數(shù)來降低單個(gè)broker上的請求負(fù)載。

關(guān)于增加消費(fèi)者數(shù)量的說明

當(dāng)擴(kuò)展消費(fèi)者時(shí)需要注意，在同一個(gè)消費(fèi)者組的消費(fèi)者會(huì)提交offset信息和心跳到broker節(jié)點(diǎn)上(controller節(jié)點(diǎn))。如果按時(shí)間間隔執(zhí)行偏移提交(auto.commit.interval.ms)，則消費(fèi)者組中的更多消費(fèi)者者會(huì)增加偏移提交率。偏移量提交本質(zhì)上是向內(nèi)部__consumer_offsets產(chǎn)生請求，因此增加consumer數(shù)量會(huì)導(dǎo)致broker上的請求負(fù)載增加，特別是auto.commit.interval.ms值很小的時(shí)候。

壓縮配置的影響

默認(rèn)情況下，Kafka producer不做壓縮。compression.type參數(shù)可以決定要不要做壓縮。

壓縮會(huì)在producer端引入額外的開銷來壓縮消息，在broker端做校驗(yàn)時(shí)解壓縮從而引入額外的開銷，另外在consumer端解壓縮也是開銷。

注意：通常情況下broker端的壓縮參數(shù)需要設(shè)置成producer，以避免壓縮方式?jīng)_突導(dǎo)致數(shù)據(jù)無法正常消費(fèi)，這樣broker只需要直接將壓縮后的日志寫入。

雖然壓縮會(huì)增加CPU開銷，但它還是可能減少端到端延時(shí)的，因?yàn)樗茱@著地降低處理數(shù)據(jù)所需的帶寬占用，進(jìn)而減少broker端的負(fù)載。壓縮是在batch級別上完成的，故打包效果越好，壓縮效果也就越好。

更多的分區(qū)可能增加延遲

一個(gè)主題的分區(qū)是kafka中的并行單元。發(fā)送到不同分區(qū)的消息可以由生產(chǎn)者并行發(fā)送，由不同的Broker并行寫入，并可以由不同的消費(fèi)者并行讀取。因此，更多的分區(qū)通常會(huì)導(dǎo)致更高的吞吐量，不過單從吞吐量的角度來看，我們能夠從每個(gè)Broker有10個(gè)分區(qū)的kafka集群，就已經(jīng)能夠達(dá)到最大的吞吐量了。您可能需要更多的主題分區(qū)來支持您的應(yīng)用程序邏輯。

但是，太多的分區(qū)可能導(dǎo)致更多的端到端的延遲。每個(gè)主題的分區(qū)越多，對生產(chǎn)者的批處理就越少。每個(gè)Broker的主題分區(qū)越多，每個(gè)follow副本獲取請求的開銷就越大。每個(gè)fetch請求必須去枚舉自己感興趣的分區(qū)，并且每個(gè)leader副本必須去檢查狀態(tài)，同時(shí)從請求的每個(gè)分區(qū)中去fetch數(shù)據(jù)，這通常會(huì)導(dǎo)致較小的磁盤I/O。因此，越多的分區(qū)，可能會(huì)導(dǎo)致更長的commit time和更高的cpu使用，最終導(dǎo)致較長的排隊(duì)延遲。

提交時(shí)間的增加和更高的CPU負(fù)載會(huì)導(dǎo)致所有共享同一個(gè)Kafka集群的客戶端端到端延遲的增加，即使是那些只生產(chǎn)和使用少量主題分區(qū)的客戶端來說，也是如此。

我們使用兩個(gè)topic來做此次測試。一個(gè)Topic有9個(gè)生產(chǎn)者生產(chǎn)5MB/s的數(shù)據(jù)，然后有一個(gè)對應(yīng)9個(gè)消費(fèi)者的消費(fèi)者組。這個(gè)實(shí)驗(yàn)持續(xù)了幾天，我們將這個(gè)主題中的分區(qū)數(shù)量從108個(gè)逐步增加到7200個(gè)(每個(gè)Broker8000個(gè))，每個(gè)步驟運(yùn)行一個(gè)小時(shí)。第二個(gè)主題在整個(gè)實(shí)驗(yàn)運(yùn)行期間有9個(gè)分區(qū)和9個(gè)生產(chǎn)者，每個(gè)生產(chǎn)者向一個(gè)分區(qū)和一個(gè)對應(yīng)的消費(fèi)者組(每個(gè)分區(qū)一個(gè))生產(chǎn)消息，該主題每秒生產(chǎn)一個(gè)512bytes的數(shù)據(jù)。

下圖顯示了分區(qū)數(shù)量對客戶端訪問9分區(qū)主題的99%的端到端延遲的影響，隨著每個(gè)broker上分區(qū)數(shù)的增加，clients的端到端延時(shí)大致呈線性增加趨勢。分區(qū)數(shù)的增加會(huì)推高broker上的CPU負(fù)載同時(shí)拖慢所有clients的備份，即使是對那些只與固定分區(qū)數(shù)量交互的clients而言，也會(huì)抬高端到端延遲。

為了減少延時(shí)，最好還是限制每個(gè)broker上的分區(qū)數(shù)，方法是減少總的分區(qū)數(shù)或擴(kuò)展集群。你還可以通過增加fetcher線程數(shù)量的方式來改善commit time。

broker節(jié)點(diǎn)負(fù)載對延遲的影響

我們已經(jīng)討論了Broker上的負(fù)載導(dǎo)致增加排隊(duì)延遲，從而增加了端到端的延遲，很容易看出為什么請求速率的增加會(huì)加劇排隊(duì)延遲，因?yàn)楦嗟恼埱髸?huì)導(dǎo)致更大的隊(duì)列。

broker節(jié)點(diǎn)上高資源利用率(磁盤或cpu)可能導(dǎo)致更高的隊(duì)列的延遲，并且延遲的增長會(huì)隨著資源利用率的增長呈指數(shù)級增長。這是一個(gè)可以有排隊(duì)理論解釋的已知屬性: Kingman公式證明等待某種資源的時(shí)間正比于資源繁忙程度/資源空閑程度(% of time resource is busy)/(% of time resource is idle)。

由于延遲隨資源利用率呈指數(shù)增長，如果broker中的任何資源的利用率接近100%，您可能會(huì)看到很高的延遲。通過減少每個(gè)Broker的資源使用(比如減少每個(gè)broker的鏈接數(shù)，請求以及分區(qū)數(shù))或擴(kuò)展集群來整體降低每個(gè)broker節(jié)點(diǎn)的資源使用率，在這種情況可以顯著降低延遲。保持負(fù)載在broker之間平均通常情況下是非常有用的，同時(shí)也可以均勻地或基于負(fù)載分布分區(qū)副本也能降低尾部延遲。

因此，通常情況下，負(fù)責(zé)kafka集群的SRE團(tuán)隊(duì)需要自動(dòng)檢測Broker節(jié)點(diǎn)上的高資源利用率(磁盤和CPU)，然后重新平衡集群上的分區(qū)，以便更均勻地在Broker之間重新分配負(fù)載，或者在需要時(shí)擴(kuò)展集群。而如果使用云廠商提供的kafka服務(wù)，則可以適當(dāng)避免這類事情的發(fā)生，因?yàn)樵品?wù)會(huì)去做相關(guān)的事情。

總結(jié)

我們已經(jīng)演示了，在為吞吐量擴(kuò)展客戶機(jī)和分區(qū)時(shí)的邊界延遲要求時(shí)可以通過限制每個(gè)broker的連接數(shù)、分區(qū)數(shù)和請求速率來限制每個(gè)broker的使用。

邊界尾延遲需要額外注意減少來自客戶機(jī)的任何突發(fā)(連接和請求)或應(yīng)用程序行為中的差異。

均勻加載的broker節(jié)點(diǎn)對于最小化尾部延遲也很重要，因?yàn)樗艿阶盥齜roker的影響。