[[432602]]

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

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

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

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

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

Leader選舉機(jī)制

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

控制器的啟動(dòng)順序如下:

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

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

3) 其他的節(jié)點(diǎn)，也會(huì)在控制器上注冊(cè)相應(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í)，會(huì)觸發(fā)相應(yīng)的監(jiān)聽函數(shù)進(jìn)行處理。

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

副本機(jī)制

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

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

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

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

ISR機(jī)制

 認(rèn)識(shí) ISR

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

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

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

如果副本不滿足上面條件的話，就會(huì)被從 ISR 列表中移除，直到滿足條件才會(huì)被再次加入。所以就可能會(huì)存在 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的管理最終都會(huì)反饋到 Zookeeper節(jié)點(diǎn)上。具體位置為：/brokers/topics/[topic]/partitions/[partition]/state。目前有兩個(gè)地方會(huì)對(duì)這個(gè)Zookeeper的節(jié)點(diǎn)進(jìn)行維護(hù)：

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

2) Leader來維護(hù)：Leader 有單獨(dú)的線程定期檢測(cè) ISR 中 Follower 是否脫離 ISR , 如果發(fā)現(xiàn) ISR 變化，則會(huì)將新的 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 是不會(huì)等待 Broker 的反饋。該消息會(huì)被立刻添加到 Socket Buffer 中就認(rèn)為已經(jīng)發(fā)送完成。在這種情況下，服務(wù)器端是否收到請(qǐng)求是無法保證的，并且參數(shù) Retries 也不會(huì)生效(因?yàn)榭蛻舳藷o法獲得失敗信息)。

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

acks = 1

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

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

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

如果說 Leader 這時(shí)候剛接收到了消息，但是 Follower 沒有收到消息，此時(shí) Leader 宕機(jī)了，那么客戶端會(huì)感知到這個(gè)消息沒發(fā)送成功，他會(huì)重試再次發(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à)，適合對(duì)數(shù)據(jù)要求比較高的業(yè)務(wù)場(chǎng)景。

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 注冊(cè) Channel，然后調(diào)用它的 select() 方法。此方法會(huì)一直阻塞到某個(gè)注冊(cè)的 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é)接受客戶端請(qǐng)求，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ì)列來緩沖請(qǐng)求。

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

生產(chǎn)消息流程

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

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

2) 封裝成對(duì)象后會(huì)對(duì)該對(duì)象進(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í)候來了一條消息就會(huì)封裝成一個(gè)請(qǐng)求發(fā)送到Broker, 這種情況下, 性能是非常差的, 在0.8版本之后, 進(jìn)行簡(jiǎn)單的改進(jìn), 性能得到了指數(shù)級(jí)上升, 即來了一條消息后不會(huì)立馬發(fā)送出去, 而是先寫入到一個(gè)緩存(RecordAccumulator)隊(duì)列中,封裝成一個(gè)個(gè)批次(RecordBatch)。

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

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

7) 在封裝Batch批次和Request請(qǐng)求的過程中, 還涉及一個(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ù), 對(duì)于普通機(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ā)生了好幾次上下文的切換, 所以相對(duì)來說是比較消耗性能的

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緩存中僅僅會(huì)拷貝一個(gè)描述符過去，不會(huì)拷貝數(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)零拷貝, 對(duì)應(yīng)java里面的 MappedByteBuffer 和 FileChannel.transferIO。

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

transferTo() 方法會(huì)將數(shù)據(jù)從文件通道傳輸?shù)搅私o定的可寫字節(jié)通道。在其內(nèi)部它依賴底層操作系統(tǒng)對(duì)零拷貝的支持;在 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 中, 壓縮可能會(huì)發(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)和請(qǐng)求(Request), 在這個(gè)過程中, Kafka還有一個(gè)重要的設(shè)計(jì)理念 即內(nèi)存池方案, 這里就詳細(xì)講述下內(nèi)存池的實(shí)現(xiàn)過程.

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

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

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

5) 封裝批次會(huì)用到內(nèi)存, Sender發(fā)送完畢內(nèi)存會(huì)進(jìn)行回收, 在Java中如果頻繁操作內(nèi)存和回收,會(huì)遇到頭疼的FullGC的問題, 工作線程的性能就會(huì)降低, 整個(gè)生產(chǎn)者的性能就會(huì)受到影響, Kafka的解決方案就是內(nèi)存池, 對(duì)內(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)存, 一條消息過來后會(huì)向內(nèi)存池申請(qǐng)內(nèi)存塊, 申請(qǐng)完后封裝批次并寫入數(shù)據(jù), sender線程就會(huì)發(fā)送并響應(yīng), 然后清空內(nèi)存放回內(nèi)存池里面進(jìn)行反復(fù)使用, 這樣就大大減少了GC的頻率, 保證了生產(chǎn)者的穩(wěn)定和高效, 性能會(huì)大大提高

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è)請(qǐng)求流轉(zhuǎn)的路徑如下:

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

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

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

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

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

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

7) 在整個(gè)過程中涉及到2個(gè)參數(shù):num.network.threads = 3 和 num.io.threads = 8 如果感覺默認(rèn)參數(shù)性能不夠好的話, 可以對(duì)這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ā)的請(qǐng)求的

5總結(jié)

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