[[432602]]

本文轉(zhuǎn)載自微信公眾號「華仔聊技術(shù)」，作者王江華。轉(zhuǎn)載本文請聯(lián)系華仔聊技術(shù)公眾號。

 1kafka三高架構(gòu)概述

由于最近事情比較多,工作也比較忙, 這篇差點(diǎn)難產(chǎn),經(jīng)過幾個(gè)周末的構(gòu)思和梳理,終于跟大家見面了, 在上一篇我們講述了 kafka 的基礎(chǔ)入門, 工作流程, 存儲機(jī)制,副本等知識, 本篇會為大家揭秘 kafka 高可用,高性能,高并發(fā)架構(gòu)設(shè)計(jì)奧秘。

Kafka 向來以高吞吐量，低延遲，高并發(fā), 高可擴(kuò)展性而自稱，并在越來越多的場景中應(yīng)用，這時(shí)候就對其穩(wěn)定性的要求就越高。接下來就為大家一一呈現(xiàn)里面的細(xì)節(jié)。

2kafka高可用設(shè)計(jì)

Leader選舉機(jī)制

Kafka 中的選舉大致分為三大類: 控制器的選舉, Leader 的選舉, 消費(fèi)者的選舉。在講解 Leader 選舉之前, 先說說 Kafka 控制器, 即 Broker。它除了具有一般 Broker 的功能外, 還具有選舉分區(qū)Leader節(jié)點(diǎn)的功能, 在啟動 Kafka 系統(tǒng)時(shí)候, 其中一個(gè) Broker 會被選舉為控制器, 負(fù)責(zé)管理主題分區(qū)和副本的狀態(tài), 還會執(zhí)行重分配的任務(wù)。

控制器的啟動順序如下:

1) 第一個(gè)啟動的節(jié)點(diǎn)，會在 Zookeeper 系統(tǒng)里面創(chuàng)建一個(gè)臨時(shí)節(jié)點(diǎn) /controller ，并寫入該節(jié)點(diǎn)的注冊信息，使該節(jié)點(diǎn)成為控制器。

2) 其他的節(jié)點(diǎn)在陸續(xù)啟動時(shí)，也會嘗試在 Zookeeper 系統(tǒng)中創(chuàng)建 /controller 節(jié)點(diǎn)，但是 /controller 節(jié)點(diǎn)已經(jīng)存在，所以會拋出 “創(chuàng)建/controller節(jié)點(diǎn)失敗異常” 的信息。創(chuàng)建失敗的節(jié)點(diǎn)會根據(jù)返回的結(jié)果，判斷出在 Kafka 集群中已經(jīng)有一個(gè)控制器被成功創(chuàng)建了，所以放棄創(chuàng)建 /controller 節(jié)點(diǎn)，這樣就確保了 Kafka 集群控制器的唯一性。

3) 其他的節(jié)點(diǎn)，也會在控制器上注冊相應(yīng)的監(jiān)聽器，各個(gè)監(jiān)聽器負(fù)責(zé)監(jiān)聽各自代理節(jié)點(diǎn)的狀態(tài)變化。當(dāng)監(jiān)聽到節(jié)點(diǎn)狀態(tài)發(fā)生變化時(shí)，會觸發(fā)相應(yīng)的監(jiān)聽函數(shù)進(jìn)行處理。

說完控制器方面的知識, 我們來講解Leader節(jié)點(diǎn)的選舉過程, 選舉控制器的核心思路是：各個(gè)節(jié)點(diǎn)公平競爭搶占 Zookeeper 系統(tǒng)中創(chuàng)建 /controller臨時(shí)節(jié)點(diǎn)，最先創(chuàng)建成功的節(jié)點(diǎn)會成為控制器，并擁有選舉主題分區(qū)Leader節(jié)點(diǎn)的功能。選舉流程如下圖所示:

副本機(jī)制

副本機(jī)制簡單來說就是備份機(jī)制，就是在分布式集群中保存著相同的數(shù)據(jù)備份。那么副本機(jī)制的好處就是提供數(shù)據(jù)冗余, 副本機(jī)制是kafka確保系統(tǒng)高可用和高持久的重要基石。

為了保證高可用，kafka 的分區(qū)是多副本的，如果其中一個(gè)副本丟失了，那么還可以從其他副本中獲取分區(qū)數(shù)據(jù)(要求對應(yīng)副本的數(shù)據(jù)必須是完整的)。這是 Kafka 數(shù)據(jù)一致性的基礎(chǔ), 下面將詳解介紹 Kafka 的副本機(jī)制。

Kafka 使用 Zookeeper 來維護(hù)集群 Brokers 的信息，每個(gè) Broker 都有一個(gè)唯一的標(biāo)識broker.id，用于標(biāo)識自己在集群中的身份。Brokers 會通過 Zookeeper 選舉出一個(gè)叫Controller Broker節(jié)點(diǎn)，它除了具備其它Brokers的功能外，還負(fù)責(zé)管理主題分區(qū)及其副本的狀態(tài)。

在 Kafka 中 Topic 被分為多個(gè)分區(qū)(Partition)，分區(qū)是 Kafka 最基本的存儲單位。在創(chuàng)建主題的時(shí)候可使用replication-factor參數(shù)指定分區(qū)的副本個(gè)數(shù)。分區(qū)副本總會有一個(gè) Leader 副本，所有的消息都直接發(fā)送給Leader 副本，其它副本都需要通過復(fù)制 Leader 中的數(shù)據(jù)來保證數(shù)據(jù)一致。當(dāng) Leader 副本不可用時(shí)，其中一個(gè) Follower 將會被選舉并成為新的 Leader。

ISR機(jī)制

 認(rèn)識 ISR

如上圖所示, 每個(gè)分區(qū)都有一個(gè) ISR(in-sync Replica) 列表，用于維護(hù)所有同步的、可用的副本。Leader 副本必然是同步副本，也就是說, ISR 不只是追隨者副本集合, 它比如包括 Leader 副本。甚至在某些情況下, ISR 只有Leader 這一個(gè)副本, 而對于 Follower 副本來說，它需要滿足以下條件才能被認(rèn)為是同步副本：

1) 必須定時(shí)向 Zookeeper 發(fā)送心跳;

2) 在規(guī)定的時(shí)間內(nèi)從 Leader 副本 "低延遲" 地獲取過消息。

如果副本不滿足上面條件的話，就會被從 ISR 列表中移除，直到滿足條件才會被再次加入。所以就可能會存在 Follower 不可能與 Leader 實(shí)時(shí)同步的風(fēng)險(xiǎn)。

Kafka 判斷 Follower 是否與 Leader 同步的條件就是 Broker 端參數(shù) replica.lag.time.max.ms 參數(shù)值。這個(gè)參數(shù)的含義就是 Follower 副本能夠落后 Leader 副本的最長時(shí)間間隔, 當(dāng)前默認(rèn)值為10秒, 也就是說, 只要一個(gè)Follower 副本落后 Leader 副本的時(shí)間不連續(xù)超過10秒, Kafka 就認(rèn)為兩者是同步的, 即使 Follower 副本中保持的消息要少于 Leader 副本中的消息。

Kafka中ISR的管理最終都會反饋到 Zookeeper節(jié)點(diǎn)上。具體位置為：/brokers/topics/[topic]/partitions/[partition]/state。目前有兩個(gè)地方會對這個(gè)Zookeeper的節(jié)點(diǎn)進(jìn)行維護(hù)：

1) Controller來維護(hù)：Kafka 集群中的其中一個(gè) Broker 會被選舉為Controller，主要負(fù)責(zé) Partition 管理和副本狀態(tài)管理，也會執(zhí)行重分配 Partition 之類的管理任務(wù)。在符合某些特定條件下，Controller 下的 LeaderSelector 會選舉新的 Leader，ISR 和新的 leader_epoch 及controller_epoch 寫入 Zookeeper 的相關(guān)節(jié)點(diǎn)中。同時(shí)發(fā)起 leaderAndIsrRequest 通知所有的 Replicas。

2) Leader來維護(hù)：Leader 有單獨(dú)的線程定期檢測 ISR 中 Follower 是否脫離 ISR , 如果發(fā)現(xiàn) ISR 變化，則會將新的 ISR 信息返回到 Zookeeper 的相關(guān)節(jié)點(diǎn)中。

ACK機(jī)制

這個(gè)acks參數(shù)在kafka的使用中，是非常核心以及關(guān)鍵的一個(gè)參數(shù)，決定了很多東西, 這個(gè)acks跟副本機(jī)制,同步機(jī)制,ISR機(jī)制都密切相關(guān), 如果無法理解這些,是無法充分理解acks參數(shù)的含義。

首先這個(gè)acks參數(shù)，是在KafkaProducer，也就是生產(chǎn)者客戶端里設(shè)置的。那么也就是說，你往kafka寫數(shù)據(jù)的時(shí)候，就可以來設(shè)置這個(gè)acks參數(shù)。這個(gè)參數(shù)實(shí)際上有三種常見的值可以設(shè)置，分別是：0、1 和 all。

acks = 0

如果acks設(shè)置為0，那么 Producer 是不會等待 Broker 的反饋。該消息會被立刻添加到 Socket Buffer 中就認(rèn)為已經(jīng)發(fā)送完成。在這種情況下，服務(wù)器端是否收到請求是無法保證的，并且參數(shù) Retries 也不會生效(因?yàn)榭蛻舳藷o法獲得失敗信息)。

這個(gè)時(shí)候每個(gè)記錄返回的 Offset 總是被設(shè)置為-1。這個(gè)模式下 Kafka 的吞吐量最大，并發(fā)最高，但是數(shù)據(jù)非常容易丟失，通常適用在一些記錄應(yīng)用日志，對數(shù)據(jù)要求不高的業(yè)務(wù)場景。

acks = 1

如果acks設(shè)置為1，這個(gè)時(shí)候 Leader 節(jié)點(diǎn)會將記錄先寫入本地日志，并且在所有 Follower 節(jié)點(diǎn)反饋之前就先確認(rèn)成功。在這種情況下，如果 Leader 節(jié)點(diǎn)在接收記錄之后，并且在 Follower 節(jié)點(diǎn)復(fù)制數(shù)據(jù)完成之前發(fā)生錯(cuò)誤，那么這條記錄會丟失。這個(gè)模式和 Mysql 的主從異步復(fù)制一樣，主從之間會有數(shù)據(jù)差異，此配置為 Kafka 默認(rèn)配置。它平衡了數(shù)據(jù)安全和性能。

acks = all & min.insync.replicas >= 2

如果acks設(shè)置為all，這個(gè)時(shí)候 Leader 節(jié)點(diǎn)會等待所有同步中的LSR副本確認(rèn)之后再確認(rèn)這條記錄是否發(fā)送完成。只要至少有一個(gè)同步副本存在，記錄就不會丟失。

如果說 Leader 這時(shí)候剛接收到了消息，但是 Follower 沒有收到消息，此時(shí) Leader 宕機(jī)了，那么客戶端會感知到這個(gè)消息沒發(fā)送成功，他會重試再次發(fā)送消息過去。

其中Broker有個(gè)配置項(xiàng)min.insync.replicas(默認(rèn)值為1)代表了正常寫入生產(chǎn)者數(shù)據(jù)所需要的最少ISR個(gè)數(shù), 當(dāng)ISR中的副本數(shù)量小于min.insync.replicas時(shí)，Leader停止寫入生產(chǎn)者生產(chǎn)的消息，并向生產(chǎn)者拋出NotEnoughReplicas異常，阻塞等待更多的 Follower 趕上并重新進(jìn)入ISR, 因此能夠容忍min.insync.replicas-1個(gè)副本同時(shí)宕機(jī)

這種方式是犧牲了性能為代價(jià)，適合對數(shù)據(jù)要求比較高的業(yè)務(wù)場景。

3kafka高性能設(shè)計(jì)

Reactor多路復(fù)用模型

提到 Reactor (多路復(fù)用), 就不得不提 Java 中的 NIO, 接下來 我們先來看下 Java 的 NIO。

Java NIO由以下幾個(gè)核心部分組成 :

1) Channels;

2) Buffers;

3) Selectors;

Channel 和 Java 中的 Stream 一樣, 用于傳輸數(shù)據(jù)的數(shù)據(jù)流, 數(shù)據(jù)可以Channel 讀取到Buffer 中, 也可以從 Buffer 寫到 Channel 中, 如下圖所示:

Selector 允許單線程處理多個(gè) Channel。使用 Selector，首先得向Selector 注冊 Channel，然后調(diào)用它的 select() 方法。此方法會一直阻塞到某個(gè)注冊的 Channel 有事件就緒。一旦這個(gè)方法返回，線程就可以處理這些事件，事件的例子如新連接進(jìn)來，數(shù)據(jù)接收等

下圖為一個(gè)單線程中使用一個(gè)Selector處理3個(gè)Channel：

Kafka SocketServer 是基于Java NIO 開發(fā)的，采用了 Reactor 的模式(已被大量實(shí)踐證明非常高效，在 Netty 和 Mina 中廣泛使用)。Kafka Reactor 的模式包含三種角色：

1) Acceptor;

2) Processor;

3) Handler;

Kafka Reacator 包含了1個(gè) Acceptor 負(fù)責(zé)接受客戶端請求，N個(gè)Processor 線程負(fù)責(zé)讀寫數(shù)據(jù)(即為每個(gè) Connection 創(chuàng)建出一個(gè) Processor 去單獨(dú)處理,每個(gè)Processor中均引用獨(dú)立的Selector)，M個(gè)Handler來處理業(yè)務(wù)邏輯。在Acceptor和Processor，Processor和Handler之間都有隊(duì)列來緩沖請求。

如下圖所示是kafka 簡版的 Reactor模型架構(gòu)圖

生產(chǎn)消息流程

生產(chǎn)者發(fā)送到 Kafka集群的詳細(xì)流程如下圖所示:

1) 首先來一條消息后,生產(chǎn)者源碼里面會對消息進(jìn)行封裝成 ProducerRecord對象。

2) 封裝成對象后會對該對象進(jìn)行序列化[涉及網(wǎng)絡(luò)傳輸], 調(diào)用Serializer組件進(jìn)行序列化, 序列化后進(jìn)行發(fā)送。

3) 在發(fā)送前要確定一件事, 到底要把這條消息發(fā)送到哪個(gè)主題的哪個(gè)分區(qū), 這個(gè)時(shí)候就需要通過 Partitioner 分區(qū)器 從 Kafka Broker集群中獲取集群元數(shù)據(jù), 獲取到元數(shù)據(jù)后就可以進(jìn)行發(fā)送了。

4) 在0.8版本之前, 這個(gè)時(shí)候來了一條消息就會封裝成一個(gè)請求發(fā)送到Broker, 這種情況下, 性能是非常差的, 在0.8版本之后, 進(jìn)行簡單的改進(jìn), 性能得到了指數(shù)級上升, 即來了一條消息后不會立馬發(fā)送出去, 而是先寫入到一個(gè)緩存(RecordAccumulator)隊(duì)列中,封裝成一個(gè)個(gè)批次(RecordBatch)。

5) 這個(gè)時(shí)候會有一個(gè)sender線程會將多個(gè)批次封裝成一個(gè)請求(Request), 然后進(jìn)行發(fā)送, 這樣會減少很多請求,提高吞吐量。這個(gè)時(shí)候有個(gè)問題, 一條消息過來后沒有立即發(fā)送出去,而是封裝成了批次, 這樣會不會有延遲的問題, 默認(rèn)的batch.size是16K, 寫滿會立即發(fā)送, 如果寫不滿, 也會在規(guī)定的時(shí)間進(jìn)行發(fā)送(linger.ms = 500ms)

6) 發(fā)送的時(shí)候 每個(gè)Request請求對應(yīng)多路復(fù)用器(Selector)中的每個(gè)kafka channel 然后將數(shù)據(jù)發(fā)送給Broker集群

7) 在封裝Batch批次和Request請求的過程中, 還涉及一個(gè)重要的設(shè)計(jì)理念即內(nèi)存池方案, 在后面的服務(wù)端內(nèi)存池部分進(jìn)行詳細(xì)說明

順序?qū)懘疟P + OS Cache

首先 Kafka 為了保證磁盤寫入性能，通過基于操作系統(tǒng)的頁緩存來實(shí)現(xiàn)文件寫入的。操作系統(tǒng)本身有一層緩存，叫做 page cache，是在內(nèi)存里的緩存，我們也可以稱之為 os cache，意思就是操作系統(tǒng)自己管理的緩存。那么在寫磁盤文件的時(shí)候，就可以先直接寫入 os cache 中，也就是僅僅寫入內(nèi)存中，接下來由操作系統(tǒng)自己決定什么時(shí)候把 os cache 里的數(shù)據(jù)真的刷入到磁盤中, 這樣大大提高寫入效率和性能。 如下圖所示:

另外還有一個(gè)非常關(guān)鍵的操作,就是 kafka 在寫數(shù)據(jù)的時(shí)候是以磁盤順序?qū)懙姆绞絹磉M(jìn)行落盤的, 即將數(shù)據(jù)追加到文件的末尾, 而不是在文件的隨機(jī)位置來修改數(shù)據(jù), 對于普通機(jī)械磁盤, 如果是隨機(jī)寫的話, 涉及到磁盤尋址的問題,導(dǎo)致性能確實(shí)極低, 但是如果只是按照順序的方式追加文件末尾的話, 這種磁盤順序?qū)懙男阅芑究梢愿鷮憙?nèi)存的性能差不多的。

零拷貝技術(shù)(zero-copy)

上面說完了寫入的過程, 我們來講講消費(fèi)的這塊流程, 從 Kafka 消費(fèi)數(shù)據(jù), 在消費(fèi)的時(shí)候?qū)嶋H上就是從 Kafka 的磁盤文件讀取數(shù)據(jù)然后發(fā)送給下游的消費(fèi)者。大概過程如下:

1) 先檢查要讀取的數(shù)據(jù)是否在 os cache 中, 如果不在的話就從磁盤文件讀取數(shù)據(jù)后放入 os cache。

2) 接著從 os cache 里面 copy 數(shù)據(jù)到應(yīng)用程序進(jìn)程的緩存里面, 在從應(yīng)用程序進(jìn)程的緩存里 copy 數(shù)據(jù)到操作系統(tǒng)層面的 socket緩存里面, 最后再從 socket 緩存里面讀取數(shù)據(jù)后發(fā)送到網(wǎng)卡, 最后從網(wǎng)卡發(fā)送到下游的消費(fèi)者。

從上圖可以看出, 整個(gè)過程有兩次沒必要的拷貝操作

1) 從操作系統(tǒng)的 os cache 拷貝數(shù)據(jù)到應(yīng)用程序進(jìn)程的緩存。

2) 接著又從應(yīng)用程序緩存里拷貝到操作系統(tǒng)的socket緩存中。

這兩次拷貝過程中, 還發(fā)生了好幾次上下文的切換, 所以相對來說是比較消耗性能的

kafka 為了解決這個(gè)問題, 在讀取數(shù)據(jù)的時(shí)候就引入了零拷貝技術(shù)。即讓操作系統(tǒng)的 os cache 中的數(shù)據(jù)直接發(fā)送到網(wǎng)卡后傳出給下游的消費(fèi)者，中間跳過了兩次拷貝數(shù)據(jù)的步驟，從而減少拷貝的 CPU 開銷, 減少用戶態(tài)內(nèi)核態(tài)的上下文切換次數(shù), 從而優(yōu)化數(shù)據(jù)傳輸?shù)男阅? 而Socket緩存中僅僅會拷貝一個(gè)描述符過去，不會拷貝數(shù)據(jù)到Socket緩存。如下圖所示:

常見的零拷貝思路主要有兩種實(shí)現(xiàn)方式:

1) 直接I/O: 數(shù)據(jù)直接跳過內(nèi)核, 在用戶空間與 I/O 設(shè)備之間進(jìn)行傳遞, 內(nèi)核在這種情況下只是進(jìn)行必要的輔助工作

2) copy-on-write: 寫時(shí)復(fù)制, 數(shù)據(jù)不需要提前進(jìn)行拷貝, 而是在當(dāng)需要修改的時(shí)候再進(jìn)行部分?jǐn)?shù)據(jù)的拷貝

這里, Kafka 主要使用到了 mmap 和 sendfile 的方式來實(shí)現(xiàn)零拷貝, 對應(yīng)java里面的 MappedByteBuffer 和 FileChannel.transferIO。

使用 java NIO 實(shí)現(xiàn)的 零拷貝, 如下:

transferTo() 方法會將數(shù)據(jù)從文件通道傳輸?shù)搅私o定的可寫字節(jié)通道。在其內(nèi)部它依賴底層操作系統(tǒng)對零拷貝的支持;在 Linux 系統(tǒng)中，此調(diào)用被傳遞到 sendfile() 系統(tǒng)調(diào)用中，調(diào)用過程如下:

壓縮傳輸

默認(rèn)情況下, 在 Kafka 生產(chǎn)者中不啟用壓縮 Compression 不僅可以更快地從生產(chǎn)者傳輸?shù)酱? 還可以在復(fù)制過程中進(jìn)行更快的傳輸。壓縮有助于提高吞吐量, 降低延遲并提高磁盤利用率。

在 Kafka 中, 壓縮可能會發(fā)生在兩個(gè)地方: 生產(chǎn)者端和Broker端, 一句話總結(jié)下壓縮和解壓縮, 即 Producer 端壓縮, Broker 端保持, Consumer 端解壓縮。

Kafka 支持多種壓縮算法: lz4, snappy, gzip, 從Kafka 2.1.0 開始新增了 ZStandard 算法, 該算法是 Facebook 開源的壓縮算法, 能提供超高的壓縮比。

Producer、Broker、Consumer 要使用相同的壓縮算法, 在 Producer 向 Broker 寫入數(shù)據(jù), Consumer 向 Broker 讀取數(shù)據(jù)的時(shí)候可以不用解壓縮, 只需要在最終 Consumer 到消息的時(shí)候才進(jìn)行解壓縮, 這樣可以節(jié)省大量的網(wǎng)絡(luò)和磁盤開銷。

服務(wù)端內(nèi)存池設(shè)計(jì)

在前面我們講解了一條消息生產(chǎn)的詳細(xì)流程, 中間涉及到了批次(Batch)和請求(Request), 在這個(gè)過程中, Kafka還有一個(gè)重要的設(shè)計(jì)理念 即內(nèi)存池方案, 這里就詳細(xì)講述下內(nèi)存池的實(shí)現(xiàn)過程.

1) 這里簡化下流程, 來一條消息會先進(jìn)行封裝然后序列化最后會計(jì)算出分區(qū)號, 并把這個(gè)消息存儲到緩存里面

2) 這個(gè)緩存里面也是有設(shè)計(jì)的 即批次隊(duì)列, 那么這個(gè)批次隊(duì)列是使用什么策略存儲呢? 一個(gè)分區(qū)對應(yīng)一個(gè)隊(duì)列, 這里有個(gè)重要的數(shù)據(jù)結(jié)構(gòu):Batches, 這個(gè)數(shù)據(jù)結(jié)構(gòu)是Key-value形式, key是消息主題的分區(qū), value是一個(gè)隊(duì)列, 里面存儲的發(fā)送到對應(yīng)分區(qū)的批次

3) 那么假設(shè)這個(gè)時(shí)候 我們有個(gè)2個(gè)topic, 每個(gè)topic有2個(gè)分區(qū), 那么是不是總共有4個(gè)的分區(qū)即4個(gè)隊(duì)列, 每個(gè)隊(duì)列里面都有一個(gè)個(gè)批次, 這個(gè)時(shí)候消息算出來分區(qū)后就會寫入隊(duì)列的最新一個(gè)批次

4) Sender線程就會檢測這個(gè)批次(Batch)是否已經(jīng)寫滿,或者時(shí)間是否到達(dá), 如果滿足Sender線程就會取出封裝成Request就會發(fā)送

5) 封裝批次會用到內(nèi)存, Sender發(fā)送完畢內(nèi)存會進(jìn)行回收, 在Java中如果頻繁操作內(nèi)存和回收,會遇到頭疼的FullGC的問題, 工作線程的性能就會降低, 整個(gè)生產(chǎn)者的性能就會受到影響, Kafka的解決方案就是內(nèi)存池, 對內(nèi)存塊的使用跟數(shù)據(jù)庫的連接池一樣

6) 整個(gè)Buffer Poll 內(nèi)存池大小是32M , 內(nèi)存池分為兩個(gè)部分, 一個(gè)部分是內(nèi)存隊(duì)列, 隊(duì)列里面有一個(gè)個(gè)內(nèi)存塊(16K), 另外一部分是可用內(nèi)存, 一條消息過來后會向內(nèi)存池申請內(nèi)存塊, 申請完后封裝批次并寫入數(shù)據(jù), sender線程就會發(fā)送并響應(yīng), 然后清空內(nèi)存放回內(nèi)存池里面進(jìn)行反復(fù)使用, 這樣就大大減少了GC的頻率, 保證了生產(chǎn)者的穩(wěn)定和高效, 性能會大大提高

4kafka高并發(fā)設(shè)計(jì)

高并發(fā)網(wǎng)絡(luò)設(shè)計(jì)

上面通過大量的篇幅講解了kafka生產(chǎn)者和服務(wù)端的高可用和高性能的方方面面, 這里主要來分析下 Kafka的超高并發(fā)網(wǎng)絡(luò)架構(gòu)設(shè)計(jì), 此架構(gòu)設(shè)計(jì)是 Kafka中最經(jīng)典的。

這里我們將 Kafka 的網(wǎng)絡(luò)架構(gòu)抽象成如上圖所示的三層架構(gòu), 整個(gè)請求流轉(zhuǎn)的路徑如下:

1) 客戶端發(fā)送請求過來, 在Kafka 服務(wù)端會有個(gè)Acceptor線程, 這個(gè)線程上面綁定了OP_ACCEPT事件, 用來監(jiān)聽發(fā)送過來的請求, 下面有個(gè)while死循環(huán)會源源不斷的監(jiān)聽Selector是否有請求發(fā)送過來, 接收到請求鏈接后封裝成socketchannel, 然后將socketChannel發(fā)送給網(wǎng)絡(luò)第一層架構(gòu)中。

2) 在第一層架構(gòu)中有3個(gè)一模一樣的Processor線程, 這個(gè)線程的里面都有一個(gè)連接隊(duì)列,里面存放socketchannel, 存放規(guī)則為輪詢存放, 隨著請求的不斷增加, 連接隊(duì)列里面就會有很多個(gè)socketchannel, 這個(gè)時(shí)候socketchannel就會在每個(gè)selector上面注冊O(shè)P_READ事件, 參考上圖第一層的第三個(gè)Processor線程, 即每個(gè)線程里面還有一個(gè)while循環(huán)會遍歷每個(gè)socketchannel, 監(jiān)聽到事件后就會接收到客戶端發(fā)送過來的請求, 這個(gè)時(shí)候Processor線程會對請求進(jìn)行解析(發(fā)送過來的請求是二進(jìn)制的, 上面已經(jīng)說過, 跨網(wǎng)絡(luò)傳輸需要進(jìn)行序列化) , 并解析封裝成Request對象發(fā)送到上圖所示的網(wǎng)絡(luò)第二層架構(gòu)中。

3) 在第二層架構(gòu)中會有兩個(gè)隊(duì)列, 一個(gè)RequestQueue(請求隊(duì)列), 一個(gè)是ResponseQueue(返回隊(duì)列), 在請求隊(duì)列中會存放一個(gè)個(gè)Request請求, 起到緩沖的作用, 這個(gè)時(shí)候就到了網(wǎng)絡(luò)第三層架構(gòu)中。

4) 在第三層架構(gòu)中有個(gè)RequestHandler線程池, 里面默認(rèn)有8個(gè)RequestHandler線程, 這8個(gè)線程啟動后會不斷的從第二層的RequestQueue隊(duì)列中獲取請求, 解析請求體里面的數(shù)據(jù), 通過內(nèi)置工具類將數(shù)據(jù)寫入到磁盤

5) 寫入成功后還要響應(yīng)客戶端, 這個(gè)時(shí)候會封裝一個(gè)Response對象, 會將返回結(jié)果存放到第二層的ResponseQueue隊(duì)列中, 此時(shí)默認(rèn)有3個(gè)小的Response隊(duì)列, 這里面的個(gè)數(shù)是同第一層架構(gòu)中的Processor線程一一對應(yīng)的。

6) 這個(gè)時(shí)候第一層的Processor線程中while循環(huán)就會遍歷Response請求, 遍歷完成后就會在selector上注冊O(shè)P_WRITE事件, 這個(gè)時(shí)候就會將響應(yīng)請求發(fā)送回客戶端。

7) 在整個(gè)過程中涉及到2個(gè)參數(shù):num.network.threads = 3 和 num.io.threads = 8 如果感覺默認(rèn)參數(shù)性能不夠好的話, 可以對這2個(gè)參數(shù)進(jìn)行優(yōu)化, 比如將num.network.threads = 9, num.io.threads = 32(和CPU個(gè)數(shù)要一致), 每個(gè)RequestHandler線程可以處理2000QPS, 2000 * 8 = 1.6萬QPS , 擴(kuò)容后可以支撐6.4萬QPS, 通過擴(kuò)容后Kafka可以支撐6萬QPS, 可以看出通過上面的架構(gòu)講解, kafka是可以支撐高并發(fā)的請求的

5總結(jié)

至此已經(jīng)跟大家全面揭秘了 Kafka 三高架構(gòu)的方方面面, 下一篇會講解 Kafka生產(chǎn)級部署和容量規(guī)劃方面的知識, 大家敬請期待......