Kafka在大型應用中的 20 項最佳實踐

原標題：Kafka如何作到1秒處理1500萬條消息？

Apache Kafka 是一款流行的分佈式數據流平臺，它已經普遍地被諸如 New Relic（數據智能平臺）、Uber、Square（移動支付公司）等大型公司用來構建可擴展的、高吞吐量的、且高可靠的實時數據流系統。

例如，在 New Relic 的生產環境中，Kafka 羣集每秒可以處理超過 1500 萬條消息，並且其數據聚合率接近 1Tbps。

可見，Kafka 大幅簡化了對於數據流的處理，所以它也得到了衆多應用開發人員和數據管理專家的青睞。

然而，在大型系統中 Kafka 的應用會比較複雜。若是您的 Consumers 沒法跟上數據流的話，各類消息每每在未被查看以前就已經消失掉了。

同時，它在自動化數據保留方面的限制，高流量的發佈+訂閱（publish-subscribe，pub/sub）模式等，可能都會影響到您系統的性能。

能夠絕不誇張地說，若是那些存放着數據流的系統沒法按需擴容、或穩定性不可靠的話，估計您常常會寢食難安。

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

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

• Partitions（分區）
• Consumers（消費者）
• Producers（生產者）
• Brokers（代理）

快速瞭解 Kafka 的概念與架構

Kafka 是一種高效的分佈式消息系統。在性能上，它具備內置的數據冗餘度與彈性，也具備高吞吐能力和可擴展性。

在功能上，它支持自動化的數據保存限制，可以以「流」的方式爲應用提供數據轉換，以及按照「鍵-值（key-value）」的建模關係「壓縮」數據流。

要了解各類最佳實踐，您須要首先熟悉以下關鍵術語：

Message（消息）

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

Producer（生產者）

Producer 將消息發佈到 Kafka 的 topics 上。Producer 決定向 topic 分區的發佈方式，如：輪詢的隨機方法、或基於消息鍵（key）的分區算法。

Broker（代理）

Kafka 以分佈式系統或集羣的方式運行。那麼羣集中的每一個節點稱爲一個 Broker。

Topic（主題）

Topic 是那些被髮布的數據記錄或消息的一種類別。消費者經過訂閱Topic，來讀取寫給它們的數據。

Topic Partition（主題分區）

不一樣的 Topic 被分爲不一樣的分區，而每一條消息都會被分配一個 Offset，一般每一個分區都會被複制至少一到兩次。

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

羣集中的全部 Broker 均可以做爲 Leader 和 Follower，可是一個 Broker 最多隻能有一個 Topic Partition 的副本。Leader 可被用來進行全部的讀寫操做。

Offset（偏移量）

單個分區中的每一條消息都被分配一個 Offset，它是一個單調遞增的整型數，可用來做爲分區中消息的惟一標識符。

Consumer（消費者）

Consumer 經過訂閱 Topic partition，來讀取 Kafka 的各類 Topic 消息。而後，消費類應用處理會收到消息，以完成指定的工做。

Consumer group（消費組）

Consumer 能夠按照 Consumer group 進行邏輯劃分。Topic Partition 被均衡地分配給組中的全部 Consumers。

所以，在同一個 Consumer group 中，全部的 Consumer 都以負載均衡的方式運做。

換言之，同一組中的每個 Consumer 都能看到每一條消息。若是某個 Consumer 處於「離線」狀態的話，那麼該分區將會被分配給同組中的另外一個 Consumer。這就是所謂的「再均衡（rebalance）」。

固然，若是組中的 Consumer 多於分區數，則某些 Consumer 將會處於閒置的狀態。

相反，若是組中的 Consumer 少於分區數，則某些 Consumer 會得到來自一個以上分區的消息。

Lag（延遲）

當 Consumer 的速度跟不上消息的產生速度時，Consumer 就會由於沒法從分區中讀取消息，而產生延遲。

延遲表示爲分區頭後面的 Offset 數量。從延遲狀態（到「追遇上來」）恢復正常所須要的時間，取決於 Consumer 每秒可以應對的消息速度。

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

針對 Partitions 的最佳實踐

①瞭解分區的數據速率，以確保提供合適的數據保存空間

此處所謂「分區的數據速率」是指數據的生成速率。換言之，它是由「平均消息大小」乘以「每秒消息數」得出的數據速率決定了在給定時間內，所能保證的數據保存空間的大小（以字節爲單位）。

若是您不知道數據速率的話，則沒法正確地計算出知足基於給定時間跨度的數據，所須要保存的空間大小。

同時，數據速率也可以標識出單個 Consumer 在不產生延時的狀況下，所須要支持的最低性能值。

②除非您有其餘架構上的須要，不然在寫 Topic 時請使用隨機分區

在您進行大型操做時，各個分區在數據速率上的良莠不齊是很是難以管理的。

其緣由來自於以下三個方面：

• 首先，「熱」（有較高吞吐量）分區上的 Consumer 勢必會比同組中的其餘 Consumer 處理更多的消息，所以極可能會致使出如今處理上和網絡上的瓶頸。
• 其次，那些爲具備最高數據速率的分區，所配置的最大保留空間，會致使Topic 中其餘分區的磁盤使用量也作相應地增加。
• 第三，根據分區的 Leader 關係所實施的最佳均衡方案，比簡單地將 Leader 關係分散到全部 Broker 上，要更爲複雜。在同一 Topic 中，「熱」分區會「承載」10 倍於其餘分區的權重。

有關 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 來協調 Consumer group，而許多已知的 Bug 會致使其長期處於再均衡狀態，或是直接致使再均衡算法的失敗（咱們稱之爲「再均衡風暴」）。

所以在再均衡期間，一個或多個分區會被分配給同一組中的每一個 Consumer。

而在再均衡風暴中，分區的全部權會持續在各個 Consumers 之間流轉，這反而阻礙了任何一個 Consumer 去真正獲取分區的全部權。

④調優 Consumer 的套接字緩衝區（socket buffers），以應對數據的高速流入

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

這兩個默認值對於高吞吐量的環境而言都過小了，特別是若是 Broker 和 Consumer 之間的網絡帶寬延遲積（bandwidth-delay product）大於局域網（local areanetwork，LAN）時。

對於延遲爲 1 毫秒或更多的高帶寬的網絡（如 10Gbps 或更高），請考慮將套接字緩衝區設置爲 8 或 16MB。

若是您的內存不足，也至少考慮設置爲 1MB。固然，您也能夠設置爲 -1，它會讓底層操做系統根據網絡的實際狀況，去調整緩衝區的大小。

可是，對於須要啓動「熱」分區的 Consumers 來講，自動調整可能不會那麼快。

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

一般，咱們應該保證系統只去處理其能力範圍內的數據，而不要超負荷「消費」，進而致使進程中斷「掛起」，或出現 Consume group 的溢出。

若是是在 Java 虛擬機（JVM）中運行，Consumers 應當使用固定大小的緩衝區，並且最好是使用堆外內存（off-heap）。請參見 Disruptor 模式：http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf

固定大小的緩衝區可以阻止 Consumer 將過多的數據拉到堆棧上，以致於 JVM 花費掉其全部的時間去執行垃圾回收，進而沒法履行其處理消息的本質工做。

⑥在 JVM 上運行各類 Consumers 時，請警戒垃圾回收對它們可能產生的影響

例如，長時間垃圾回收的停滯，可能致使 ZooKeeper 的會話被丟棄、或 Consumer group 處於再均衡狀態。

對於 Broker 來講也如此，若是垃圾回收停滯的時間太長，則會產生集羣掉線的風險。

針對 Producers 的最佳實踐

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

籍此 Producer 可以獲知消息是否真正被髮送到了 Broker 的分區上。在 Kafka 的 0.10.x 版本上，其設置是 Acks；而在 0.8.x 版本上，則爲 request.required.acks。

Kafka 經過複製，來提供容錯功能，所以單個節點的故障、或分區 Leader 關係的更改不會影響到系統的可用性。

若是您沒有用 Acks 來配置 Producer（或稱「fireand forget」）的話，則消息可能會悄然丟失。

⑧爲各個 Producer 配置 Retries

其默認值爲 3，固然是很是低的。不過，正確的設定值取決於您的應用程序，即：就那些對於數據丟失零容忍的應用而言，請考慮設置爲 Integer.MAX_VALUE（有效且最大）。

這樣將可以應對 Broker 的 Leader 分區出現沒法馬上響應 Produce 請求的狀況。

⑨爲高吞吐量的 Producer，調優緩衝區的大小

特別是 buffer.memory 和 batch.size（以字節爲單位）。因爲 batch.size 是按照分區設定的，而 Producer 的性能和內存的使用量，均可以與 Topic 中的分區數量相關聯。

所以，此處的設定值將取決於以下幾個因素：

• Producer 數據速率（消息的大小和數量）
• 要生成的分區數
• 可用的內存量

請記住，將緩衝區調大並不老是好事，若是 Producer 因爲某種緣由而失效了（例如，某個 Leader 的響應速度比確認還要慢），那麼在堆內內存（on-heap）中的緩衝的數據量越多，其須要回收的垃圾也就越多。

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

針對 Brokers 的最佳實踐

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

日誌壓縮（請參見https://kafka.apache.org/documentation/#compaction）須要各個 Broker 上的堆棧（內存）和 CPU 週期都能成功地配合實現而若是讓那些失敗的日誌壓縮數據持續增加的話，則會給 Brokers 分區帶來風險。

您能夠在 Broker 上調整 log.cleaner.dedupe.buffer.size 和 log.cleaner.threads 這兩個參數，可是請記住，這兩個值都會影響到各個 Brokers 上的堆棧使用。

若是某個 Broker 拋出 OutOfMemoryError 異常，那麼它將會被關閉、並可能形成數據的丟失。

而緩衝區的大小和線程的計數，則取決於須要被清除的 Topic Partition 數量、以及這些分區中消息的數據速率與密鑰的大小。

對於 Kafka 的 0.10.2.1 版本而言，經過 ERROR 條目來監控日誌清理程序的日誌文件，是檢測其線程可能出現問題的最可靠方法。

⑫經過網絡吞吐量來監控 Brokers

請監控發向（transmit，TX）和收向（receive，RX）的流量，以及磁盤的 I/O、磁盤的空間、以及 CPU 的使用率，並且容量規劃是維護羣集總體性能的關鍵步驟。

⑬在羣集的各個 Brokers 之間分配分區的 Leader 關係

Leader 一般會須要大量的網絡 I/O 資源。例如，當咱們將複製因子（replication factor）配置爲 三、並運行起來時。

Leader 必須首先獲取分區的數據，而後將兩套副本發送給另兩個 Followers，進而再傳輸到多個須要該數據的 Consumers 上。

所以在該例子中，單個 Leader 所使用的網絡 I/O，至少是 Follower 的四倍。並且，Leader 還可能須要對磁盤進行讀操做，而 Follower 只需進行寫操做。

⑭不要忽略監控 Brokers 的 in-sync replica（ISR）shrinks、under-replicatedpartitions 和 unpreferred leaders

這些都是集羣中潛在問題的跡象。例如，單個分區頻繁出現 ISR 收縮，則暗示着該分區的數據速率超過了 Leader 的能力，已沒法爲 Consumer 和其餘副本線程提供服務了。

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

詳細內容能夠參考：https://github.com/apache/kafka/blob/trunk/config/log4j.properties

Kafka 的 Broker 日誌記錄會耗費大量的磁盤空間，可是咱們卻不能徹底關閉它。

由於有時在發生事故以後，須要重建事件序列，那麼 Broker 日誌就會是咱們最好的、甚至是惟一的方法。

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

例如，在設定的 x 天內，若是未出現新的消息，您應該考慮該 Topic 是否已經失效，並將其從羣集中予以刪除。此舉可避免您花時間去管理羣集中被額外建立的元數據。

⑰對於那些具備持續高吞吐量的 Brokers，請提供足夠的內存，以免它們從磁盤子系統中進行讀操做

咱們應儘量地直接從操做系統的緩存中直接獲取分區的數據。然而，這就意味着您必須確保本身的 Consumers 可以跟得上「節奏」，而對於那些延遲的 Consumer 就只能強制 Broker 從磁盤中讀取了。

⑱對於具備高吞吐量服務級別目標（service level objectives，SLOs）的大型羣集，請考慮爲 Brokers 的子集隔離出不一樣的 Topic

至於如何肯定須要隔離的 Topics，則徹底取決於您本身的業務須要。例如，您有一些使用相同羣集的聯機事務處理（multipleonline transaction processing，OLTP）系統。

那麼將每一個系統的 Topics 隔離到不一樣 Brokers 子集中，則可以有助於限制潛在事件的影響半徑。

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

固然，最好仍是要儘可能避免這種狀況的發生。

⑳不要錯誤地認爲在本地主機上測試好 Broker，就能表明生產環境中的真實性能了

要知道，若是使用複製因子爲 1，並在環回接口上對分區所作的測試，是與大多數生產環境大相徑庭的。

在環回接口上網絡延遲幾乎能夠被忽略的，而在不涉及到複製的狀況下，接收 Leader 確認所需的時間則一樣會出現巨大的差別。

總結

但願上述各項建議可以有助於您更有效地去使用 Kafka。若是您想提升本身在 Kafka 方面的專業知識，請進一步查閱 Kafka 配套文檔中的「操做」部分，其中包含了有關操做羣集等實用信息。