[[245749]]

【51CTO.com原創(chuàng)稿件】Apache Kafka 是一款流行的分布式數(shù)據(jù)流平臺，它已經(jīng)廣泛地被諸如 New Relic(數(shù)據(jù)智能平臺)、Uber、Square(移動支付公司)等大型公司用來構(gòu)建可擴展的、高吞吐量的、且高可靠的實時數(shù)據(jù)流系統(tǒng)。

例如，在 New Relic 的生產(chǎn)環(huán)境中，Kafka 群集每秒能夠處理超過 1500 萬條消息，而且其數(shù)據(jù)聚合率接近 1Tbps。

可見，Kafka 大幅簡化了對于數(shù)據(jù)流的處理，因此它也獲得了眾多應(yīng)用開發(fā)人員和數(shù)據(jù)管理專家的青睞。

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

同時，它在自動化數(shù)據(jù)保留方面的限制，高流量的發(fā)布+訂閱(publish-subscribe，pub/sub)模式等，可能都會影響到您系統(tǒng)的性能。

可以毫不夸張地說，如果那些存放著數(shù)據(jù)流的系統(tǒng)無法按需擴容、或穩(wěn)定性不可靠的話，估計您經(jīng)常會寢食難安。

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

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

• Partitions(分區(qū))
• Consumers(消費者)
• Producers(生產(chǎn)者)
• Brokers(代理)

快速了解 Kafka 的概念與架構(gòu)

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

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

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

Message(消息)

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

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ū)都會被復(fù)制至少一到兩次。

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

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

Offset(偏移量)

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

Consumer(消費者)

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

Consumer group(消費組)

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

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

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

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

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

Lag(延遲)

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

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

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

針對 Partitions 的***實踐

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

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

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

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

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

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

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

• 首先，“熱”(有較高吞吐量)分區(qū)上的 Consumer 勢必會比同組中的其他 Consumer 處理更多的消息，因此很可能會導(dǎo)致出現(xiàn)在處理上和網(wǎng)絡(luò)上的瓶頸。
• 其次，那些為具有***數(shù)據(jù)速率的分區(qū)，所配置的***保留空間，會導(dǎo)致Topic 中其他分區(qū)的磁盤使用量也做相應(yīng)地增長。
• 第三，根據(jù)分區(qū)的 Leader 關(guān)系所實施的***均衡方案，比簡單地將 Leader 關(guān)系分散到所有 Broker 上，要更為復(fù)雜。在同一 Topic 中，“熱”分區(qū)會“承載”10 倍于其他分區(qū)的權(quán)重。

有關(guān) Topic Partition 的使用，可以參閱《Kafka Topic Partition的各種有效策略》https://blog.newrelic.com/engineering/effective-strategies-kafka-topic-partitioning/。

針對 Consumers 的***實踐

③如果 Consumers 運行的是比 Kafka 0.10 還要舊的版本，那么請馬上升級

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

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

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

④調(diào)優(yōu) Consumer 的套接字緩沖區(qū)(socket buffers)，以應(yīng)對數(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)絡(luò)帶寬延遲積(bandwidth-delay product)大于局域網(wǎng)(local areanetwork，LAN)時。

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

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

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

⑤設(shè)計具有高吞吐量的 Consumers，以便按需實施背壓(back-pressure)

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

如果是在 Java 虛擬機(JVM)中運行，Consumers 應(yīng)當(dāng)使用固定大小的緩沖區(qū)，而且***是使用堆外內(nèi)存(off-heap)。請參見 Disruptor 模式：http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf

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

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

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

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

針對 Producers 的***實踐

⑦配置 Producer，以等待各種確認

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

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

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

⑧為各個 Producer 配置 Retries

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

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

⑨為高吞吐量的 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è)定值將取決于如下幾個因素：

• Producer 數(shù)據(jù)速率(消息的大小和數(shù)量)
• 要生成的分區(qū)數(shù)
• 可用的內(nèi)存量

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

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

針對 Brokers 的***實踐

⑪在各個 Brokers 上，請壓縮 Topics 所需的內(nèi)存和 CPU 資源。

日志壓縮(請參見https://kafka.apache.org/documentation/#compaction)需要各個 Broker 上的堆棧(內(nèi)存)和 CPU 周期都能成功地配合實現(xiàn)而如果讓那些失敗的日志壓縮數(shù)據(jù)持續(xù)增長的話，則會給 Brokers 分區(qū)帶來風(fēng)險。

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

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

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

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

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

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

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

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

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

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

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

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

⑮按需修改 Apache Log4j 的各種屬性

詳細內(nèi)容可以參考：https://github.com/apache/kafka/blob/trunk/config/log4j.properties

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

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

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

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

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

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

⑱對于具有高吞吐量服務(wù)級別目標(biāo)(service level objectives，SLOs)的大型群集，請考慮為 Brokers 的子集隔離出不同的 Topic

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

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

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

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

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

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

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

總結(jié)

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

【51CTO原創(chuàng)稿件，合作站點轉(zhuǎn)載請注明原文作者和出處為51CTO.com】