Kafka原了解析
文章目錄
- Kafka原了解析
- 一、什麼是消息隊列？什麼是Kafka？
-
- 1.1 消息隊列
- 1.2 Kafka概念與基礎架構
- 二、Kafka架構深入！！
-
- 2.1 Kafka存儲模型
- 2.2 Kafka Producer
-
- 2.2.1 資料分區
- 2.2.2 資料可靠性保證
- 2.2.3 Exactly-Once語義
- 2.2.4 Producer資料送出流程
- 2.3 Kafka Consumer
-
- 2.3.1 消費模式
- 2.3.2 分區配置設定政策與重平衡(rebalance)
- 2.3.3 offset維護
- 2.4 Kafka高效讀寫的保證～
- 2.5 Kafka如何通過ZK來進行選舉和狀态更新？
- 2.6 Kafka事務！
- 2.7 手動維護offset
Java、大資料開發學習要點（持續更新中…）
一、什麼是消息隊列？什麼是Kafka？
1.1 消息隊列
消息隊列就是用于資料生産方和消費方解耦合的中間件。顧名思義，主體就是一個隊列的形式收集消息，資料在消費端按照FIFO的原則被消費。
1.2 Kafka概念與基礎架構
Kafka是一個基于釋出訂閱模式的分布式消息隊列
Kafka基礎架構如下圖所示：
- 為了友善擴充和提高吞吐量，每個Topic分為多個Partition。
- 配合Topic分區的設計，提出消費者組的概念，每個組内的消費者并行消費一個Topic中的不同Partition中的資料。(但是整體上還是一個Topic為一個隊列，并且消費者數與分區數可以不同)
- 為了高可用性，為每個Partition在不同broker節點存放了副本。(僅僅是Follower不接受請求)
二、Kafka架構深入！！
2.1 Kafka存儲模型
Kafka中消息是以Topic進行分類的，生産者生産消息，消費者消費消息，都是面向Topic的。而Topic在實體上的存儲是分區存儲的，即按Partition分布式存儲。每個Partition中的資料又是順序寫入log檔案1中進行存儲。
但這樣還是會出現分區log檔案過大，導緻的讀取性能下降的問題。是以Kafka将log檔案切分成了segment，每個segment由 .log資料存儲檔案 和 .index索引檔案 和 .timeindex檔案組成。詳細的結構如下圖所示：
而每個log檔案和index檔案的命名就是 檔案中起始資料的偏移量，一個segment中由index定位到對應log檔案中執行資料的原理如下圖：
index檔案中根據需要查找的offset根據儲存起始偏移量(檔案名)的相對偏移量，定位到log中資料真實的位置。
- 1注解1：類似HBase中的順序寫入HFile，磁頭尋址次數少，順序讀/寫性能好。
2.2 Kafka Producer
2.2.1 資料分區
如前面所提到的，Kafka是分布式的消息隊列，分區的目的是：
充分利用分布式的優勢
- 分區友善後期拓展
- 分區能夠增大讀寫的吞吐量。
分區原則：指定了Partition的直接寫入對應Partition、否則根據key進行hash後對分區數取餘、沒有key的通過random-robin算法得到分區(第一個得到一個随機數，後續的在此基礎上自增)
2.2.2 資料可靠性保證
為保證Producer發送的資料，能可靠的發送到指定的Topic，Topic的每個Partition收到Producer發送的資料後，都需要向Producer發送ack響應，如果Producer收到ack，就會進行下一輪的發送(不是同步的，而是異步的，分批次檢查前面發送的消息ack是否收到)，否則重新發送資料。
- ack應答機制：Kafka設定了三種類型的應答機制
:Producer不用等待broker的ack，但是broker未将資料寫入前當機會産生資料丢失。acks = 0
:Producer等待broker的ack，Partition的leader落盤成功後傳回ack，如果在follower同步成功之前leader故障，那麼将會資料丢失。acks = 1
(常用):Producer等待broker的ack，Partition的leader和follower全部落盤成功後才傳回ack。但是如果在follower同步完成後，broker發送ack之前，leader發生故障，那麼會造成資料重複。acks = -1
- ISR：ISR是用于解決Partition的leader等待所有follower同步，然而某些follower由于某些原因遲遲不能完成同步的問題的follower動态集合。那些無法完成同步的follower會被踢出集合；在其恢複後，同步了follower資料後再重新加入集合。
- 故障恢複：
- follower故障後會被踢出ISR，在其恢複後擷取集合的HW(high watermark，ISR中所有副本中LEO的最小值，LEO則是每個副本待寫入的偏移量)，重新向leader同步HW前的資料。完成後重新加入ISR。
- leader故障後，ZK在ISR中選出新的leader，為保證副本資料一緻性，follower會将自身HW後的資料截斷，重新向新的leader同步。
*注意：上述隻能保證副本之間的資料一緻性，并不能保證資料不丢失或者不重複。
2.2.3 Exactly-Once語義
在保證資料不丢失的場景下，一般将Kafka的ack應答設定為-1，那麼也就存在資料重複的可能性(At-Least-Once)。而要實作資料的Exactly-Once（每條資料在Kafka中有且隻有一條不會重複也不會丢失），需要額外實作幂等性，也即：
At-Least-Once + 幂等性 = Exactly-Once
為了實作幂等性，Producer端會生成一個PID，Producer發往同一個Partition的消息會附帶Sequence Number(offset)；而Broker端會對
<PID，Partition，SeqNum>
進行緩存，當收到相同主鍵的消息并且Sequence Number值比緩存值小則不再重複存儲。
幂等性在跨分區跨會話時會失效(當Producer挂了重新開機後，主鍵PID會發生變化) ——需要通過事務解決
2.2.4 Producer資料送出流程
Kafka的Producer發送消息采用的是異步發送的方式。在消息發送的過程中，涉及到了兩個線程——main線程和Sender線程，以及一個線程共享變量——RecordAccumulator。main線程将消息發送給RecordAccumulator，Sender線程不斷從RecordAccumulator中拉取消息發送到Kafka broker。
注意：
1.消息在Producer端就已經分區了。
2.消息是按照batch size發送的，但當資料未達到batch size逾時了也會被強制發送。
2.3 Kafka Consumer
2.3.1 消費模式
Kafka的消費模式是poll模式，就是每個消費者按照自己的消費能力在Broker中讀取資料。但有一個問題就是，如果沒有新的資料，那麼消費者就在循環中空轉，這個問題Kafka設定了一個短暫的timeout來讓消費者在沒有資料可以消費的時候等待一小會。
2.3.2 分區配置設定政策與重平衡(rebalance)
如果Consumer組中
Consumer數量 < Topic Partition數量
根據Random-Robin政策(實際是個輪詢，對多個topic的配置設定較為均衡)或者Range(劃分範圍，一個範圍内的給到一個消費者，隻保證每個topic配置設定的相對均衡)對Partition進行配置設定。此後這個消費者組中的消費者消費的Partition就被定下來了(若是
Consumer數量 > Topic Partition數量
，則多出來的消費者會被閑置)。
重平衡問題：重平衡需要借助 Kafka Broker 端的 Coordinator 元件(每個Broker都有，配置設定消費Topic對應分區的時候找到分區Leader所在的Broker的Coordinator)。
- 有三個條件會觸發重平衡：
- 消費者組内成員發生變更，這個變更包括了增加和減少消費者。注意這裡的減少有很大的可能是被動的，就是某個消費者崩潰退出了(或者心跳消息由于網絡原因延時過大)
- 主題的分區數發生變更，Kafka目前隻支援增加分區，當增加的時候就會觸發重平衡
- 訂閱的主題發生變化(新增主題，比如消費者組使用正規表達式訂閱主題)，而恰好又建立了對應的主題，就會觸發重平衡
- 重平衡的弊端：因為重平衡過程中，消費者無法從Kafka消費消息，這對Kafka的TPS影響極大，而如果Kafka集内節點較多，比如數百個，那重平衡可能會耗時極多。
- Rebalance 過程分為兩步：Join 和 Sync。
- Join，顧名思義就是加入組。這一步中，所有成員都向Coordinator發送JoinGroup請求，請求加入消費組。一旦所有成員都發送了JoinGroup請求，Coordinator會從中選擇一個Consumer擔任Leader的角色，并把組成員資訊以及訂閱資訊發給Leader。Leader負責消費配置設定方案的制定。
-
Sync，這一步Leader開始配置設定消費方案，即哪個Consumer負責消費哪些Topic的哪些Partition。一旦完成配置設定，Leader會将這個方案封裝進SyncGroup請求中發給Coordinator，非Leader也會發SyncGroup請求，隻是内容為空。Coordinator接收到配置設定方案之後會把方案塞進SyncGroup的response中發給各個Consumer。這樣組内的所有成員就都知道自己應該消費哪些分區了。
如果是Topic增加或者Partion增加，則由Leader consumer通知Coordinator進行rebalance
2.3.3 offset維護
comsumer需要記錄已經消費到的偏移量以便故障或者後續繼續消費。在0.9版本後Kafka将這些資訊都儲存在一個内置的Topic中(
__comsumer_offset
)，而此前的版本是儲存在ZK中的(1.優化效率，減輕ZK壓力 2.可以自己實作偏移量維護)
2.4 Kafka高效讀寫的保證～
- 如同HBase順序寫HFile檔案一樣，Kafka順序寫log檔案寫入磁盤效率極高，并且分段存儲并使用索引檔案加快查找(據Kafka官網文檔說比随機寫快6000倍)。
-
使用作業系統的Page Cache來緩存要寫入的資料，好處在于：
（1）寫入前可以做一些優化，提高磁盤寫入性能
（2）緩存也可以用于資料被讀取，當資料寫入與讀取速率相近的情況下，可以直接記憶體讀取。
（3）Page Cache非JVM記憶體，不會影響JVM，導緻GC的增加。同時，Kafka節點當機，資料還在此機器緩存。
-
零拷貝機制：
很多應用其實在檔案從磁盤拷貝到磁盤、從磁盤拷貝到Socket緩存，這些應用不需要接手這些資料。而一般的拷貝機制要經曆從例如磁盤 -> 核心page cache -> 應用緩存 -> 核心page cache -> 磁盤的過程，如果資料隻是單純的拷貝而不需要修改，那麼拷貝到應用緩存的步驟完全是多餘的。是以Kafka利用了作業系統提供的零拷貝機制，來減少不需要的系統調用和資料拷貝次數。
2.5 Kafka如何通過ZK來進行選舉和狀态更新？
首先，Kafka叢集啟動時，會從Broker中選舉一個Controller(分布式鎖實作搶先建立臨時節點的broker當選)，負責管理叢集Broker的上下線(監聽zk的/brokers/ids/節點)、所有Topic分區副本配置設定、leader選舉等工作。
當某個Broker挂了以後，Controller監聽到臨時節點/brokers/ids/中的變化，從ZK各個分區狀态資訊中擷取ISR(此時去除了挂掉節點所有的Partition，失去leader的Partition重新選舉leader)，并完成ZK各個分區狀态更新
2.6 Kafka事務！
Kafka事務在0.11版本後引入，主要解決的是 Producer在Exactly Once語義上跨分區跨會話的精準一次寫入，要麼成功要麼失敗。
-
Producer事務(斷點續傳)
為了實作跨分區跨會話的事務，每個Producer需要自己提供一個全局唯一的Transaction ID(對相同的Transaction ID，通過維護一個遞增producer epoch來使得相同Producer隻有最新事務有效)，并将Producer獲得的PID和Transaction ID綁定。這樣當Producer重新開機後就可以通過正在進行的Transaction ID獲得原來的PID。
為了管理Transaction，Kafka引入了一個新的元件Transaction Coordinator(事務排程器)。Producer就是通過和Transaction Coordinator互動獲得綁定的PID和對應的任務狀态。Transaction Coordinator還負責将事務所有寫入Kafka的一個内部Topic，這樣即使整個服務重新開機，由于事務狀态得到儲存，進行中的事務狀态可以得到恢複，進而繼續進行。
*注意：Kakfa事務復原不會直接去删除消息，而是将消息對Consumer不可見。
-
Consumer事務
Kafka對Consumer的事務較弱，一般是通過Consumer端自己實作精确一次性消費(将消費過程和送出offset作為一個原子操作實作)。
2.7 手動維護offset
實作精準一次性消費，手動維護offset是基礎，但也無法避免重複消費
Kafka消費偏移量的維護預設由Kafka通過維護Topic(
__comsumer_offset
)來實作，預設5s自動送出一次。但這樣做Kafka在記錄偏移量前當機，則消費者會重複消費。
單純的手動送出偏移量如果消費行為和手動送出行為不是一個原子行為，那麼消費者在消費完未送出偏移量期間當機，資料會發生重複消費的現象。解決方法有兩種：
- 在消費者端去重，在每次當機重新開機後對新消費資料去重。
- 将消費和偏移量送出綁定為原子化操作(事務)，消費端将offset維護到外部媒體中。