摘要：例如，在的生產(chǎn)環(huán)境中，群集每秒能夠處理超過萬條消息，而且其數(shù)據(jù)聚合率接近。為高吞吐量的，調(diào)優(yōu)緩沖區(qū)的大小特別是和以字節(jié)為單位。

來源：51CTO技術(shù)棧（ID：blog51cto）

Apache Kafka是一款流行的分布式數(shù)據(jù)流平臺(tái)，它已經(jīng)廣泛地被諸如New Relic（數(shù)據(jù)智能平臺(tái)）、Uber、Square（移動(dòng)支付公司）等大型公司用來構(gòu)建可擴(kuò)展的、高吞吐量的、高可靠的實(shí)時(shí)數(shù)據(jù)流系統(tǒng)。

例如，在New Relic的生產(chǎn)環(huán)境中，Kafka群集每秒能夠處理超過1500萬條消息，而且其數(shù)據(jù)聚合率接近1Tbps。可見，Kafka大幅簡(jiǎn)化了對(duì)于數(shù)據(jù)流的處理，因此它也獲得了眾多應(yīng)用開發(fā)人員和數(shù)據(jù)管理專家的青睞。

然而，在大型系統(tǒng)中Kafka的應(yīng)用會(huì)比較復(fù)雜。如果你的Consumers無法跟上數(shù)據(jù)流的話，各種消息往往在未被查看之前就已經(jīng)消失掉了。

同時(shí)，它在自動(dòng)化數(shù)據(jù)保留方面的限制，高流量的發(fā)布+訂閱（publish-subscribe，pub/sub）模式等，可能都會(huì)影響到系統(tǒng)的性能。可以毫不夸張地說，如果那些存放著數(shù)據(jù)流的系統(tǒng)無法按需擴(kuò)容、或穩(wěn)定性不可靠的話，估計(jì)大家經(jīng)常會(huì)寢食難安。

為了減少上述復(fù)雜性，我在此分享New Relic公司為Kafka集群在應(yīng)對(duì)高吞吐量方面的20項(xiàng)最佳實(shí)踐。

我將從如下四個(gè)方面進(jìn)行展開：

Partitions（分區(qū)）

Consumers（消費(fèi)者）

Producers（生產(chǎn)者）

Brokers（代理）

一、快速了解Kafka的概念與架構(gòu)
Kafka是一種高效的分布式消息系統(tǒng)。在性能上，它具有內(nèi)置的數(shù)據(jù)冗余度與彈性，也具有高吞吐能力和可擴(kuò)展性。

在功能上，它支持自動(dòng)化的數(shù)據(jù)保存限制，能夠以“流”的方式為應(yīng)用提供數(shù)據(jù)轉(zhuǎn)換，以及按照“鍵-值（key-value）”的建模關(guān)系“壓縮”數(shù)據(jù)流。

要了解各種最佳實(shí)踐，首先需要熟悉如下關(guān)鍵術(shù)語：

Message（消息）

Kafka中的一條記錄或數(shù)據(jù)單位。每條消息都有一個(gè)鍵和對(duì)應(yīng)的一個(gè)值，有時(shí)還會(huì)有可選的消息頭。

Producer（生產(chǎn)者）

Producer將消息發(fā)布到Kafka的topics上。Producer決定向topic分區(qū)的發(fā)布方式，如：輪詢的隨機(jī)方法、或基于消息鍵（key）的分區(qū)算法。

Broker（代理）

Kafka以分布式系統(tǒng)或集群的方式運(yùn)行，那么群集中的每個(gè)節(jié)點(diǎn)稱為一個(gè)Broker。

Topic（主題）

Topic是那些被發(fā)布的數(shù)據(jù)記錄或消息的一種類別。消費(fèi)者通過訂閱Topic來讀取寫給它們的數(shù)據(jù)。

Topic Partition（主題分區(qū)）

不同的Topic被分為不同的分區(qū)，而每一條消息都會(huì)被分配一個(gè)Offset，通常每個(gè)分區(qū)都會(huì)被復(fù)制至少一到兩次。

每個(gè)分區(qū)都有一個(gè)Leader和存放在各個(gè)Follower上的一到多個(gè)副本（即：數(shù)據(jù)的副本），此法可防止某個(gè)Broker的失效。

群集中的所有Broker都可以作為L(zhǎng)eader和Follower，但是一個(gè)Broker最多只能有一個(gè)Topic Partition的副本。Leader可被用來進(jìn)行所有的讀寫操作。

Offset（偏移量）

單個(gè)分區(qū)中的每一條消息都被分配一個(gè)Offset，它是一個(gè)單調(diào)遞增的整型數(shù)，可用來作為分區(qū)中消息的唯一標(biāo)識(shí)符。

Consumer（消費(fèi)者）

Consumer通過訂閱Topic partition，來讀取Kafka的各種Topic消息。然后，消費(fèi)類應(yīng)用處理會(huì)收到消息，以完成指定的工作。

Consumer group（消費(fèi)組）

Consumer可以按照Consumer group進(jìn)行邏輯劃分。Topic Partition被均衡地分配給組中的所有Consumers。

因此，在同一個(gè)Consumer group中，所有的Consumer都以負(fù)載均衡的方式運(yùn)作。

換言之，同一組中的每一個(gè)Consumer都能群組看到分配給他的相應(yīng)分區(qū)的所有消息。如果某個(gè)Consumer處于“離線”狀態(tài)的話，那么該分區(qū)將會(huì)被分配給同組中的另一個(gè)Consumer。這就是所謂的“再均衡（rebalance）”。

當(dāng)然，如果組中的Consumer多于分區(qū)數(shù)，則某些Consumer將會(huì)處于閑置的狀態(tài)。

相反，如果組中的Consumer少于分區(qū)數(shù)，則某些Consumer會(huì)獲得來自一個(gè)以上分區(qū)的消息。

Lag（延遲）

當(dāng)Consumer的速度跟不上消息的產(chǎn)生速度時(shí)，Consumer就會(huì)因?yàn)闊o法從分區(qū)中讀取消息，而產(chǎn)生延遲。

延遲表示為分區(qū)頭后面的Offset數(shù)量。從延遲狀態(tài)（到“追趕上來”）恢復(fù)正常所需要的時(shí)間，取決于Consumer每秒能夠應(yīng)對(duì)的消息速度。

其公式如下：time=messages/(consume rate per second - produce rate per second)

1針對(duì)Partitions
1）了解分區(qū)的數(shù)據(jù)速率，以確保提供合適的數(shù)據(jù)保存空間

此處所謂“分區(qū)的數(shù)據(jù)速率”是指數(shù)據(jù)的生成速率。換言之，它是由“平均消息大小”乘以“每秒消息數(shù)”得出的數(shù)據(jù)速率決定了在給定時(shí)間內(nèi)，所能保證的數(shù)據(jù)保存空間的大小（以字節(jié)為單位）。

如果你不知道數(shù)據(jù)速率的話，則無法正確地計(jì)算出滿足基于給定時(shí)間跨度的數(shù)據(jù)，所需要保存的空間大小。

同時(shí)，數(shù)據(jù)速率也能夠標(biāo)識(shí)出單個(gè)Consumer在不產(chǎn)生延時(shí)的情況下，所需要支持的最低性能值。

2）除非有其他架構(gòu)上的需要，否則在寫Topic時(shí)請(qǐng)使用隨機(jī)分區(qū)

在進(jìn)行大型操作時(shí)，各個(gè)分區(qū)在數(shù)據(jù)速率上的參差不齊是非常難以管理的。

其原因來自于如下三個(gè)方面：

首先，“熱”（有較高吞吐量）分區(qū)上的Consumer勢(shì)必會(huì)比同組中的其他Consumer處理更多的消息，因此很可能會(huì)導(dǎo)致出現(xiàn)在處理上和網(wǎng)絡(luò)上的瓶頸。

其次，那些為具有最高數(shù)據(jù)速率的分區(qū)，所配置的最大保留空間，會(huì)導(dǎo)致Topic中其他分區(qū)的磁盤使用量也做相應(yīng)地增長(zhǎng)。

第三，根據(jù)分區(qū)的Leader關(guān)系所實(shí)施的最佳均衡方案，比簡(jiǎn)單地將Leader關(guān)系分散到所有Broker上，要更為復(fù)雜。在同一Topic中，“熱”分區(qū)會(huì)“承載”10倍于其他分區(qū)的權(quán)重。

有關(guān)Topic Partition的使用，可以參閱《Kafka Topic Partition的各種有效策略》

參考鏈接：

https://blog.newrelic.com/eng...

2針對(duì)Consumers
3）如果Consumers運(yùn)行的是比Kafka 0.10還要舊的版本，那么請(qǐng)馬上升級(jí)

在0.8.x版中，Consumer使用Apache ZooKeeper來協(xié)調(diào)Consumer group，而許多已知的Bug會(huì)導(dǎo)致其長(zhǎng)期處于再均衡狀態(tài)，或是直接導(dǎo)致再均衡算法的失敗（我們稱之為“再均衡風(fēng)暴”）。

因此在再均衡期間，一個(gè)或多個(gè)分區(qū)會(huì)被分配給同一組中的每個(gè)Consumer。

而在再均衡風(fēng)暴中，分區(qū)的所有權(quán)會(huì)持續(xù)在各個(gè)Consumers之間流轉(zhuǎn)，這反而阻礙了任何一個(gè)Consumer去真正獲取分區(qū)的所有權(quán)。

4）調(diào)優(yōu)Consumer的套接字緩沖區(qū)（socket buffers），以應(yīng)對(duì)數(shù)據(jù)的高速流入

在Kafka的0.10.x版本中，參數(shù)receive.buffer.bytes的默認(rèn)值為64KB。而在Kafka的0.8.x版本中，參數(shù)socket.receive.buffer.bytes的默認(rèn)值為100KB。

這兩個(gè)默認(rèn)值對(duì)于高吞吐量的環(huán)境而言都太小了，特別是如果Broker和Consumer之間的網(wǎng)絡(luò)帶寬延遲積（bandwidth-delay product）大于局域網(wǎng)（local areanetwork，LAN）時(shí)。

對(duì)于延遲為1毫秒或更多的高帶寬的網(wǎng)絡(luò)（如10Gbps或更高），請(qǐng)考慮將套接字緩沖區(qū)設(shè)置為8或16MB。

如果內(nèi)存不足，也至少考慮設(shè)置為1MB。當(dāng)然，也可以設(shè)置為-1，它會(huì)讓底層操作系統(tǒng)根據(jù)網(wǎng)絡(luò)的實(shí)際情況，去調(diào)整緩沖區(qū)的大小。

但是，對(duì)于需要啟動(dòng)“熱”分區(qū)的Consumers來說，自動(dòng)調(diào)整可能不會(huì)那么快。

5）設(shè)計(jì)具有高吞吐量的Consumers，以便按需實(shí)施背壓（back-pressure）

通常，我們應(yīng)該保證系統(tǒng)只去處理其能力范圍內(nèi)的數(shù)據(jù)，而不要超負(fù)荷“消費(fèi)”，進(jìn)而導(dǎo)致進(jìn)程中斷“掛起”，或出現(xiàn)Consume group的溢出。

如果是在Java虛擬機(jī)（JVM）中運(yùn)行，Consumers應(yīng)當(dāng)使用固定大小的緩沖區(qū)，而且最好是使用堆外內(nèi)存（off-heap）。

請(qǐng)參見Disruptor模式：

http://lmax-exchange.github.i...

固定大小的緩沖區(qū)能夠阻止Consumer將過多的數(shù)據(jù)拉到堆棧上，以至于JVM花費(fèi)掉其所有的時(shí)間去執(zhí)行垃圾回收，進(jìn)而無法履行其處理消息的本質(zhì)工作。

6）在JVM上運(yùn)行各種Consumers時(shí)，請(qǐng)警惕垃圾回收對(duì)它們可能產(chǎn)生的影響

例如，長(zhǎng)時(shí)間垃圾回收的停滯，可能導(dǎo)致ZooKeeper的會(huì)話被丟棄、或Consumer group處于再均衡狀態(tài)。

對(duì)于Broker來說也如此，如果垃圾回收停滯的時(shí)間太長(zhǎng)，則會(huì)產(chǎn)生集群掉線的風(fēng)險(xiǎn)。

3針對(duì)Producers
7）配置Producer，以等待各種確認(rèn)

籍此Producer能夠獲知消息是否真正被發(fā)送到了Broker的分區(qū)上。在Kafka的0.10.x版本上，其設(shè)置是Acks；而在0.8.x版本上，則為request.required.acks。

Kafka通過復(fù)制，來提供容錯(cuò)功能，因此單個(gè)節(jié)點(diǎn)的故障、或分區(qū)Leader關(guān)系的更改不會(huì)影響到系統(tǒng)的可用性。

如果沒有用Acks來配置Producer（或稱“fireand forget”）的話，則消息可能會(huì)悄然丟失。

8）為各個(gè)Producer配置Retries

其默認(rèn)值為3，當(dāng)然是非常低的。不過，正確的設(shè)定值取決于你的應(yīng)用程序，即：就那些對(duì)于數(shù)據(jù)丟失零容忍的應(yīng)用而言，請(qǐng)考慮設(shè)置為Integer.MAX_VALUE（有效且最大）。

這樣將能夠應(yīng)對(duì)Broker的Leader分區(qū)出現(xiàn)無法立刻響應(yīng)Produce請(qǐng)求的情況。

9）為高吞吐量的Producer，調(diào)優(yōu)緩沖區(qū)的大小

特別是buffer.memory和batch.size（以字節(jié)為單位）。由于batch.size是按照分區(qū)設(shè)定的，而Producer的性能和內(nèi)存的使用量，都可以與Topic中的分區(qū)數(shù)量相關(guān)聯(lián)。

因此，此處的設(shè)定值將取決于如下幾個(gè)因素：

Producer數(shù)據(jù)速率（消息的大小和數(shù)量）；

要生成的分區(qū)數(shù)；

可用的內(nèi)存量。

請(qǐng)記住，將緩沖區(qū)調(diào)大并不總是好事，如果Producer由于某種原因而失效了（例如，某個(gè)Leader的響應(yīng)速度比確認(rèn)還要慢），那么在堆內(nèi)內(nèi)存（on-heap）中的緩沖的數(shù)據(jù)量越多，其需要回收的垃圾也就越多。

10）檢測(cè)應(yīng)用程序，以跟蹤諸如生成的消息數(shù)、平均消息大小、以及已使用的消息數(shù)等指標(biāo)

4針對(duì)Brokers
11）在各個(gè)Brokers上，請(qǐng)壓縮Topics所需的內(nèi)存和CPU資

日志壓縮需要各個(gè)Broker上的堆棧（內(nèi)存）和CPU周期都能成功地配合實(shí)現(xiàn)，而如果讓那些失敗的日志壓縮數(shù)據(jù)持續(xù)增長(zhǎng)的話，則會(huì)給Brokers分區(qū)帶來風(fēng)險(xiǎn)。

請(qǐng)參見：

https://kafka.apache.org/docu...

你可以在Broker上調(diào)整log.cleaner.dedupe.buffer.size和log.cleaner.threads這兩個(gè)參數(shù)，但是請(qǐng)記住，這兩個(gè)值都會(huì)影響到各個(gè)Brokers上的堆棧使用。

如果某個(gè)Broker拋出OutOfMemoryError異常，那么它將會(huì)被關(guān)閉、并可能造成數(shù)據(jù)的丟失。

而緩沖區(qū)的大小和線程的計(jì)數(shù)，則取決于需要被清除的Topic Partition數(shù)量、以及這些分區(qū)中消息的數(shù)據(jù)速率與密鑰的大小。

對(duì)于Kafka的0.10.2.1版本而言，通過ERROR條目來監(jiān)控日志清理程序的日志文件，是檢測(cè)其線程可能出現(xiàn)問題的最可靠方法。

12）通過網(wǎng)絡(luò)吞吐量來監(jiān)控Brokers

請(qǐng)監(jiān)控發(fā)向（transmit，TX）和收向（receive，RX）的流量，以及磁盤的I/O、磁盤的空間和CPU的使用率，而且容量規(guī)劃是維護(hù)群集整體性能的關(guān)鍵步驟。

13）在群集的各個(gè)Brokers之間分配分區(qū)的Leader關(guān)系

Leader通常會(huì)需要大量的網(wǎng)絡(luò)I/O資源。例如，當(dāng)我們將復(fù)制因子（replication factor）配置為3、并運(yùn)行起來時(shí)。

Leader必須首先獲取分區(qū)的數(shù)據(jù)，然后將兩套副本發(fā)送給另兩個(gè)Followers，進(jìn)而再傳輸?shù)蕉鄠€(gè)需要該數(shù)據(jù)的Consumers上。

因此在該例子中，單個(gè)Leader所使用的網(wǎng)絡(luò)I/O，至少是Follower的四倍。而且，Leader還可能需要對(duì)磁盤進(jìn)行讀操作，而Follower只需進(jìn)行寫操作。

14）不要忽略監(jiān)控Brokers的in-sync replica（ISR）shrinks、under-replicatedpartitions和unpreferred leaders

這些都是集群中潛在問題的跡象。例如，單個(gè)分區(qū)頻繁出現(xiàn)ISR收縮，則暗示著該分區(qū)的數(shù)據(jù)速率超過了Leader的能力，已無法為Consumer和其他副本線程提供服務(wù)了。

15）按需修改Apache Log4j的各種屬性

詳細(xì)內(nèi)容可以參考：

https://github.com/apache/kaf...

Kafka的Broker日志記錄會(huì)耗費(fèi)大量的磁盤空間，但是我們卻不能完全關(guān)閉它。

因?yàn)橛袝r(shí)在發(fā)生事故之后，需要重建事件序列，那么Broker日志就會(huì)是我們最好的、甚至是唯一的方法。

16）禁用Topic的自動(dòng)創(chuàng)建，或針對(duì)那些未被使用的Topics建立清除策略

例如，在設(shè)定的x天內(nèi)，如果未出現(xiàn)新的消息，你應(yīng)該考慮該Topic是否已經(jīng)失效，并將其從群集中予以刪除。此舉可避免花時(shí)間去管理群集中被額外創(chuàng)建的元數(shù)據(jù)。

17）對(duì)于那些具有持續(xù)高吞吐量的Brokers，請(qǐng)?zhí)峁┳銐虻膬?nèi)存，以避免它們從磁盤子系統(tǒng)中進(jìn)行讀操作

我們應(yīng)盡可能地直接從操作系統(tǒng)的緩存中直接獲取分區(qū)的數(shù)據(jù)。然而，這就意味著你必須確保自己的Consumers能夠跟得上“節(jié)奏”，而對(duì)于那些延遲的Consumer就只能強(qiáng)制Broker從磁盤中讀取了。

18）對(duì)于具有高吞吐量服務(wù)級(jí)別目標(biāo)（service level objectives，SLOs）的大型群集，請(qǐng)考慮為Brokers的子集隔離出不同的Topic

至于如何確定需要隔離的Topics，則完全取決于自己的業(yè)務(wù)需要。例如，你有一些使用相同群集的聯(lián)機(jī)事務(wù)處理（multipleonline transaction processing，OLTP）系統(tǒng)。

那么將每個(gè)系統(tǒng)的Topics隔離到不同Brokers子集中，則能夠有助于限制潛在事件的影響半徑。

19）在舊的客戶端上使用新的Topic消息格式。應(yīng)當(dāng)代替客戶端，在各個(gè)Brokers上加載額外的格式轉(zhuǎn)換服務(wù)

當(dāng)然，最好還是要盡量避免這種情況的發(fā)生

20）不要錯(cuò)誤地認(rèn)為在本地主機(jī)上測(cè)試好Broker，就能代表生產(chǎn)環(huán)境中的真實(shí)性能了

要知道，如果使用復(fù)制因子為1，并在環(huán)回接口上對(duì)分區(qū)所做的測(cè)試，是與大多數(shù)生產(chǎn)環(huán)境截然不同的。

在環(huán)回接口上網(wǎng)絡(luò)延遲幾乎可以被忽略的，而在不涉及到復(fù)制的情況下，接收Leader確認(rèn)所需的時(shí)間則同樣會(huì)出現(xiàn)巨大的差異。

二、總結(jié)
希望上述各項(xiàng)建議能夠有助于大家更有效地去使用Kafka。如果你想提高自己在Kafka方面的專業(yè)知識(shí)，請(qǐng)進(jìn)一步查閱Kafka配套文檔中的“操作”部分，其中包含了有關(guān)操作群集等實(shí)用信息