Kafka面試，看這一篇文章就夠了

如需轉載，務必在文章開頭、結尾加粗以下內容：本文轉自微信公衆號： WeCoding,原文連接：https://mp.weixin.qq.com/s/zxPz_aFEMrshApZQ727h4g

MQ（消息隊列）是跨進程通訊的方式之一，可理解爲異步rpc，上游系統對調用結果的態度每每是重要不緊急。使用消息隊列有如下好處：業務解耦、流量削峯、靈活擴展。接下來介紹消息中間件Kafka。
Kafka是什麼？
Kafka是一個分佈式的消息引擎。具備如下特徵

1. 可以發佈和訂閱消息流（相似於消息隊列）
2. 以容錯的、持久的方式存儲消息流
3. 多分區概念，提升了並行能力

Kafka架構總覽

Topic
消息的主題、隊列，每個消息都有它的topic，Kafka經過topic對消息進行歸類。Kafka中能夠將Topic從物理上劃分紅一個或多個分區（Partition），每一個分區在物理上對應一個文件夾，以」topicName_partitionIndex」的命名方式命名，該dir包含了這個分區的全部消息(.log)和索引文件(.index)，這使得Kafka的吞吐率能夠水平擴展。
Partition
每一個分區都是一個 順序的、不可變的消息隊列， 而且能夠持續的添加;分區中的消息都被分了一個序列號，稱之爲偏移量(offset)，在每一個分區中此偏移量都是惟一的。
producer在發佈消息的時候，能夠爲每條消息指定Key，這樣消息被髮送到broker時，會根據分區算法把消息存儲到對應的分區中（一個分區存儲多個消息），若是分區規則設置的合理，那麼全部的消息將會被均勻的分佈到不一樣的分區中，這樣就實現了負載均衡。

Broker
Kafka server，用來存儲消息，Kafka集羣中的每個服務器都是一個Broker，消費者將從broker拉取訂閱的消息
Producer
向Kafka發送消息，生產者會根據topic分發消息。生產者也負責把消息關聯到Topic上的哪個分區。最簡單的方式從分區列表中輪流選擇。也能夠根據某種算法依照權重選擇分區。算法可由開發者定義。
Cousumer
Consermer實例能夠是獨立的進程，負責訂閱和消費消息。消費者用consumerGroup來標識本身。同一個消費組能夠併發地消費多個分區的消息，同一個partition也能夠由多個consumerGroup併發消費，可是在consumerGroup中一個partition只能由一個consumer消費
CousumerGroup
Consumer Group：同一個Consumer Group中的Consumers，Kafka將相應Topic中的每一個消息只發送給其中一個Consumer


Kafka producer 設計原理
發送消息的流程

1.序列化消息&&.計算partition
根據key和value的配置對消息進行序列化,而後計算partition：
ProducerRecord對象中若是指定了partition，就使用這個partition。不然根據key和topic的partition數目取餘，若是key也沒有的話就隨機生成一個counter，使用這個counter來和partition數目取餘。這個counter每次使用的時候遞增。


2發送到batch&&喚醒Sender 線程
根據topic-partition獲取對應的batchs（Dueue<ProducerBatch>），而後將消息append到batch中.若是有batch滿了則喚醒Sender 線程。隊列的操做是加鎖執行，因此batch內消息時有序的。後續的Sender操做當前方法異步操做。

3.Sender把消息有序發到 broker（tp replia leader）
3.1 肯定tp relica leader 所在的broker

• Kafka中 每臺broker都保存了kafka集羣的metadata信息，metadata信息裏包括了每一個topic的全部partition的信息: leader, leader_epoch, controller_epoch, isr, replicas等;Kafka客戶端從任一broker均可以獲取到須要的metadata信息;sender線程經過metadata信息能夠知道tp leader的brokerId
• producer也保存了metada信息，同時根據metadata更新策略（按期更新metadata.max.age.ms、失效檢測，強制更新：檢查到metadata失效之後，調用metadata.requestUpdate()強制更新

public class PartitionInfo {
    private final String topic;
    private final int partition;
    private final Node leader;
    private final Node[] replicas;
    private final Node[] inSyncReplicas;
    private final Node[] offlineReplicas;
}
複製代碼

3.2 冪等性發送
爲實現Producer的冪等性，Kafka引入了Producer ID（即PID）和Sequence Number。對於每一個PID，該Producer發送消息的每一個<Topic, Partition>都對應一個單調遞增的Sequence Number。一樣，Broker端也會爲每一個<PID, Topic, Partition>維護一個序號，而且每Commit一條消息時將其對應序號遞增。對於接收的每條消息，若是其序號比Broker維護的序號）大一，則Broker會接受它，不然將其丟棄：

• 若是消息序號比Broker維護的序號差值比一大，說明中間有數據還沒有寫入，即亂序，此時Broker拒絕該消息，Producer拋出InvalidSequenceNumber
• 若是消息序號小於等於Broker維護的序號，說明該消息已被保存，即爲重複消息，Broker直接丟棄該消息，Producer拋出DuplicateSequenceNumber
• Sender發送失敗後會重試，這樣能夠保證每一個消息都被髮送到broker

4. Sender處理broker發來的produce response
一旦broker處理完Sender的produce請求，就會發送produce response給Sender，此時producer將執行咱們爲send（）設置的回調函數。至此producer的send執行完畢。


吞吐性&&延時：

• buffer.memory：buffer設置大了有助於提高吞吐性，可是batch太大會增大延遲，可搭配linger_ms參數使用
• linger_ms：若是batch太大，或者producer qps不高，batch添加的會很慢，咱們能夠強制在linger_ms時間後發送batch數據
• ack：producer收到多少broker的答覆纔算真的發送成功
• 0表示producer無需等待leader的確認(吞吐最高、數據可靠性最差)
• 1表明須要leader確認寫入它的本地log並當即確認
• -1/all 表明全部的ISR都完成後確認(吞吐最低、數據可靠性最高)

Sender線程和長鏈接
每初始化一個producer實例，都會初始化一個Sender實例，新增到broker的長鏈接。
代碼角度：每初始化一次KafkaProducer，都賦一個空的client

public KafkaProducer(final Map<String, Object> configs) { this(configs, null, null, null, null, null, Time.SYSTEM); }

複製代碼

終端查看TCP鏈接數：
lsof -p portNum -np | grep TCP


Consumer設計原理
poll消息

• 消費者經過fetch線程拉消息（單線程）
• 消費者經過心跳線程來與broker發送心跳。超時會認爲掛掉
• 每一個consumer group在broker上都有一個coordnator來管理，消費者加入和退出，以及消費消息的位移都由coordnator處理。

位移管理
consumer的消息位移表明了當前group對topic-partition的消費進度，consumer宕機重啓後能夠繼續從該offset開始消費。
在kafka0.8以前，位移信息存放在zookeeper上，因爲zookeeper不適合高併發的讀寫，新版本Kafka把位移信息當成消息，發往__consumers_offsets 這個topic所在的broker，__consumers_offsets默認有50個分區。
消息的key 是groupId+topic_partition,value 是offset.

Kafka Group 狀態

• Empty：初始狀態，Group 沒有任何成員，若是全部的 offsets 都過時的話就會變成 Dead
• PreparingRebalance：Group 正在準備進行 Rebalance
• AwaitingSync：Group 正在等待來 group leader 的 分配方案
• Stable：穩定的狀態（Group is stable）；
• Dead：Group 內已經沒有成員，而且它的 Metadata 已經被移除
  注意

重平衡reblance
當一些緣由致使consumer對partition消費再也不均勻時，kafka會自動執行reblance，使得consumer對partition的消費再次平衡。

何時發生rebalance？：

• 組訂閱topic數變動
• topic partition數變動
• consumer成員變動
• consumer 加入羣組或者離開羣組的時候
• consumer被檢測爲崩潰的時候

reblance過程


舉例1 consumer被檢測爲崩潰引發的reblance
好比心跳線程在timeout時間內沒和broker發送心跳，此時coordnator認爲該group應該進行reblance。接下來其餘consumer發來fetch請求後，coordnator將回復他們進行reblance通知。當consumer成員收到請求後，只有leader會根據分配策略進行分配，而後把各自的分配結果返回給coordnator。這個時候只有consumer leader返回的是實質數據，其餘返回的都爲空。收到分配方法後，consumer將會把分配策略同步給各consumer


舉例2 consumer加入引發的reblance

1. 使用join協議，表示有consumer 要加入到group中
2. 使用sync 協議，根據分配規則進行分配
   
   

(上圖圖片摘自網絡)


引伸：以上reblance機制存在的問題
在大型系統中，一個topic可能對應數百個consumer實例。這些consumer陸續加入到一個空消費組將致使屢次的rebalance；此外consumer 實例啓動的時間不可控，頗有可能超出coordinator肯定的rebalance timeout(即max.poll.interval.ms)，將會再次觸發rebalance，而每次rebalance的代價又至關地大，由於不少狀態都須要在rebalance前被持久化，而在rebalance後被從新初始化。


新版本改進
經過延遲進入PreparingRebalance狀態減小reblance次數

新版本新增了group.initial.rebalance.delay.ms參數。空消費組接受到成員加入請求時，不當即轉化到PreparingRebalance狀態來開啓reblance。當時間超過group.initial.rebalance.delay.ms後，再把group狀態改成PreparingRebalance（開啓reblance）。實現機制是在coordinator底層新增一個group狀態：InitialReblance。假設此時有多個consumer陸續啓動，那麼group狀態先轉化爲InitialReblance，待group.initial.rebalance.delay.ms時間後，再轉換爲PreparingRebalance（開啓reblance）






Broker設計原理
Broker 是Kafka 集羣中的節點。負責處理生產者發送過來的消息，消費者消費的請求。以及集羣節點的管理等。因爲涉及內容較多，先簡單介紹，後續專門抽出一篇文章分享

broker zk註冊

broker消息存儲

• Kafka的消息以二進制的方式緊湊地存儲，節省了很大空間
• 此外消息存在ByteBuffer而不是堆，這樣broker進程掛掉時，數據不會丟失，同時避免了gc問題
• 經過零拷貝和順序尋址，讓消息存儲和讀取速度都很是快
• 處理fetch請求的時候經過zero-copy 加快速度

broker狀態數據
broker設計中，每臺機器都保存了相同的狀態數據。主要包括如下：

• controller所在的broker ID，即保存了當前集羣中controller是哪臺broker
• 集羣中全部broker的信息：好比每臺broker的ID、機架信息以及配置的若干組鏈接信息
• 集羣中全部節點的信息：嚴格來講，它和上一個有些重複，不過此項是按照broker ID和監聽器類型進行分組的。對於超大集羣來講，使用這一項緩存能夠快速地定位和查找給定節點信息，而無需遍歷上一項中的內容，算是一個優化吧
• 集羣中全部分區的信息：所謂分區信息指的是分區的leader、ISR和AR信息以及當前處於offline狀態的副本集合。這部分數據按照topic-partitionID進行分組，能夠快速地查找到每一個分區的當前狀態。（注：AR表示assigned replicas，即建立topic時爲該分區分配的副本集合）

broker負載均衡
分區數量負載：各臺broker的partition數量應該均勻
partition Replica分配算法以下：

1. 將全部Broker（假設共n個Broker）和待分配的Partition排序
2. 將第i個Partition分配到第（i mod n）個Broker上
3. 將第i個Partition的第j個Replica分配到第（(i + j) mod n）個Broker上

容量大小負載：每臺broker的硬盤佔用大小應該均勻
在kafka1.1以前，Kafka可以保證各臺broker上partition數量均勻，但因爲每一個partition內的消息數不一樣，可能存在不一樣硬盤之間內存佔用差別大的狀況。在Kafka1.1中增長了副本跨路徑遷移功能kafka-reassign-partitions.sh，咱們能夠結合它和監控系統，實現自動化的負載均衡


Kafka高可用
在介紹kafka高可用以前先介紹幾個概念

• 同步複製：要求全部能工做的Follower都複製完，這條消息纔會被認爲commit，這種複製方式極大的影響了吞吐率
• 異步複製：Follower異步的從Leader pull數據，data只要被Leader寫入log認爲已經commit，這種狀況下若是Follower落後於Leader的比較多，若是Leader忽然宕機，會丟失數據

Isr
Kafka結合同步複製和異步複製，使用ISR（與Partition Leader保持同步的Replica列表）的方式在確保數據不丟失和吞吐率之間作了平衡。Producer只需把消息發送到Partition Leader，Leader將消息寫入本地Log。Follower則從Leader pull數據。Follower在收到該消息向Leader發送ACK。一旦Leader收到了ISR中全部Replica的ACK，該消息就被認爲已經commit了，Leader將增長HW而且向Producer發送ACK。這樣若是leader掛了，只要Isr中有一個replica存活，就不會丟數據。


Isr動態更新
Leader會跟蹤ISR，若是ISR中一個Follower宕機，或者落後太多，Leader將把它從ISR中移除。這裏所描述的「落後太多」指Follower複製的消息落後於Leader後的條數超過預約值（replica.lag.max.messages）或者Follower超過必定時間（replica.lag.time.max.ms）未向Leader發送fetch請求。


broker Nodes In Zookeeper
/brokers/topics/[topic]/partitions/[partition]/state 保存了topic-partition的leader和Isr等信息

Controller負責broker故障檢查&&故障轉移（fail/recover）

1. Controller在Zookeeper上註冊Watch，一旦有Broker宕機，其在Zookeeper對應的znode會自動被刪除，Zookeeper會觸發 Controller註冊的watch，Controller讀取最新的Broker信息
2. Controller肯定set_p，該集合包含了宕機的全部Broker上的全部Partition
3. 對set_p中的每個Partition，選舉出新的leader、Isr，並更新結果。
   　　3.1 從/brokers/topics/[topic]/partitions/[partition]/state讀取該Partition當前的ISR
   　　3.2 決定該Partition的新Leader和Isr。若是當前ISR中有至少一個Replica還倖存，則選擇其中一個做爲新Leader，新的ISR則包含當前ISR中全部倖存的Replica。不然選擇該Partition中任意一個倖存的Replica做爲新的Leader以及ISR（該場景下可能會有潛在的數據丟失）
   

　 3.3 更新Leader、ISR、leader_epoch、controller_epoch：寫入/brokers/topics/[topic]/partitions/[partition]/state

1. 直接經過RPC向set_p相關的Broker發送LeaderAndISRRequest命令。Controller能夠在一個RPC操做中發送多個命令從而提升效率。

Controller掛掉
每一個 broker 都會在 zookeeper 的臨時節點 "/controller" 註冊 watcher，當 controller 宕機時 "/controller" 會消失，觸發broker的watch，每一個 broker 都嘗試建立新的 controller path，只有一個競選成功並當選爲 controller。


使用Kafka如何保證冪等性
不丟消息

• 首先kafka保證了對已提交消息的at least保證
• Sender有重試機制
• producer業務方在使用producer發送消息時，註冊回調函數。在onError方法中重發消息
• consumer 拉取到消息後，處理完畢再commit，保證commit的消息必定被處理完畢

不重複

• consumer拉取到消息先保存，commit成功後刪除緩存數據

Kafka高性能

• partition提高了併發
• zero-copy
• 順序寫入
• 消息彙集batch
• 頁緩存

業務方對 Kafka producer的優化

• 增大producer數量
• ack配置
• batch