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

來源：51CTO技術棧（ID：blog51cto）

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

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

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

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

為了減少上述復雜性，我在此分享New Relic公司為Kafka集群在應對高吞吐量方面的20項最佳實踐。

我將從如下四個方面進行展開：

Partitions（分區(qū)）

Consumers（消費者）

Producers（生產(chǎn)者）

Brokers（代理）

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

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

要了解各種最佳實踐，首先需要熟悉如下關鍵術語：

Message（消息）

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

Producer（生產(chǎn)者）

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

Broker（代理）

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

Topic（主題）

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

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

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

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

群集中的所有Broker都可以作為Leader和Follower，但是一個Broker最多只能有一個Topic Partition的副本。Leader可被用來進行所有的讀寫操作。

Offset（偏移量）

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

Consumer（消費者）

Consumer通過訂閱Topic partition，來讀取Kafka的各種Topic消息。然后，消費類應用處理會收到消息，以完成指定的工作。

Consumer group（消費組）

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

因此，在同一個Consumer group中，所有的Consumer都以負載均衡的方式運作。

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

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

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

Lag（延遲）

當Consumer的速度跟不上消息的產(chǎn)生速度時，Consumer就會因為無法從分區(qū)中讀取消息，而產(chǎn)生延遲。

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

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

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

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

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

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

2）除非有其他架構上的需要，否則在寫Topic時請使用隨機分區(qū)

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

其原因來自于如下三個方面：

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

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

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

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

參考鏈接：

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

2針對Consumers
3）如果Consumers運行的是比Kafka 0.10還要舊的版本，那么請馬上升級

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

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

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

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

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

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

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

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

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

5）設計具有高吞吐量的Consumers，以便按需實施背壓（back-pressure）

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

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

請參見Disruptor模式：

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

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

6）在JVM上運行各種Consumers時，請警惕垃圾回收對它們可能產(chǎn)生的影響

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

對于Broker來說也如此，如果垃圾回收停滯的時間太長，則會產(chǎn)生集群掉線的風險。

3針對Producers
7）配置Producer，以等待各種確認

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

Kafka通過復制，來提供容錯功能，因此單個節(jié)點的故障、或分區(qū)Leader關系的更改不會影響到系統(tǒng)的可用性。

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

8）為各個Producer配置Retries

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

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

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

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

因此，此處的設定值將取決于如下幾個因素：

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

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

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

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

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

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

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

請參見：

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Kafka的Broker日志記錄會耗費大量的磁盤空間，但是我們卻不能完全關閉它。

因為有時在發(fā)生事故之后，需要重建事件序列，那么Broker日志就會是我們最好的、甚至是唯一的方法。

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

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

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

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

18）對于具有高吞吐量服務級別目標（service level objectives，SLOs）的大型群集，請考慮為Brokers的子集隔離出不同的Topic

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

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

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

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

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

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

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

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