Kafka的生產(chǎn)者優(yōu)秀架構(gòu)設(shè)計(jì)的示例分析

本篇文章給大家分享的是有關(guān)Kafka的生產(chǎn)者優(yōu)秀架構(gòu)設(shè)計(jì)的示例分析，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話(huà)不多說(shuō)，跟著小編一起來(lái)看看吧。

Kafka 是一個(gè)高吞吐量的分布式的發(fā)布訂閱消息系統(tǒng)，在全世界都很流行，在大數(shù)據(jù)項(xiàng)目里面使用尤其頻繁。筆者看過(guò)多個(gè)大數(shù)據(jù)開(kāi)源產(chǎn)品的源碼，感覺(jué) Kafka 的源碼是其中質(zhì)量比較上乘的一個(gè)，這得益于作者高超的編碼水平和高超的架構(gòu)設(shè)計(jì)能力。

Kafka 的核心源碼分為兩部分：客戶(hù)端源碼和服務(wù)端源碼，客戶(hù)端又分為生產(chǎn)者和消費(fèi)者，而個(gè)人認(rèn)為 Kafka 的源碼里面生產(chǎn)者的源碼技術(shù)含量最高，所以今天給大家剖析 Kafka 的生產(chǎn)者的架構(gòu)設(shè)計(jì)，Kafka 是一個(gè)飛速發(fā)展的消息系統(tǒng)，其架構(gòu)也在一直演進(jìn)中，我們今天分析的 Kafka 的版本是比較成熟穩(wěn)定的 Kafka1.0.0 版本源碼。

圖1 Kafka核心模塊

生產(chǎn)者流程概述 

先給大家介紹一下生產(chǎn)者的大概的運(yùn)行的流程。

圖2 Kafka運(yùn)行方式

如上圖所示：步驟一：一條消息過(guò)來(lái)首先會(huì)被封裝成為一個(gè) ProducerRecord 對(duì)象。

步驟二：接下來(lái)要對(duì)這個(gè)對(duì)象進(jìn)行序列化，因?yàn)?Kafka 的消息需要從客戶(hù)端傳到服務(wù)端，涉及到網(wǎng)絡(luò)傳輸，所以需要實(shí)現(xiàn)序列。Kafka 提供了默認(rèn)的序列化機(jī)制，也支持自定義序列化（這種設(shè)計(jì)也值得我們積累，提高項(xiàng)目的擴(kuò)展性）。

步驟三：消息序列化完了以后，對(duì)消息要進(jìn)行分區(qū)，分區(qū)的時(shí)候需要獲取集群的元數(shù)據(jù)。分區(qū)的這個(gè)過(guò)程很關(guān)鍵，因?yàn)檫@個(gè)時(shí)候就決定了，我們的這條消息會(huì)被發(fā)送到 Kafka 服務(wù)端到哪個(gè)主題的哪個(gè)分區(qū)了。

步驟四：分好區(qū)的消息不是直接被發(fā)送到服務(wù)端，而是放入了生產(chǎn)者的一個(gè)緩存里面。在這個(gè)緩存里面，多條消息會(huì)被封裝成為一個(gè)批次（batch），默認(rèn)一個(gè)批次的大小是 16K。

步驟五：Sender 線(xiàn)程啟動(dòng)以后會(huì)從緩存里面去獲取可以發(fā)送的批次。

步驟六：Sender 線(xiàn)程把一個(gè)一個(gè)批次發(fā)送到服務(wù)端。大家要注意這個(gè)設(shè)計(jì)，在 Kafka0.8 版本以前，Kafka 生產(chǎn)者的設(shè)計(jì)是來(lái)一條數(shù)據(jù)，就往服務(wù)端發(fā)送一條數(shù)據(jù)，頻繁的發(fā)生網(wǎng)絡(luò)請(qǐng)求，結(jié)果性能很差。后面的版本再次架構(gòu)演進(jìn)的時(shí)候把這兒改成了批處理的方式，性能指數(shù)級(jí)的提升，這個(gè)設(shè)計(jì)值得我們積累。 

生產(chǎn)者細(xì)節(jié)深度剖析 

接下來(lái)我們生產(chǎn)者這兒技術(shù)含量比較高的一個(gè)地方，前面概述那兒我們看到，一個(gè)消息被分區(qū)以后，消息就會(huì)被放到一個(gè)緩存里面，我們看一下里面具體的細(xì)節(jié)。默認(rèn)緩存塊的大小是 32M，這個(gè)緩存塊里面有一個(gè)重要的數(shù)據(jù)結(jié)構(gòu)：batches，這個(gè)數(shù)據(jù)結(jié)構(gòu)是 key-value 的結(jié)果，key 就是消息主題的分區(qū)，value 是一個(gè)隊(duì)列，里面存的是發(fā)送到對(duì)應(yīng)分區(qū)的批次，Sender 線(xiàn)程就是把這些批次發(fā)送到服務(wù)端。

圖3 生產(chǎn)者架構(gòu)

01 /  生產(chǎn)者高級(jí)設(shè)計(jì)之自定義數(shù)據(jù)結(jié)構(gòu)

生產(chǎn)者把批次信息用 batches 這個(gè)對(duì)象進(jìn)行存儲(chǔ)。如果是大家，大家會(huì)考慮用什么數(shù)據(jù)結(jié)構(gòu)去存儲(chǔ)批次信息？

Kafka 這兒采取的方式是自定義了一個(gè)數(shù)據(jù)結(jié)構(gòu)：CopyOnWriteMap。熟悉 Java 的同學(xué)都知道，JUC 下面是有一個(gè) CopyOnWriteArrayList 的數(shù)據(jù)結(jié)構(gòu)的，但是沒(méi)有 CopyOnWriteMap，我這兒給大家解釋一下 Kafka 為什么要設(shè)計(jì)這樣的一個(gè)數(shù)據(jù)結(jié)構(gòu)。

1.他們存儲(chǔ)的信息的是 key-value 的結(jié)構(gòu)，key 是分區(qū)，value 是要存到這個(gè)分區(qū)的對(duì)應(yīng)批次（批次可能有多個(gè)，所以用的是隊(duì)列），故因?yàn)槭?key-value 的數(shù)據(jù)結(jié)構(gòu)，所以鎖定用 Map 數(shù)據(jù)結(jié)構(gòu)。

2.這個(gè) Kafka 生產(chǎn)者面臨的是一個(gè)高并發(fā)的場(chǎng)景，大量的消息會(huì)涌入這個(gè)這個(gè)數(shù)據(jù)結(jié)構(gòu)，所以這個(gè)數(shù)據(jù)結(jié)構(gòu)需要保證線(xiàn)程安全，這樣我們就不能使用 HashMap 這樣的數(shù)據(jù)結(jié)構(gòu)了。

3.這個(gè)數(shù)據(jù)結(jié)構(gòu)需要支持的是讀多寫(xiě)少的場(chǎng)景。讀多是因?yàn)槊織l消息過(guò)來(lái)都會(huì)根據(jù) key 讀取 value 的信息，假如有 1000 萬(wàn)條消息，那么就會(huì)讀取 batches 對(duì)象 1000 萬(wàn)次。寫(xiě)少是因?yàn)椋热缥覀兩a(chǎn)者發(fā)送數(shù)據(jù)需要往一個(gè)主題里面去發(fā)送數(shù)據(jù)，假設(shè)這個(gè)主題有 50 個(gè)分區(qū)，那么這個(gè) batches 里面就需要寫(xiě) 50 個(gè) key-value 數(shù)據(jù)就可以了（大家要搞清楚我們雖然要寫(xiě) 1000 萬(wàn)條數(shù)據(jù)，但是這 1000 萬(wàn)條是寫(xiě)入 queue 隊(duì)列的 batch 里的，并不是直接寫(xiě)入 batches,所以就我們剛剛說(shuō)的這個(gè)場(chǎng)景，batches 里只需要最多寫(xiě) 50 條數(shù)據(jù)就可以了）。

根據(jù)第二和第三個(gè)場(chǎng)景我們總結(jié)出來(lái)，Kafka 這兒需要一個(gè)能保證線(xiàn)程安全的，支持讀多寫(xiě)少的 Map 數(shù)據(jù)結(jié)構(gòu)。但是 Java 里面并沒(méi)有提供出來(lái)的這樣的一個(gè)數(shù)據(jù)，唯一跟這個(gè)需求比較接近的是 CopyOnWriteArrayList，但是偏偏它又不是 Map 結(jié)構(gòu)，所以 Kafka 這兒模仿 CopyOnWriteArrayList 設(shè)計(jì)了 CopyOnWriteMap。采用了讀寫(xiě)分離的思想解決了線(xiàn)程安全且支持讀多寫(xiě)少等問(wèn)題。

高效的數(shù)據(jù)結(jié)構(gòu)保證了生產(chǎn)者的性能。（CopyOnWriteArrayList 不熟悉的同學(xué)，可以嘗試百度學(xué)習(xí)）。這兒筆者建議大家可以去看看 Kafka 生產(chǎn)者往 batches 里插入數(shù)據(jù)的源碼，生產(chǎn)者為了保證插入數(shù)據(jù)的高性能，采用了多線(xiàn)程，又為了線(xiàn)程安全，使用了分段加鎖等多種手段，源碼非常精彩。

02 /  生產(chǎn)者高級(jí)設(shè)計(jì)之內(nèi)存池設(shè)計(jì)

剛剛我們看到 batches 里面存儲(chǔ)的是批次，批次默認(rèn)的大小是 16K，整個(gè)緩存的大小是 32M，生產(chǎn)者每封裝一個(gè)批次都需要去申請(qǐng)內(nèi)存，正常情況下如果一個(gè)批次發(fā)送出去了以后，那么這 16K 的內(nèi)存就等著 GC 來(lái)回收了。但是如果是這樣的話(huà)，就可能會(huì)頻繁的引發(fā) FullGC，故而影響生產(chǎn)者的性能，所以在緩存里面設(shè)計(jì)了一個(gè)內(nèi)存池（類(lèi)似于我們平時(shí)用的數(shù)據(jù)庫(kù)的連接池），一個(gè) 16K 的內(nèi)存用完了以后，把數(shù)據(jù)清空，放入到內(nèi)存池里，下個(gè)批次用的時(shí)候直接從里面獲取就可以。這樣大大的減少了 GC 的頻率，保證了生產(chǎn)者的穩(wěn)定和高效（Java 的 GC 問(wèn)題是一個(gè)頭疼的問(wèn)題，所以這種設(shè)計(jì)也非常值得我們?nèi)シe累）。

以上就是Kafka的生產(chǎn)者優(yōu)秀架構(gòu)設(shè)計(jì)的示例分析，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見(jiàn)到或用到的。希望你能通過(guò)這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注億速云行業(yè)資訊頻道。