Kafka如何做到1秒處理1500萬(wàn)條消息？

原創(chuàng)
作者：陳峻編譯 2018-10-11 09:33:51

開(kāi)發(fā)

架構(gòu)

開(kāi)發(fā)工具

Kafka 一位軟件工程師將通過(guò)本文向您呈現(xiàn) Apache Kafka 在大型應(yīng)用中的 20 項(xiàng)最佳實(shí)踐。

界首ssl適用于網(wǎng)站、小程序/APP、API接口等需要進(jìn)行數(shù)據(jù)傳輸應(yīng)用場(chǎng)景，ssl證書(shū)未來(lái)市場(chǎng)廣闊！成為創(chuàng)新互聯(lián)建站的ssl證書(shū)銷(xiāo)售渠道，可以享受市場(chǎng)價(jià)格4-6折優(yōu)惠！如果有意向歡迎電話(huà)聯(lián)系或者加微信：13518219792（備注：SSL證書(shū)合作）期待與您的合作！

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

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

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

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

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

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

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

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

• 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ù)語(yǔ)：

Message(消息)

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

Producer(生產(chǎn)者)

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

Broker(代理)

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

Topic(主題)

Topic 是那些被發(fā)布的數(shù)據(jù)記錄或消息的一種類(lèi)別。消費(fèi)者通過(guò)訂閱Topic，來(lái)讀取寫(xiě)給它們的數(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 可被用來(lái)進(jìn)行所有的讀寫(xiě)操作。

Offset(偏移量)

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

Consumer(消費(fèi)者)

Consumer 通過(guò)訂閱 Topic partition，來(lái)讀取 Kafka 的各種 Topic 消息。然后，消費(fèi)類(lè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 處于“離線(xiàn)”狀態(tài)的話(huà)，那么該分區(qū)將會(huì)被分配給同組中的另一個(gè) Consumer。這就是所謂的“再均衡(rebalance)”。

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

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

Lag(延遲)

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

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

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

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

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

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

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

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

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

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

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

• 首先，“熱”(有較高吞吐量)分區(qū)上的 Consumer 勢(shì)必會(huì)比同組中的其他 Consumer 處理更多的消息，因此很可能會(huì)導(dǎo)致出現(xiàn)在處理上和網(wǎng)絡(luò)上的瓶頸。
• 其次，那些為具有***數(shù)據(jù)速率的分區(qū)，所配置的***保留空間，會(huì)導(dǎo)致Topic 中其他分區(qū)的磁盤(pán)使用量也做相應(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/engineering/effective-strategies-kafka-topic-partitioning/。

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

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

在 0.8.x 版中，Consumer 使用 Apache ZooKeeper 來(lái)協(xié)調(diào) Consumer group，而許多已知的 Bug 會(huì)導(dǎo)致其長(zhǎng)期處于再均衡狀態(tài)，或是直接導(dǎo)致再均衡算法的失敗(我們稱(chēng)之為“再均衡風(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)對(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 來(lái)說(shuō)，自動(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)。請(qǐng)參見(jiàn) Disruptor 模式：http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf

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

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

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

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

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

⑦配置 Producer，以等待各種確認(rèn)

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

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

如果您沒(méi)有用 Acks 來(lái)配置 Producer(或稱(chēng)“fireand forget”)的話(huà)，則消息可能會(huì)悄然丟失。

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

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

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

⑨為高吞吐量的 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ù)量越多，其需要回收的垃圾也就越多。

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

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

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

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

您可以在 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ū)的大小和線(xiàn)程的計(jì)數(shù)，則取決于需要被清除的 Topic Partition 數(shù)量、以及這些分區(qū)中消息的數(shù)據(jù)速率與密鑰的大小。

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

?通過(guò)網(wǎng)絡(luò)吞吐量來(lái)監(jiān)控 Brokers

請(qǐng)監(jiān)控發(fā)向(transmit，TX)和收向(receive，RX)的流量，以及磁盤(pán)的 I/O、磁盤(pán)的空間、以及 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)行起來(lái)時(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ì)磁盤(pán)進(jìn)行讀操作，而 Follower 只需進(jìn)行寫(xiě)操作。

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

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

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

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

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

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

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

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

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

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

?對(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 子集中，則能夠有助于限制潛在事件的影響半徑。

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

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

?不要錯(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 方面的專(zhuān)業(yè)知識(shí)，請(qǐng)進(jìn)一步查閱 Kafka 配套文檔中的“操作”部分，其中包含了有關(guān)操作群集等實(shí)用信息。

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

當(dāng)前題目：Kafka如何做到1秒處理1500萬(wàn)條消息？
標(biāo)題路徑：http://uogjgqi.cn/article/cdecsje.html