Apache Kafka在大型應(yīng)用中的20項(xiàng)優(yōu)秀實(shí)踐

譯文
作者：陳峻編譯 2018-08-30 09:00:00

開(kāi)源

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

【51CTO.com快譯】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ù)聚合率接近1 Tbps。

可見(jiàn)，Kafka大幅簡(jiǎn)化了對(duì)于數(shù)據(jù)流的處理，因此它也獲得了眾多應(yīng)用開(kāi)發(fā)人員和數(shù)據(jù)管理專家的青睞。然而，在大型系統(tǒng)中Kafka的應(yīng)用會(huì)比較復(fù)雜。如果您的consumers無(wú)法跟上數(shù)據(jù)流的話，各種消息往往在未被查看之前就已經(jīng)消失掉了。同時(shí)，它在自動(dòng)化數(shù)據(jù)保留方面的限制，高流量的發(fā)布+訂閱（publish-subscribe，pub/sub）模式等，可能都會(huì)影響到您系統(tǒng)的性能。可以毫不夸張地說(shuō)，如果那些存放著數(shù)據(jù)流的系統(tǒng)無(wú)法按需擴(kuò)容、或穩(wěn)定性不可靠的話，估計(jì)您經(jīng)常會(huì)寢食難安了。

為了減少上述復(fù)雜性，我在此分享New Relic公司為Kafka集群在應(yīng)對(duì)高吞吐量方面的20項(xiàng)***實(shí)踐。我將從如下四個(gè)方面進(jìn)行展開(kāi)：

1. Partitions（分區(qū)）
2. Consumers（消費(fèi)者）
3. Producers（生產(chǎn)者）
4. 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ā)布方式，如：輪詢的隨機(jī)方法、或基于消息鍵（key）的分區(qū)算法。
• Broker（代理）：Kafka以分布式系統(tǒng)或集群的方式運(yùn)行。那么群集中的每個(gè)節(jié)點(diǎn)稱為一個(gè)broker。
• Topic（主題）：topic是那些被發(fā)布的數(shù)據(jù)記錄或消息的一種類別。消費(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（偏移量）：?jiǎn)蝹€(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)類應(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ì)獲得來(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ù)速率的話，則無(wú)法正確地計(jì)算出滿足基于給定時(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)致再均衡算法的失?。ㄎ覀兎Q之為“再均衡風(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)值為64 kB。而在Kafka的0.8.x版本中，參數(shù)socket.receive.buffer.bytes的默認(rèn)值為100 kB。這兩個(gè)默認(rèn)值對(duì)于高吞吐量的環(huán)境而言都太小了，特別是如果broker和consumer之間的網(wǎng)絡(luò)帶寬延遲積（bandwidth-delay product）大于局域網(wǎng)（local area network，LAN）時(shí)。對(duì)于延遲為1毫秒或更多的高帶寬的網(wǎng)絡(luò)（如10 Gbps或更高），請(qǐng)考慮將套接字緩沖區(qū)設(shè)置為8或16 MB。如果您的內(nèi)存不足，也至少考慮設(shè)置為1 MB。當(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ū)（請(qǐng)參見(jiàn)Disruptor模式：http://lmax-exchange.github.io/disruptor/files/Disruptor-1.0.pdf），而且***是使用堆外內(nèi)存（off-heap）。固定大小的緩沖區(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ì)話被丟棄、或consumer group處于再均衡狀態(tài)。對(duì)于broker來(lái)說(shuō)也如此，如果垃圾回收停滯的時(shí)間太長(zhǎng)，則會(huì)產(chǎ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（或稱“fire and forget”）的話，則消息可能會(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ì)給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ū)的大小和線程的計(jì)數(shù)，則取決于需要被清除的topic partition數(shù)量、以及這些分區(qū)中消息的數(shù)據(jù)速率與密鑰的大小。對(duì)于Kafka的0.10.2.1版本而言，通過(guò)ERROR條目來(lái)監(jiān)控日志清理程序的日志文件，是檢測(cè)其線程可能出現(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-replicated partitions和unpreferred leaders。這些都是集群中潛在問(wèn)題的跡象。例如，單個(gè)分區(qū)頻繁出現(xiàn)ISR收縮，則暗示著該分區(qū)的數(shù)據(jù)速率超過(guò)了leader的能力，已無(wú)法為consumer和其他副本線程提供服務(wù)了。
• 按需修改Apache Log4j（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ù)處理（multiple online 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ī)上測(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ù)制的情況下，接收l(shuí)eader確認(rèn)所需的時(shí)間則同樣會(huì)出現(xiàn)巨大的差異。

其他資源

希望上述各項(xiàng)建議能夠有助于您更有效地去使用Kafka。如果您想提高自己在Kafka方面的專業(yè)知識(shí)，請(qǐng)進(jìn)一步查閱Kafka配套文檔中的“操作”部分，其中包含了有關(guān)操作群集等實(shí)用信息。此外，Confluent（https://www.confluent.io/）也會(huì)定期舉行并發(fā)布各種在線討論，以幫助您更好地了解Kafka。

本文標(biāo)題：Apache Kafka在大型應(yīng)用中的20項(xiàng)優(yōu)秀實(shí)踐
轉(zhuǎn)載來(lái)源：http://www.dlmjj.cn/article/dpgeshd.html