Java工程師的進(jìn)階之路Kafka篇

作者：佚名 2021-04-26 09:25:10

開發(fā)

后端

Kafka Kafka 是最初由 Linkedin 公司開發(fā)，是一個分布式、支持分區(qū)的（partition）、多副本的（replica），基于zookeeper 協(xié)調(diào)的分布式消息系統(tǒng)，它的最大的特性就是可以實時的處理大量數(shù)據(jù)以滿足各種需求場景。

創(chuàng)新互聯(lián)建站服務(wù)項目包括平城網(wǎng)站建設(shè)、平城網(wǎng)站制作、平城網(wǎng)頁制作以及平城網(wǎng)絡(luò)營銷策劃等。多年來，我們專注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢、行業(yè)經(jīng)驗、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機(jī)構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，平城網(wǎng)站推廣取得了明顯的社會效益與經(jīng)濟(jì)效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到平城省份的部分城市，未來相信會繼續(xù)擴(kuò)大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！

 1. Kafka 誕生背景

Kafka 是最初由 Linkedin 公司開發(fā)，是一個分布式、支持分區(qū)的（partition）、多副本的（replica），基于 zookeeper 協(xié)調(diào)的分布式消息系統(tǒng)，它的最大的特性就是可以實時的處理大量數(shù)據(jù)以滿足各種需求場景：比如基于 hadoop 的批處理系統(tǒng)、低延遲的實時系統(tǒng)、storm/Spark 流式處理引擎，web/nginx 日志、訪問日志，消息服務(wù)等等，用 scala 語言編寫，Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會并成為頂級開源項目。

當(dāng)今社會各種應(yīng)用系統(tǒng)諸如商業(yè)、社交、搜索、瀏覽等像信息工廠一樣不斷的生產(chǎn)出各種信息，在大數(shù)據(jù)時代，我們面臨如下幾個挑戰(zhàn)：

1.  如何收集這些巨大的信息;
2.  如何分析它;
3.  如何及時做到如上兩點;

以上幾個挑戰(zhàn)形成了一個業(yè)務(wù)需求模型，即 生產(chǎn)者生產(chǎn)（produce）各種信息，消費者消費（consume）（處理分析）這些信息，而在生產(chǎn)者與消費者之間，需要一個溝通兩者的橋梁-消息系統(tǒng) 。從一個微觀層面來說，這種需求也可理解為不同的系統(tǒng)之間如何傳遞消息。

Kafka 一個分布式消息系統(tǒng)應(yīng)運而生：

1.  Kafka-由 linked-in 開源；
2.  kafka-即是解決上述這類問題的一個框架，它實現(xiàn)了生產(chǎn)者和消費者之間的無縫連接；
3.  kafka-高產(chǎn)出的分布式消息系統(tǒng)(A high-throughput distributed messaging system)；

2. 為何使用消息系統(tǒng)

•  解耦 ：

允許你獨立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。

•  冗余 ：

消息隊列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理，通過這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險。許多消息隊列所采用的"插入-獲取-刪除"范式中，在把一個消息從隊列中刪除之前，需要你的處理系統(tǒng)明確的指出該消息已經(jīng)被處理完畢，從而確保你的數(shù)據(jù)被安全的保存直到你使用完畢。

•  擴(kuò)展性 ：

因為消息隊列解耦了你的處理過程，所以增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。

•  靈活性 & 峰值處理能力 ：

在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會因為突發(fā)的超負(fù)荷的請求而完全崩潰。

•  可恢復(fù)性 ：

系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。消息隊列降低了進(jìn)程間的耦合度，所以即使一個處理消息的進(jìn)程掛掉，加入隊列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。

•  順序保證 ：

在大多使用場景下，數(shù)據(jù)處理的順序都很重要。大部分消息隊列本來就是排序的，并且能保證數(shù)據(jù)會按照特定的順序來處理。（Kafka 保證一個 Partition 內(nèi)的消息的有序性）

•  緩沖 ：

有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費消息的處理速度不一致的情況。

•  異步通信 ：

很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機(jī)制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。

3. Kafka 基本架構(gòu)

3.1. 拓?fù)浣Y(jié)構(gòu)

3.2. 名詞概念

•  producer ：消息生產(chǎn)者，發(fā)布消息到 kafka 集群的終端或服務(wù)。
•  broker ：kafka 集群中包含的服務(wù)器。
•  topic ：每條發(fā)布到 kafka 集群的消息屬于的類別，即 kafka 是面向 topic 的。
•  partition ：partition 是物理上的概念，每個 topic 包含一個或多個 partition。kafka 分配的單位是 partition。
•  consumer ：從 kafka 集群中消費消息的終端或服務(wù)。
•  consumer group ：high-level consumer API 中，每個 consumer 都屬于一個 consumer group，每條消息只能被 consumer group 中的一個 Consumer 消費，但可以被多個 consumer group 消費。
•  replica ：partition 的副本，保障 partition 的高可用。
•  leader ：replica 中的一個角色， producer 和 consumer 只跟 leader 交互。
•  follower ：replica 中的一個角色，從 leader 中復(fù)制數(shù)據(jù)。
•  controller ：kafka 集群中的其中一個服務(wù)器，用來進(jìn)行 leader election 以及 各種 failover。
•  zookeeper ：kafka 通過 zookeeper 來存儲集群的 meta 信息。

4. Kafka 基本特性

1.  高吞吐量、低延遲 ：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒；
2.  可擴(kuò)展性 ：kafka集群支持熱擴(kuò)展；
3.  持久性、可靠性 ：消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失；
4.  容錯性 ：允許集群中節(jié)點失?。ㄈ舾北緮?shù)量為n,則允許n-1個節(jié)點失敗）；
5.  高并發(fā) ：支持?jǐn)?shù)千個客戶端同時讀寫；

4.1. 設(shè)計思想

•  consumergroup ：各個 consumer 可以組成一個組，每個消息只能被組中的一個 consumer 消費，如果一個消息可以被多個 consumer 消費的話，那么這些 consumer 必須在不同的組。
•  消息狀態(tài) ：在 Kafka 中，消息的狀態(tài)被保存在 consumer 中，broker 不會關(guān)心哪個消息被消費了被誰消費了，只記錄一個 offset 值（指向 partition 中下一個要被消費的消息位置），這就意味著如果 consumer 處理不好的話，broker 上的一個消息可能會被消費多次。
•  消息持久化 ：Kafka 中會把消息持久化到本地文件系統(tǒng)中，并且保持極高的效率。
•  消息有效期 ：Kafka 會長久保留其中的消息，以便 consumer 可以多次消費，當(dāng)然其中很多細(xì)節(jié)是可配置的。
•  批量發(fā)送 ：Kafka 支持以消息集合為單位進(jìn)行批量發(fā)送，以提高 push 效率。
•  push-and-pull : Kafka 中的 Producer 和 consumer 采用的是 push-and-pull 模式，即 Producer 只管向 broker push 消息，consumer 只管從 broker pull 消息，兩者對消息的生產(chǎn)和消費是異步的。Kafka集群中 broker 之間的關(guān)系：不是主從關(guān)系，各個 broker 在集群中地位一樣，我們可以隨意的增加或刪除任何一個 broker 節(jié)點。
•  負(fù)載均衡方面 ：Kafka 提供了一個 metadata API 來管理 broker 之間的負(fù)載（對 Kafka 0.8.x 而言，對于 0.7.x 主要靠 zookeeper 來實現(xiàn)負(fù)載均衡）。
•  同步異步 ：Producer 采用異步 push 方式，極大提高 Kafka 系統(tǒng)的吞吐率（可以通過參數(shù)控制是采用同步還是異步方式）。
•  分區(qū)機(jī)制 partition ：Kafka 的 broker 端支持消息分區(qū)，Producer 可以決定把消息發(fā)到哪個分區(qū)，在一個分區(qū)中消息的順序就是 Producer 發(fā)送消息的順序，一個主題中可以有多個分區(qū)，具體分區(qū)的數(shù)量是可配置的。分區(qū)的意義很重大，后面的內(nèi)容會逐漸體現(xiàn)。
•  離線數(shù)據(jù)裝載 ：Kafka 由于對可拓展的數(shù)據(jù)持久化的支持，它也非常適合向 Hadoop 或者數(shù)據(jù)倉庫中進(jìn)行數(shù)據(jù)裝載。
•  插件支持 ：現(xiàn)在不少活躍的社區(qū)已經(jīng)開發(fā)出不少插件來拓展 Kafka 的功能，如用來配合 Storm、Hadoop、flume 相關(guān)的插件。

4.2. 應(yīng)用場景

•  日志收集 ：一個公司可以用Kafka可以收集各種服務(wù)的 log，通過 kafka 以統(tǒng)一接口服務(wù)的方式開放給各種 consumer，例如 hadoop、Hbase、Solr 等。
•  消息系統(tǒng) ：解耦和生產(chǎn)者和消費者、緩存消息等。
•  用戶活動跟蹤 ：Kafka 經(jīng)常被用來記錄 web 用戶或者 app 用戶的各種活動，如瀏覽網(wǎng)頁、搜索、點擊等活動，這些活動信息被各個服務(wù)器發(fā)布到 kafka 的 topic 中，然后訂閱者通過訂閱這些 topic 來做實時的監(jiān)控分析，或者裝載到 hadoop、數(shù)據(jù)倉庫中做離線分析和挖掘。
•  運營指標(biāo) ：Kafka 也經(jīng)常用來記錄運營監(jiān)控數(shù)據(jù)。包括收集各種分布式應(yīng)用的數(shù)據(jù)，生產(chǎn)各種操作的集中反饋，比如報警和報告。
•  流式處理 ：比如 spark streaming 和 storm

5. Push 模式 vs Pull 模式

5.1. 點對點模式

如上圖所示，點對點模式通常是基于拉取或者輪詢的消息傳送模型，這個模型的特點是發(fā)送到隊列的消息被一個且只有一個消費者進(jìn)行處理。生產(chǎn)者將消息放入消息隊列后，由消費者主動的去拉取消息進(jìn)行消費。點對點模型的的優(yōu)點是消費者拉取消息的頻率可以由自己控制。但是消息隊列是否有消息需要消費，在消費者端無法感知，所以在消費者端需要額外的線程去監(jiān)控。

5.2. 發(fā)布訂閱模式

如上圖所示，發(fā)布訂閱模式是一個基于消息送的消息傳送模型，改模型可以有多種不同的訂閱者。生產(chǎn)者將消息放入消息隊列后，隊列會將消息推送給訂閱過該類消息的消費者（類似微信公眾號）。由于是消費者被動接收推送，所以無需感知消息隊列是否有待消費的消息！但是 consumer1、consumer2、consumer3 由于機(jī)器性能不一樣，所以處理消息的能力也會不一樣，但消息隊列卻無法感知消費者消費的速度！所以推送的速度成了發(fā)布訂閱模模式的一個問題！假設(shè)三個消費者處理速度分別是 8M/s、5M/s、2M/s，如果隊列推送的速度為5M/s，則 consumer3 無法承受！如果隊列推送的速度為 2M/s，則 consumer1、consumer2 會出現(xiàn)資源的極大浪費！

5.3. Kafka 的選擇

作為一個消息系統(tǒng)， Kafka 遵循了傳統(tǒng)的方式，選擇由 Producer 向 broker push 消息并由 Consumer 從 broker pull 消息 。一些日志收集系統(tǒng) (logging-centric system)，比如 Facebook 的 Scribe 和 Cloudera 的 Flume，采用 push 模式。事實上，push 模式和 pull 模式各有優(yōu)劣。

push 模式很難適應(yīng)消費速率不同的消費者，因為消息發(fā)送速率是由 broker 決定的。push 模式的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成 Consumer 來不及處理消息，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞。而 pull 模式則可以根據(jù) Consumer 的消費能力以適當(dāng)?shù)乃俾氏M消息。

對于 Kafka 而言，pull 模式更合適。pull 模式可簡化 broker 的設(shè)計，Consumer 可自主控制消費消息的速率，同時 Consumer 可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現(xiàn)不同的傳輸語義 。

6. Kafka 工作流程

6.1. 發(fā)送數(shù)據(jù)

我們看上面的架構(gòu)圖中，producer 就是生產(chǎn)者，是數(shù)據(jù)的入口。注意看圖中的紅色箭頭， Producer 在寫入數(shù)據(jù)的時候永遠(yuǎn)的找 leader，不會直接將數(shù)據(jù)寫入 follower ！那 leader 怎么找呢？寫入的流程又是什么樣的呢？我們看下圖：

1.  先從集群獲取分區(qū)的 leader；
2.  producer 將消息發(fā)送給 leader；
3.  Leader 將消息寫入本地文件；
4.  followers 從l eader 拉取消息；
5.  followers 將消息寫入本地后向 leader 發(fā)送 ACK 確認(rèn)；
6.  leader 收到所有副本的 ACK 后向 producer 發(fā)送 ACK 確認(rèn)；

6.1.1. 保證消息有序

需要注意的一點是，消息寫入 leader 后，follower 是主動的去 leader 進(jìn)行同步的！producer 采用 push 模式將數(shù)據(jù)發(fā)布到 broker，每條消息追加到分區(qū)中，順序?qū)懭氪疟P，所以保證同一分區(qū)內(nèi)的數(shù)據(jù)是有序的 ！寫入示意圖如下：

6.1.2. 消息負(fù)載分區(qū)

上面說到數(shù)據(jù)會寫入到不同的分區(qū)，那 kafka 為什么要做分區(qū)呢？相信大家應(yīng)該也能猜到，分區(qū)的主要目的是：

•  方便擴(kuò)展 ：因為一個 topic 可以有多個 partition，所以我們可以通過擴(kuò)展機(jī)器去輕松的應(yīng)對日益增長的數(shù)據(jù)量。
•  提高并發(fā) ：以 partition 為讀寫單位，可以多個消費者同時消費數(shù)據(jù)，提高了消息的處理效率。

熟悉負(fù)載均衡的朋友應(yīng)該知道，當(dāng)我們向某個服務(wù)器發(fā)送請求的時候，服務(wù)端可能會對請求做一個負(fù)載，將流量分發(fā)到不同的服務(wù)器，那在 kafka 中，如果某個 topic 有多個 partition，producer 又怎么知道該將數(shù)據(jù)發(fā)往哪個 partition 呢？kafka 中有幾個原則：

1.  partition 在寫入的時候可以指定需要寫入的 partition，如果有指定，則寫入對應(yīng)的 partition；
2.   如果沒有指定 partition，但是設(shè)置了數(shù)據(jù)的 key，則會根據(jù) key 的值 hash 出一個 partition；
3.  如果既沒指定 partition，又沒有設(shè)置 key，則會輪詢選出一個 partition；

6.1.3. 保證消息不丟

保證消息不丟失是一個消息隊列中間件的基本保證，那 producer 在向 kafka 寫入消息的時候， 怎么保證消息不丟失呢 ？其實上面的寫入流程圖中有描述出來， 那就是通過 ACK 應(yīng)答機(jī)制！在生產(chǎn)者向隊列寫入數(shù)據(jù)的時候可以設(shè)置參數(shù)來確定是否確認(rèn) kafka 接收到數(shù)據(jù)，這個參數(shù)可設(shè)置的值為 0、1、all 。

    0 代表 producer 往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性最低但是效率最高。

    1 代表 producer 往集群發(fā)送數(shù)據(jù)只要 leader 應(yīng)答就可以發(fā)送下一條，只確保 leader 發(fā)送成功。

    all 代表 producer 往集群發(fā)送數(shù)據(jù)需要所有的 follower 都完成從 leader 的同步才會發(fā)送下一條，確保 leader 發(fā)送成功和所有的副本都完成備份。安全性最高，但是效率最低。

最后要注意的是，如果往不存在的 topic 寫數(shù)據(jù)，能不能寫入成功呢？kafka 會自動創(chuàng)建 topic，分區(qū)和副本的數(shù)量根據(jù)默認(rèn)配置都是 1。

6.2. 保存數(shù)據(jù)

Producer 將數(shù)據(jù)寫入 kafka 后，集群就需要對數(shù)據(jù)進(jìn)行保存了！kafka 將數(shù)據(jù)保存在磁盤，可能在我們的一般的認(rèn)知里，寫入磁盤是比較耗時的操作，不適合這種高并發(fā)的組件。Kafka 初始會單獨開辟一塊磁盤空間，順序?qū)懭霐?shù)據(jù)（效率比隨機(jī)寫入高）。

6.2.1. Partition 結(jié)構(gòu)

前面說過了每個 topic 都可以分為一個或多個 partition，如果你覺得 topic 比較抽象，那 partition 就是比較具體的東西了！Partition 在服務(wù)器上的表現(xiàn)形式就是一個一個的文件夾，每個 partition 的文件夾下面會有多組 segment 文件，每組 segment 文件又包含 .index 文件、.log 文件、.timeindex 文件（早期版本中沒有）三個文件， log 文件就實際是存儲 message 的地方，而 index 和 timeindex 文件為索引文件，用于檢索消息。

如上圖，這個 partition 有三組 segment 文件，每個 log 文件的大小是一樣的，但是存儲的 message 數(shù)量是不一定相等的（每條的 message 大小不一致）。文件的命名是以該 segment 最小 offset 來命名的，如 000.index 存儲 offset 為 0~368795 的消息， kafka 就是利用分段+索引的方式來解決查找效率的問題 。

6.2.2. Message結(jié)構(gòu)

上面說到 log 文件就實際是存儲 message 的地方，我們在 producer 往 kafka 寫入的也是一條一條的 message，那存儲在 log 中的 message 是什么樣子的呢？消息主要包含消息體、消息大小、offset、壓縮類型...我們重點需要知道的是下面三個：

1.  offset ：offset 是一個占 8byte 的有序 id 號，它可以唯一確定每條消息在 parition 內(nèi)的位置；
2.  消息大小 ：消息大小占用 4byte，用于描述消息的大??；
3.  消息體 ：消息體存放的是實際的消息數(shù)據(jù)（被壓縮過），占用的空間根據(jù)具體的消息而不一樣。

6.2.3. 存儲策略

無論消息是否被消費，kafka 都會保存所有的消息。那對于舊數(shù)據(jù)有什么刪除策略呢？

•  基于時間 ，默認(rèn)配置是 168 小時（7天）；
•  基于大小 ，默認(rèn)配置是 1073741824。

需要注意的是， kafka 讀取特定消息的時間復(fù)雜度是 O(1) O ( 1 ) ，所以這里刪除過期的文件并不會提高 kafka 的性能 ！

6.3. 消費數(shù)據(jù)

消息存儲在 log 文件后，消費者就可以進(jìn)行消費了。在講消息隊列通信的兩種模式的時候講到過點對點模式和發(fā)布訂閱模式。Kafka 采用的是發(fā)布訂閱模式，消費者主動的去 kafka 集群拉取消息，與 producer 相同的是，消費者在拉取消息的時候也是找 leader 去拉取 。

多個消費者可以組成一個消費者組（consumer group），每個消費者組都有一個組 id！同一個消費組者的消費者可以消費同一 topic 下不同分區(qū)的數(shù)據(jù)，但是不會組內(nèi)多個消費者消費同一分區(qū)的數(shù)據(jù)！我們看下圖：

圖示是消費者組內(nèi)的消費者小于 partition 數(shù)量的情況，所以會出現(xiàn)某個消費者消費多個 partition 數(shù)據(jù)的情況，消費的速度也就不及只處理一個 partition 的消費者的處理速度！ 如果是消費者組的消費者多于 partition 的數(shù)量，那會不會出現(xiàn)多個消費者消費同一個 partition 的數(shù)據(jù)呢 ？上面已經(jīng)提到過不會出現(xiàn)這種情況！ 多出來的消費者不消費任何 partition 的數(shù)據(jù) 。 所以在實際的應(yīng)用中，建議消費者組的 consumer 的數(shù)量與 partition 的數(shù)量一致 ！

在保存數(shù)據(jù)的小節(jié)里面，我們聊到了 partition 劃分為多組 segment，每個 segment 又包含 .log、.index、.timeindex 文件，存放的每條 message 包含 offset、消息大小、消息體……我們多次提到 segment 和 offset，查找消息的時候是怎么利用 segment+offset 配合查找的呢？假如現(xiàn)在需要查找一個 offset 為 368801 的 message 是什么樣的過程呢？我們先看看下面的圖：

    1.  先找到 offset 的 368801 message 所在的 segment 文件（利用二分法查找），這里找到的就是在第二個 segment 文件。

    2.  打開找到的 segment 中的 .index 文件（也就是 368796.index 文件，該文件起始偏移量為 368796+1，我們要查找的 offset 為 368801 的 message 在該 index 內(nèi)的偏移量為 368796+5=368801，所以這里要查找的相對 offset 為 5）。由于該文件采用的是稀疏索引的方式存儲著相對 offset 及對應(yīng) message 物理偏移量的關(guān)系，所以直接找相對 offset 為 5 的索引找不到，這里同樣利用二分法查找相對 offset 小于或者等于指定的相對 offset 的索引條目中最大的那個相對 offset，所以找到的是相對 offset為 4 的這個索引。

    3.  根據(jù)找到的相對 offset 為 4 的索引確定 message 存儲的物理偏移位置為 256。打開數(shù)據(jù)文件，從位置為 256 的那個地方開始順序掃描直到找到 offset 為 368801 的那條 Message。

這套機(jī)制是建立在 offset 為有序的基礎(chǔ)上，利用 segment+有序offset+稀疏索引+二分查找+順序查找 等多種手段來高效的查找數(shù)據(jù)。至此，消費者就能拿到需要處理的數(shù)據(jù)進(jìn)行處理了。那每個消費者又是怎么記錄自己消費的位置呢？在早期的版本中，消費者將消費到的 offset 維護(hù) zookeeper 中，consumer 每間隔一段時間上報一次，這里容易導(dǎo)致重復(fù)消費，且性能不好！在新的版本中消費者消費到的 offset 已經(jīng)直接維護(hù)在kafka 集群的 consumer_offsets 這個 topic 中了。 

 

當(dāng)前名稱：Java工程師的進(jìn)階之路Kafka篇
URL分享：http://m.5511xx.com/article/coidsdc.html