[[245749]]

【51CTO.com原創(chuàng)稿件】Apache Kafka 是一款流行的分布式數(shù)據(jù)流平臺，它已經(jīng)廣泛地被諸如 New Relic(數(shù)據(jù)智能平臺)、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 大幅簡化了對于數(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)對高吞吐量方面的 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è)鍵和對應(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 都可以作為 Leader 和 Follower，但是一個(gè) Broker 最多只能有一個(gè) Topic Partition 的副本。Leader 可被用來進(jìn)行所有的讀寫操作。

Offset(偏移量)

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

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 都能看到每一條消息。如果某個(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)對的消息速度。

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

針對 Partitions 的***實(shí)踐

①了解分區(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)識出單個(gè) Consumer 在不產(chǎn)生延時(shí)的情況下，所需要支持的***性能值。

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

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

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

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

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

針對 Consumers 的***實(shí)踐

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

在 0.8.x 版中，Consumer 使用 Apache ZooKeeper 來協(xié)調(diào) Consumer group，而許多已知的 Bug 會(huì)導(dǎo)致其長期處于再均衡狀態(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)。

④調(diào)優(yōu) Consumer 的套接字緩沖區(qū)(socket buffers)，以應(yīng)對數(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)值對于高吞吐量的環(huán)境而言都太小了，特別是如果 Broker 和 Consumer 之間的網(wǎng)絡(luò)帶寬延遲積(bandwidth-delay product)大于局域網(wǎng)(local areanetwork，LAN)時(shí)。

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

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

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

⑤設(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)。請參見 Disruptor 模式：http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf

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

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

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

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

針對 Producers 的***實(shí)踐

⑦配置 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ì)悄然丟失。

⑧為各個(gè) Producer 配置 Retries

其默認(rèn)值為 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è)定值將取決于如下幾個(gè)因素：

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

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

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

針對 Brokers 的***實(shí)踐

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

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

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

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

而緩沖區(qū)的大小和線程的計(jì)數(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ī)劃是維護(hù)群集整體性能的關(guān)鍵步驟。

⑬在群集的各個(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 還可能需要對磁盤進(jìn)行讀操作，而 Follower 只需進(jìn)行寫操作。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

總結(jié)

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

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