知識(shí)分享：Java中的Kafka
1. Kafka 誕生背景
Kafka 是最初由 Linkedin 公司開發(fā)，是一個(gè)分布式、支持分區(qū)的（partition）、多副本的（replica），基于 zookeeper 協(xié)調(diào)的分布式消息系統(tǒng)，它的最大的特性就是可以實(shí)時(shí)的處理大量數(shù)據(jù)以滿足各種需求場(chǎng)景：比如基于 hadoop 的批處理系統(tǒng)、低延遲的實(shí)時(shí)系統(tǒng)、storm/Spark 流式處理引擎，web/nginx 日志、訪問日志，消息服務(wù)等等，用 scala 語言編寫，Linkedin 于 2010 年貢獻(xiàn)給了 Apache 基金會(huì)并成為頂級(jí)開源項(xiàng)目。
當(dāng)今社會(huì)各種應(yīng)用系統(tǒng)諸如商業(yè)、社交、搜索、瀏覽等像信息工廠一樣不斷的生產(chǎn)出各種信息，在大數(shù)據(jù)時(shí)代，我們面臨如下幾個(gè)挑戰(zhàn)：
如何收集這些巨大的信息;
如何分析它;
如何及時(shí)做到如上兩點(diǎn);
以上幾個(gè)挑戰(zhàn)形成了一個(gè)業(yè)務(wù)需求模型，即?生產(chǎn)者生產(chǎn)（produce）各種信息，消費(fèi)者消費(fèi)（consume）（處理分析）這些信息，而在生產(chǎn)者與消費(fèi)者之間，需要一個(gè)溝通兩者的橋梁-消息系統(tǒng)。從一個(gè)微觀層面來說，這種需求也可理解為不同的系統(tǒng)之間如何傳遞消息。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
Kafka 一個(gè)分布式消息系統(tǒng)應(yīng)運(yùn)而生：
Kafka-由 linked-in 開源；
kafka-即是解決上述這類問題的一個(gè)框架，它實(shí)現(xiàn)了生產(chǎn)者和消費(fèi)者之間的無縫連接；
kafka-高產(chǎn)出的分布式消息系統(tǒng)(A high-throughput distributed messaging system)；
2. 為何使用消息系統(tǒng)
解耦：
允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
冗余：
消息隊(duì)列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理，通過這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險(xiǎn)。許多消息隊(duì)列所采用的"插入-獲取-刪除"范式中，在把一個(gè)消息從隊(duì)列中刪除之前，需要你的處理系統(tǒng)明確的指出該消息已經(jīng)被處理完畢，從而確保你的數(shù)據(jù)被安全的保存直到你使用完畢。
擴(kuò)展性：
因?yàn)橄㈥?duì)列解耦了你的處理過程，所以增大消息入隊(duì)和處理的頻率是很容易的，只要另外增加處理過程即可。
靈活性 & 峰值處理能力：
在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見。如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。
可恢復(fù)性：
系統(tǒng)的一部分組件失效時(shí)，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。
順序保證：
在大多使用場(chǎng)景下，數(shù)據(jù)處理的順序都很重要。大部分消息隊(duì)列本來就是排序的，并且能保證數(shù)據(jù)會(huì)按照特定的順序來處理。（Kafka 保證一個(gè) Partition 內(nèi)的消息的有序性）
緩沖：
有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度，解決生產(chǎn)消息和消費(fèi)消息的處理速度不一致的情況。
異步通信：
很多時(shí)候，用戶不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許用戶把一個(gè)消息放入隊(duì)列，但并不立即處理它。想向隊(duì)列中放入多少消息就放多少，然后在需要的時(shí)候再去處理它們。
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 是物理上的概念，每個(gè) topic 包含一個(gè)或多個(gè) partition。kafka 分配的單位是 partition。
consumer：從 kafka 集群中消費(fèi)消息的終端或服務(wù)。
consumer group：high-level consumer API 中，每個(gè) consumer 都屬于一個(gè) consumer group，每條消息只能被 consumer group 中的一個(gè) Consumer 消費(fèi)，但可以被多個(gè) consumer group 消費(fèi)。
replica：partition 的副本，保障 partition 的高可用。
Leader：replica 中的一個(gè)角色， producer 和 consumer 只跟 Leader 交互。
follower：replica 中的一個(gè)角色，從 Leader 中復(fù)制數(shù)據(jù)。
controller：kafka 集群中的其中一個(gè)服務(wù)器，用來進(jìn)行 Leader election 以及 各種 failover。
zookeeper：kafka 通過 zookeeper 來存儲(chǔ)集群的 meta 信息。
4. Kafka 基本特性
高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒；
可擴(kuò)展性：kafka集群支持熱擴(kuò)展；
持久性、可靠性：消息被持久化到本地磁盤，并且支持?jǐn)?shù)據(jù)備份防止數(shù)據(jù)丟失；
容錯(cuò)性：允許集群中節(jié)點(diǎn)失敗（若副本數(shù)量為n,則允許n-1個(gè)節(jié)點(diǎn)失?。?/p>
高并發(fā)：支持?jǐn)?shù)千個(gè)客戶端同時(shí)讀寫；
4.1. 設(shè)計(jì)思想
consumergroup：各個(gè) consumer 可以組成一個(gè)組，每個(gè)消息只能被組中的一個(gè) consumer 消費(fèi)，如果一個(gè)消息可以被多個(gè) consumer 消費(fèi)的話，那么這些 consumer 必須在不同的組。
消息狀態(tài)：在 Kafka 中，消息的狀態(tài)被保存在 consumer 中，broker 不會(huì)關(guān)心哪個(gè)消息被消費(fèi)了被誰消費(fèi)了，只記錄一個(gè) offset 值（指向 partition 中下一個(gè)要被消費(fèi)的消息位置），這就意味著如果 consumer 處理不好的話，broker 上的一個(gè)消息可能會(huì)被消費(fèi)多次。
消息持久化：Kafka 中會(huì)把消息持久化到本地文件系統(tǒng)中，并且保持極高的效率。
消息有效期：Kafka 會(huì)長(zhǎng)久保留其中的消息，以便 consumer 可以多次消費(fèi)，當(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 消息，兩者對(duì)消息的生產(chǎn)和消費(fèi)是異步的。Kafka集群中 broker 之間的關(guān)系：不是主從關(guān)系，各個(gè) broker 在集群中地位一樣，我們可以隨意的增加或刪除任何一個(gè) broker 節(jié)點(diǎn)。
負(fù)載均衡方面： Kafka 提供了一個(gè) metadata API 來管理 broker 之間的負(fù)載（對(duì) Kafka 0.8.x 而言，對(duì)于 0.7.x 主要靠 zookeeper 來實(shí)現(xiàn)負(fù)載均衡）。
同步異步：Producer 采用異步 push 方式，極大提高 Kafka 系統(tǒng)的吞吐率（可以通過參數(shù)控制是采用同步還是異步方式）。
分區(qū)機(jī)制 partition：Kafka 的 broker 端支持消息分區(qū)，Producer 可以決定把消息發(fā)到哪個(gè)分區(qū)，在一個(gè)分區(qū)中消息的順序就是 Producer 發(fā)送消息的順序，一個(gè)主題中可以有多個(gè)分區(qū)，具體分區(qū)的數(shù)量是可配置的。分區(qū)的意義很重大，后面的內(nèi)容會(huì)逐漸體現(xiàn)。
離線數(shù)據(jù)裝載：Kafka 由于對(duì)可拓展的數(shù)據(jù)持久化的支持，它也非常適合向 Hadoop 或者數(shù)據(jù)倉庫中進(jìn)行數(shù)據(jù)裝載。
插件支持：現(xiàn)在不少活躍的社區(qū)已經(jīng)開發(fā)出不少插件來拓展 Kafka 的功能，如用來配合 Storm、Hadoop、flume 相關(guān)的插件。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
4.2. 應(yīng)用場(chǎng)景
日志收集：一個(gè)公司可以用Kafka可以收集各種服務(wù)的 log，通過 kafka 以統(tǒng)一接口服務(wù)的方式開放給各種 consumer，例如 hadoop、Hbase、Solr 等。
消息系統(tǒng)：解耦和生產(chǎn)者和消費(fèi)者、緩存消息等。
用戶活動(dòng)跟蹤：Kafka 經(jīng)常被用來記錄 web 用戶或者 app 用戶的各種活動(dòng)，如瀏覽網(wǎng)頁、搜索、點(diǎn)擊等活動(dòng)，這些活動(dòng)信息被各個(gè)服務(wù)器發(fā)布到 kafka 的 topic 中，然后訂閱者通過訂閱這些 topic 來做實(shí)時(shí)的監(jiān)控分析，或者裝載到 hadoop、數(shù)據(jù)倉庫中做離線分析和挖掘。
運(yùn)營指標(biāo)：Kafka 也經(jīng)常用來記錄運(yùn)營監(jiān)控?cái)?shù)據(jù)。包括收集各種分布式應(yīng)用的數(shù)據(jù)，生產(chǎn)各種操作的集中反饋，比如報(bào)警和報(bào)告。
流式處理：比如 spark streaming 和 storm
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
5. Push 模式 vs Pull 模式
5.1. 點(diǎn)對(duì)點(diǎn)模式
如上圖所示，點(diǎn)對(duì)點(diǎn)模式通常是基于拉取或者輪詢的消息傳送模型，這個(gè)模型的特點(diǎn)是發(fā)送到隊(duì)列的消息被一個(gè)且只有一個(gè)消費(fèi)者進(jìn)行處理。生產(chǎn)者將消息放入消息隊(duì)列后，由消費(fèi)者主動(dòng)的去拉取消息進(jìn)行消費(fèi)。點(diǎn)對(duì)點(diǎn)模型的的優(yōu)點(diǎn)是消費(fèi)者拉取消息的頻率可以由自己控制。但是消息隊(duì)列是否有消息需要消費(fèi)，在消費(fèi)者端無法感知，所以在消費(fèi)者端需要額外的線程去監(jiān)控。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
5.2. 發(fā)布訂閱模式
如上圖所示，發(fā)布訂閱模式是一個(gè)基于消息送的消息傳送模型，改模型可以有多種不同的訂閱者。生產(chǎn)者將消息放入消息隊(duì)列后，隊(duì)列會(huì)將消息推送給訂閱過該類消息的消費(fèi)者（類似微信公眾號(hào)）。由于是消費(fèi)者被動(dòng)接收推送，所以無需感知消息隊(duì)列是否有待消費(fèi)的消息！但是 consumer1、consumer2、consumer3 由于機(jī)器性能不一樣，所以處理消息的能力也會(huì)不一樣，但消息隊(duì)列卻無法感知消費(fèi)者消費(fèi)的速度！所以推送的速度成了發(fā)布訂閱模模式的一個(gè)問題！假設(shè)三個(gè)消費(fèi)者處理速度分別是 8M/s、5M/s、2M/s，如果隊(duì)列推送的速度為5M/s，則 consumer3 無法承受！如果隊(duì)列推送的速度為 2M/s，則 consumer1、consumer2 會(huì)出現(xiàn)資源的極大浪費(fèi)！
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
5.3. Kafka 的選擇
作為一個(gè)消息系統(tǒng)，Kafka 遵循了傳統(tǒng)的方式，選擇由 Producer 向 broker push 消息并由 Consumer 從 broker pull 消息。一些日志收集系統(tǒng) (logging-centric system)，比如 Facebook 的 Scribe 和 Cloudera 的 Flume，采用 push 模式。事實(shí)上，push 模式和 pull 模式各有優(yōu)劣。
push 模式很難適應(yīng)消費(fèi)速率不同的消費(fèi)者，因?yàn)橄l(fā)送速率是由 broker 決定的。push 模式的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成 Consumer 來不及處理消息，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞。而 pull 模式則可以根據(jù) Consumer 的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息。
對(duì)于 Kafka 而言，pull 模式更合適。pull 模式可簡(jiǎn)化 broker 的設(shè)計(jì)，Consumer 可自主控制消費(fèi)消息的速率，同時(shí) Consumer 可以自己控制消費(fèi)方式——即可批量消費(fèi)也可逐條消費(fèi)，同時(shí)還能選擇不同的提交方式從而實(shí)現(xiàn)不同的傳輸語義。
6. Kafka 工作流程
6.1. 發(fā)送數(shù)據(jù)
我們看上面的架構(gòu)圖中，producer 就是生產(chǎn)者，是數(shù)據(jù)的入口。注意看圖中的紅色箭頭，Producer 在寫入數(shù)據(jù)的時(shí)候永遠(yuǎn)的找 Leader，不會(huì)直接將數(shù)據(jù)寫入 follower！那 Leader 怎么找呢？寫入的流程又是什么樣的呢？我們看下圖：
先從集群獲取分區(qū)的 Leader；
producer 將消息發(fā)送給 Leader；
Leader 將消息寫入本地文件；
followers 從l eader 拉取消息；
followers 將消息寫入本地后向 Leader 發(fā)送 ACK 確認(rèn)；
Leader 收到所有副本的 ACK 后向 producer 發(fā)送 ACK 確認(rèn)；
6.1.1. 保證消息有序
需要注意的一點(diǎn)是，消息寫入 Leader 后，follower 是主動(dòng)的去 Leader 進(jìn)行同步的！producer 采用 push 模式將數(shù)據(jù)發(fā)布到 broker，每條消息追加到分區(qū)中，順序?qū)懭氪疟P，所以保證同一分區(qū)內(nèi)的數(shù)據(jù)是有序的！寫入示意圖如下：
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
6.1.2. 消息負(fù)載分區(qū)
上面說到數(shù)據(jù)會(huì)寫入到不同的分區(qū)，那 kafka 為什么要做分區(qū)呢？相信大家應(yīng)該也能猜到，分區(qū)的主要目的是：
方便擴(kuò)展：因?yàn)橐粋€(gè) topic 可以有多個(gè) partition，所以我們可以通過擴(kuò)展機(jī)器去輕松的應(yīng)對(duì)日益增長(zhǎng)的數(shù)據(jù)量。
提高并發(fā)：以 partition 為讀寫單位，可以多個(gè)消費(fèi)者同時(shí)消費(fèi)數(shù)據(jù)，提高了消息的處理效率。
熟悉負(fù)載均衡的朋友應(yīng)該知道，當(dāng)我們向某個(gè)服務(wù)器發(fā)送請(qǐng)求的時(shí)候，服務(wù)端可能會(huì)對(duì)請(qǐng)求做一個(gè)負(fù)載，將流量分發(fā)到不同的服務(wù)器，那在 kafka 中，如果某個(gè) topic 有多個(gè) partition，producer 又怎么知道該將數(shù)據(jù)發(fā)往哪個(gè) partition 呢？kafka 中有幾個(gè)原則：
partition 在寫入的時(shí)候可以指定需要寫入的 partition，如果有指定，則寫入對(duì)應(yīng)的 partition；
如果沒有指定 partition，但是設(shè)置了數(shù)據(jù)的 key，則會(huì)根據(jù) key 的值 hash 出一個(gè) partition；
如果既沒指定 partition，又沒有設(shè)置 key，則會(huì)輪詢選出一個(gè) partition；
6.1.3. 保證消息不丟
保證消息不丟失是一個(gè)消息隊(duì)列中間件的基本保證，那 producer 在向 kafka 寫入消息的時(shí)候，怎么保證消息不丟失呢？其實(shí)上面的寫入流程圖中有描述出來，那就是通過 ACK 應(yīng)答機(jī)制！在生產(chǎn)者向隊(duì)列寫入數(shù)據(jù)的時(shí)候可以設(shè)置參數(shù)來確定是否確認(rèn) kafka 接收到數(shù)據(jù)，這個(gè)參數(shù)可設(shè)置的值為 0、1、all?。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ??? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
0 代表 producer 往集群發(fā)送數(shù)據(jù)不需要等到集群的返回，不確保消息發(fā)送成功。安全性最低但是效率最高。
1 代表 producer 往集群發(fā)送數(shù)據(jù)只要 Leader 應(yīng)答就可以發(fā)送下一條，只確保 Leader 發(fā)送成功。
all 代表 producer 往集群發(fā)送數(shù)據(jù)需要所有的 follower 都完成從 Leader 的同步才會(huì)發(fā)送下一條，確保 Leader 發(fā)送成功和所有的副本都完成備份。安全性最高，但是效率最低。
最后要注意的是，如果往不存在的 topic 寫數(shù)據(jù)，能不能寫入成功呢？kafka 會(huì)自動(dòng)創(chuàng)建 topic，分區(qū)和副本的數(shù)量根據(jù)默認(rèn)配置都是 1。
6.2. 保存數(shù)據(jù)
Producer 將數(shù)據(jù)寫入 kafka 后，集群就需要對(duì)數(shù)據(jù)進(jìn)行保存了！kafka 將數(shù)據(jù)保存在磁盤，可能在我們的一般的認(rèn)知里，寫入磁盤是比較耗時(shí)的操作，不適合這種高并發(fā)的組件。Kafka 初始會(huì)單獨(dú)開辟一塊磁盤空間，順序?qū)懭霐?shù)據(jù)（效率比隨機(jī)寫入高）。
6.2.1. Partition 結(jié)構(gòu)
前面說過了每個(gè) topic 都可以分為一個(gè)或多個(gè) partition，如果你覺得 topic 比較抽象，那 partition 就是比較具體的東西了！Partition 在服務(wù)器上的表現(xiàn)形式就是一個(gè)一個(gè)的文件夾，每個(gè) partition 的文件夾下面會(huì)有多組 segment 文件，每組 segment 文件又包含 .index 文件、.log 文件、.timeindex 文件（早期版本中沒有）三個(gè)文件， log 文件就實(shí)際是存儲(chǔ) message 的地方，而 index 和 timeindex 文件為索引文件，用于檢索消息。
如上圖，這個(gè) partition 有三組 segment 文件，每個(gè) log 文件的大小是一樣的，但是存儲(chǔ)的 message 數(shù)量是不一定相等的（每條的 message 大小不一致）。文件的命名是以該 segment 最小 offset 來命名的，如 000.index 存儲(chǔ) offset 為 0~368795 的消息，kafka 就是利用分段+索引的方式來解決查找效率的問題。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
6.2.2. Message結(jié)構(gòu)
上面說到 log 文件就實(shí)際是存儲(chǔ) message 的地方，我們?cè)?producer 往 kafka 寫入的也是一條一條的 message，那存儲(chǔ)在 log 中的 message 是什么樣子的呢？消息主要包含消息體、消息大小、offset、壓縮類型...我們重點(diǎn)需要知道的是下面三個(gè)：
offset：offset 是一個(gè)占 8byte 的有序 id 號(hào)，它可以唯一確定每條消息在 parition 內(nèi)的位置；
消息大小：消息大小占用 4byte，用于描述消息的大?。?/p>
消息體：消息體存放的是實(shí)際的消息數(shù)據(jù)（被壓縮過），占用的空間根據(jù)具體的消息而不一樣。
6.2.3. 存儲(chǔ)策略
無論消息是否被消費(fèi)，kafka 都會(huì)保存所有的消息。那對(duì)于舊數(shù)據(jù)有什么刪除策略呢？
基于時(shí)間，默認(rèn)配置是 168 小時(shí)（7天）；
基于大小，默認(rèn)配置是 1073741824。
需要注意的是，kafka 讀取特定消息的時(shí)間復(fù)雜度是?O(1)O(1)O(1)，所以這里刪除過期的文件并不會(huì)提高 kafka 的性能！
6.3. 消費(fèi)數(shù)據(jù)
消息存儲(chǔ)在 log 文件后，消費(fèi)者就可以進(jìn)行消費(fèi)了。在講消息隊(duì)列通信的兩種模式的時(shí)候講到過點(diǎn)對(duì)點(diǎn)模式和發(fā)布訂閱模式。Kafka 采用的是發(fā)布訂閱模式，消費(fèi)者主動(dòng)的去 kafka 集群拉取消息，與 producer 相同的是，消費(fèi)者在拉取消息的時(shí)候也是找 Leader 去拉取。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
多個(gè)消費(fèi)者可以組成一個(gè)消費(fèi)者組（consumer group），每個(gè)消費(fèi)者組都有一個(gè)組 id！同一個(gè)消費(fèi)組者的消費(fèi)者可以消費(fèi)同一 topic 下不同分區(qū)的數(shù)據(jù)，但是不會(huì)組內(nèi)多個(gè)消費(fèi)者消費(fèi)同一分區(qū)的數(shù)據(jù)！我們看下圖：
圖示是消費(fèi)者組內(nèi)的消費(fèi)者小于 partition 數(shù)量的情況，所以會(huì)出現(xiàn)某個(gè)消費(fèi)者消費(fèi)多個(gè) partition 數(shù)據(jù)的情況，消費(fèi)的速度也就不及只處理一個(gè) partition 的消費(fèi)者的處理速度！如果是消費(fèi)者組的消費(fèi)者多于 partition 的數(shù)量，那會(huì)不會(huì)出現(xiàn)多個(gè)消費(fèi)者消費(fèi)同一個(gè) partition 的數(shù)據(jù)呢？上面已經(jīng)提到過不會(huì)出現(xiàn)這種情況！多出來的消費(fèi)者不消費(fèi)任何 partition 的數(shù)據(jù)。所以在實(shí)際的應(yīng)用中，建議消費(fèi)者組的 consumer 的數(shù)量與 partition 的數(shù)量一致！
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
在保存數(shù)據(jù)的小節(jié)里面，我們聊到了 partition 劃分為多組 segment，每個(gè) segment 又包含 .log、.index、.timeindex 文件，存放的每條 message 包含 offset、消息大小、消息體……我們多次提到 segment 和 offset，查找消息的時(shí)候是怎么利用 segment+offset 配合查找的呢？假如現(xiàn)在需要查找一個(gè) offset 為 368801 的 message 是什么樣的過程呢？我們先看看下面的圖：
先找到 offset 的 368801 message 所在的 segment 文件（利用二分法查找），這里找到的就是在第二個(gè) segment 文件。
打開找到的 segment 中的 .index 文件（也就是 368796.index 文件，該文件起始偏移量為 368796+1，我們要查找的 offset 為 368801 的 message 在該 index 內(nèi)的偏移量為 368796+5=368801，所以這里要查找的相對(duì) offset 為 5）。由于該文件采用的是稀疏索引的方式存儲(chǔ)著相對(duì) offset 及對(duì)應(yīng) message 物理偏移量的關(guān)系，所以直接找相對(duì) offset 為 5 的索引找不到，這里同樣利用二分法查找相對(duì) offset 小于或者等于指定的相對(duì) offset 的索引條目中最大的那個(gè)相對(duì) offset，所以找到的是相對(duì) offset為 4 的這個(gè)索引。
根據(jù)找到的相對(duì) offset 為 4 的索引確定 message 存儲(chǔ)的物理偏移位置為 256。打開數(shù)據(jù)文件，從位置為 256 的那個(gè)地方開始順序掃描直到找到 offset 為 368801 的那條 Message。
這套機(jī)制是建立在 offset 為有序的基礎(chǔ)上，利用 segment+有序offset+稀疏索引+二分查找+順序查找 等多種手段來高效的查找數(shù)據(jù)。至此，消費(fèi)者就能拿到需要處理的數(shù)據(jù)進(jìn)行處理了。那每個(gè)消費(fèi)者又是怎么記錄自己消費(fèi)的位置呢？在早期的版本中，消費(fèi)者將消費(fèi)到的 offset 維護(hù) zookeeper 中，consumer 每間隔一段時(shí)間上報(bào)一次，這里容易導(dǎo)致重復(fù)消費(fèi)，且性能不好！在新的版本中消費(fèi)者消費(fèi)到的 offset 已經(jīng)直接維護(hù)在kafka 集群的 consumer_offsets 這個(gè) topic 中了。
學(xué)習(xí)更多，請(qǐng)點(diǎn)擊：https://www.bilibili.com/video/BV1jK4y1o7QM
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1dy4y1W7Db
? ? ? ? ? ? ? ? ? ? ? ? ? ? ???https://www.bilibili.com/video/BV1Qb4y1D75J
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1qo4y1f7Uw
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?https://www.bilibili.com/video/BV1s64y1i77s
作者：白菜Java自習(xí)室
鏈接：https://juejin.cn/post/6953903023454158879
來源：掘金
著作權(quán)歸作者所有。商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請(qǐng)注明出處。
