微服務(wù)架構(gòu)之–消息隊列Kafka圖解最全知識點
MQ(消息隊列)是跨進(jìn)程通信的方式之一，可理解為異步rpc，上游系統(tǒng)對調(diào)用結(jié)果的態(tài)度往往是重要不緊急。使用消息隊列有以下好處：業(yè)務(wù)解耦、流量削峰、靈活擴(kuò)展。接下來介紹消息中間件Kafka。
Kafka是什么?
Kafka是一個分布式的消息引擎。具有以下特征
- 能夠發(fā)布和訂閱消息流(類似于消息隊列)
- 以容錯的、持久的方式存儲消息流
- 多分區(qū)概念，提高了并行能力
Kafka架構(gòu)總覽
Topic
消息的主題、隊列，每一個消息都有它的topic，Kafka通過topic對消息進(jìn)行歸類。Kafka中可以將Topic從物理上劃分成一個或多個分區(qū)(Partition)，每個分區(qū)在物理上對應(yīng)一個文件夾，以”topicName_partitionIndex”的命名方式命名，該dir包含了這個分區(qū)的所有消息(.log)和索引文件(.index)，這使得Kafka的吞吐率可以水平擴(kuò)展。
Partition
每個分區(qū)都是一個 順序的、不可變的消息隊列， 并且可以持續(xù)的添加;分區(qū)中的消息都被分了一個序列號，稱之為偏移量(offset)，在每個分區(qū)中此偏移量都是唯一的。
producer在發(fā)布消息的時候，可以為每條消息指定Key，這樣消息被發(fā)送到broker時，會根據(jù)分區(qū)算法把消息存儲到對應(yīng)的分區(qū)中(一個分區(qū)存儲多個消息)，如果分區(qū)規(guī)則設(shè)置的合理，那么所有的消息將會被均勻的分布到不同的分區(qū)中，這樣就實現(xiàn)了負(fù)載均衡。
Broker
Kafka server，用來存儲消息，Kafka集群中的每一個服務(wù)器都是一個Broker，消費(fèi)者將從broker拉取訂閱的消息
Producer
向Kafka發(fā)送消息，生產(chǎn)者會根據(jù)topic分發(fā)消息。生產(chǎn)者也負(fù)責(zé)把消息關(guān)聯(lián)到Topic上的哪一個分區(qū)。最簡單的方式從分區(qū)列表中輪流選擇。也可以根據(jù)某種算法依照權(quán)重選擇分區(qū)。算法可由開發(fā)者定義。
Cousumer
Consermer實例可以是獨(dú)立的進(jìn)程，負(fù)責(zé)訂閱和消費(fèi)消息。消費(fèi)者用consumerGroup來標(biāo)識自己。同一個消費(fèi)組可以并發(fā)地消費(fèi)多個分區(qū)的消息，同一個partition也可以由多個consumerGroup并發(fā)消費(fèi)，但是在consumerGroup中一個partition只能由一個consumer消費(fèi)
CousumerGroup
Consumer Group：同一個Consumer Group中的Consumers，Kafka將相應(yīng)Topic中的每個消息只發(fā)送給其中一個Consumer
Kafka producer 設(shè)計原理
發(fā)送消息的流程
1.序列化消息&&.計算partition
根據(jù)key和value的配置對消息進(jìn)行序列化,然后計算partition：
ProducerRecord對象中如果指定了partition，就使用這個partition。否則根據(jù)key和topic的partition數(shù)目取余，如果key也沒有的話就隨機(jī)生成一個counter，使用這個counter來和partition數(shù)目取余。這個counter每次使用的時候遞增。
2發(fā)送到batch&&喚醒Sender 線程
根據(jù)topic-partition獲取對應(yīng)的batchs(Dueue
3.Sender把消息有序發(fā)到 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信息，同時根據(jù)metadata更新策略(定期更新metadata.max.age.ms、失效檢測，強(qiáng)制更新：檢查到metadata失效以后，調(diào)用metadata.requestUpdate()強(qiáng)制更新
- 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 冪等性發(fā)送
為實現(xiàn)Producer的冪等性，Kafka引入了Producer ID(即PID)和Sequence Number。對于每個PID，該P(yáng)roducer發(fā)送消息的每個
- 如果消息序號比Broker維護(hù)的序號差值比一大，說明中間有數(shù)據(jù)尚未寫入，即亂序，此時Broker拒絕該消息，Producer拋出InvalidSequenceNumber
- 如果消息序號小于等于Broker維護(hù)的序號，說明該消息已被保存，即為重復(fù)消息，Broker直接丟棄該消息，Producer拋出DuplicateSequenceNumber
- Sender發(fā)送失敗后會重試，這樣可以保證每個消息都被發(fā)送到broker
4. Sender處理broker發(fā)來的produce response
一旦broker處理完Sender的produce請求，就會發(fā)送produce response給Sender，此時producer將執(zhí)行我們?yōu)閟end()設(shè)置的回調(diào)函數(shù)。至此producer的send執(zhí)行完畢。
吞吐性&&延時：
- buffer.memory：buffer設(shè)置大了有助于提升吞吐性，但是batch太大會增大延遲，可搭配linger_ms參數(shù)使用
- linger_ms：如果batch太大，或者producer qps不高，batch添加的會很慢，我們可以強(qiáng)制在linger_ms時間后發(fā)送batch數(shù)據(jù)
- ack：producer收到多少broker的答復(fù)才算真的發(fā)送成功
- 0表示producer無需等待leader的確認(rèn)(吞吐最高、數(shù)據(jù)可靠性最差)
- 1代表需要leader確認(rèn)寫入它的本地log并立即確認(rèn)
- -1/all 代表所有的ISR都完成后確認(rèn)(吞吐最低、數(shù)據(jù)可靠性最高)
Sender線程和長連接
每初始化一個producer實例，都會初始化一個Sender實例，新增到broker的長連接。
代碼角度：每初始化一次KafkaProducer，都賦一個空的client
- public KafkaProducer(final Map configs) {
- this(configs, null, null, null, null, null, Time.SYSTEM);
- }
終端查看TCP連接數(shù)：
- lsof -p portNum -np | grep TCP，適當(dāng)增大producer數(shù)量能提升吞吐
Consumer設(shè)計原理
poll消息
- 消費(fèi)者通過fetch線程拉消息(單線程)
- 消費(fèi)者通過心跳線程來與broker發(fā)送心跳。超時會認(rèn)為掛掉
- 每個consumer group在broker上都有一個coordnator來管理，消費(fèi)者加入和退出，以及消費(fèi)消息的位移都由coordnator處理。
位移管理
consumer的消息位移代表了當(dāng)前group對topic-partition的消費(fèi)進(jìn)度，consumer宕機(jī)重啟后可以繼續(xù)從該offset開始消費(fèi)。在kafka0.8之前，位移信息存放在zookeeper上，由于zookeeper不適合高并發(fā)的讀寫，新版本Kafka把位移信息當(dāng)成消息，發(fā)往__consumers_offsets 這個topic所在的broker，__consumers_offsets默認(rèn)有50個分區(qū)。消息的key 是groupId+topic_partition,value 是offset.
Kafka Group 狀態(tài)
- Empty：初始狀態(tài)，Group 沒有任何成員，如果所有的 offsets 都過期的話就會變成 Dead
- PreparingRebalance：Group 正在準(zhǔn)備進(jìn)行 Rebalance
- AwaitingSync：Group 正在等待來 group leader 的 分配方案
- Stable：穩(wěn)定的狀態(tài)(Group is stable);
- Dead：Group 內(nèi)已經(jīng)沒有成員，并且它的 Metadata 已經(jīng)被移除
- 注意
重平衡reblance
當(dāng)一些原因?qū)е耤onsumer對partition消費(fèi)不再均勻時，kafka會自動執(zhí)行reblance，使得consumer對partition的消費(fèi)再次平衡。
什么時候發(fā)生rebalance?：
- 組訂閱topic數(shù)變更
- topic partition數(shù)變更
- consumer成員變更
- consumer 加入群組或者離開群組的時候
- consumer被檢測為崩潰的時候
reblance過程
舉例1 consumer被檢測為崩潰引起的reblance
比如心跳線程在timeout時間內(nèi)沒和broker發(fā)送心跳，此時coordnator認(rèn)為該group應(yīng)該進(jìn)行reblance。接下來其他consumer發(fā)來fetch請求后，coordnator將回復(fù)他們進(jìn)行reblance通知。當(dāng)consumer成員收到請求后，只有l(wèi)eader會根據(jù)分配策略進(jìn)行分配，然后把各自的分配結(jié)果返回給coordnator。這個時候只有consumer leader返回的是實質(zhì)數(shù)據(jù)，其他返回的都為空。收到分配方法后，consumer將會把分配策略同步給各consumer
舉例2 consumer加入引起的reblance
- 使用join協(xié)議，表示有consumer 要加入到group中
- 使用sync 協(xié)議，根據(jù)分配規(guī)則進(jìn)行分配
(上圖圖片摘自網(wǎng)絡(luò))
引申：以上reblance機(jī)制存在的問題
在大型系統(tǒng)中，一個topic可能對應(yīng)數(shù)百個consumer實例。這些consumer陸續(xù)加入到一個空消費(fèi)組將導(dǎo)致多次的rebalance;此外consumer 實例啟動的時間不可控，很有可能超出coordinator確定的rebalance timeout(即max.poll.interval.ms)，將會再次觸發(fā)rebalance，而每次rebalance的代價又相當(dāng)?shù)卮?，因為很多狀態(tài)都需要在rebalance前被持久化，而在rebalance后被重新初始化。
新版本改進(jìn)
通過延遲進(jìn)入PreparingRebalance狀態(tài)減少reblance次數(shù)
新版本新增了group.initial.rebalance.delay.ms參數(shù)。空消費(fèi)組接受到成員加入請求時，不立即轉(zhuǎn)化到PreparingRebalance狀態(tài)來開啟reblance。當(dāng)時間超過group.initial.rebalance.delay.ms后，再把group狀態(tài)改為PreparingRebalance(開啟reblance)。實現(xiàn)機(jī)制是在coordinator底層新增一個group狀態(tài)：InitialReblance。假設(shè)此時有多個consumer陸續(xù)啟動，那么group狀態(tài)先轉(zhuǎn)化為InitialReblance，待group.initial.rebalance.delay.ms時間后，再轉(zhuǎn)換為PreparingRebalance(開啟reblance)
Broker設(shè)計原理
Broker 是Kafka 集群中的節(jié)點。負(fù)責(zé)處理生產(chǎn)者發(fā)送過來的消息，消費(fèi)者消費(fèi)的請求。以及集群節(jié)點的管理等。由于涉及內(nèi)容較多，先簡單介紹，后續(xù)專門抽出一篇文章分享
broker zk注冊
broker消息存儲
- Kafka的消息以二進(jìn)制的方式緊湊地存儲，節(jié)省了很大空間
- 此外消息存在ByteBuffer而不是堆，這樣broker進(jìn)程掛掉時，數(shù)據(jù)不會丟失，同時避免了gc問題
- 通過零拷貝和順序?qū)ぶ?，讓消息存儲和讀取速度都非?？?/li>
- 處理fetch請求的時候通過zero-copy 加快速度
broker狀態(tài)數(shù)據(jù)
- broker設(shè)計中，每臺機(jī)器都保存了相同的狀態(tài)數(shù)據(jù)。主要包括以下：
- controller所在的broker ID，即保存了當(dāng)前集群中controller是哪臺broker
- 集群中所有broker的信息：比如每臺broker的ID、機(jī)架信息以及配置的若干組連接信息
- 集群中所有節(jié)點的信息：嚴(yán)格來說，它和上一個有些重復(fù)，不過此項是按照broker ID和監(jiān)聽器類型進(jìn)行分組的。對于超大集群來說，使用這一項緩存可以快速地定位和查找給定節(jié)點信息，而無需遍歷上一項中的內(nèi)容，算是一個優(yōu)化吧
- 集群中所有分區(qū)的信息：所謂分區(qū)信息指的是分區(qū)的leader、ISR和AR信息以及當(dāng)前處于offline狀態(tài)的副本集合。這部分?jǐn)?shù)據(jù)按照topic-partitionID進(jìn)行分組，可以快速地查找到每個分區(qū)的當(dāng)前狀態(tài)。(注：AR表示assigned replicas，即創(chuàng)建topic時為該分區(qū)分配的副本集合)
broker負(fù)載均衡
分區(qū)數(shù)量負(fù)載：各臺broker的partition數(shù)量應(yīng)該均勻
partition Replica分配算法如下：
- 將所有Broker(假設(shè)共n個Broker)和待分配的Partition排序
- 將第i個Partition分配到第(i mod n)個Broker上
- 將第i個Partition的第j個Replica分配到第((i + j) mod n)個Broker上
容量大小負(fù)載：每臺broker的硬盤占用大小應(yīng)該均勻
在kafka1.1之前，Kafka能夠保證各臺broker上partition數(shù)量均勻，但由于每個partition內(nèi)的消息數(shù)不同，可能存在不同硬盤之間內(nèi)存占用差異大的情況。在Kafka1.1中增加了副本跨路徑遷移功能kafka-reassign-partitions.sh，我們可以結(jié)合它和監(jiān)控系統(tǒng)，實現(xiàn)自動化的負(fù)載均衡
Kafka高可用
在介紹kafka高可用之前先介紹幾個概念
- 同步復(fù)制：要求所有能工作的Follower都復(fù)制完，這條消息才會被認(rèn)為commit，這種復(fù)制方式極大的影響了吞吐率
- 異步復(fù)制：Follower異步的從Leader pull數(shù)據(jù)，data只要被Leader寫入log認(rèn)為已經(jīng)commit，這種情況下如果Follower落后于Leader的比較多，如果Leader突然宕機(jī)，會丟失數(shù)據(jù)
Isr
Kafka結(jié)合同步復(fù)制和異步復(fù)制，使用ISR(與Partition Leader保持同步的Replica列表)的方式在確保數(shù)據(jù)不丟失和吞吐率之間做了平衡。Producer只需把消息發(fā)送到Partition Leader，Leader將消息寫入本地Log。Follower則從Leader pull數(shù)據(jù)。Follower在收到該消息向Leader發(fā)送ACK。一旦Leader收到了ISR中所有Replica的ACK，該消息就被認(rèn)為已經(jīng)commit了，Leader將增加HW并且向Producer發(fā)送ACK。這樣如果leader掛了，只要Isr中有一個replica存活，就不會丟數(shù)據(jù)。
Isr動態(tài)更新
Leader會跟蹤ISR，如果ISR中一個Follower宕機(jī)，或者落后太多，Leader將把它從ISR中移除。這里所描述的“落后太多”指Follower復(fù)制的消息落后于Leader后的條數(shù)超過預(yù)定值(replica.lag.max.messages)或者Follower超過一定時間(replica.lag.time.max.ms)未向Leader發(fā)送fetch請求。
broker Nodes In Zookeeper
/brokers/topics/[topic]/partitions/[partition]/state 保存了topic-partition的leader和Isr等信息
Controller負(fù)責(zé)broker故障檢查&&故障轉(zhuǎn)移(fail/recover)
- Controller在Zookeeper上注冊Watch，一旦有Broker宕機(jī)，其在Zookeeper對應(yīng)的znode會自動被刪除，Zookeeper會觸發(fā) Controller注冊的watch，Controller讀取最新的Broker信息
- Controller確定set_p，該集合包含了宕機(jī)的所有Broker上的所有Partition
- 對set_p中的每一個Partition，選舉出新的leader、Isr，并更新結(jié)果。
3.1 從/brokers/topics/[topic]/partitions/[partition]/state讀取該P(yáng)artition當(dāng)前的ISR
3.2 決定該P(yáng)artition的新Leader和Isr。如果當(dāng)前ISR中有至少一個Replica還幸存，則選擇其中一個作為新Leader，新的ISR則包含當(dāng)前ISR中所有幸存的Replica。否則選擇該P(yáng)artition中任意一個幸存的Replica作為新的Leader以及ISR(該場景下可能會有潛在的數(shù)據(jù)丟失)
3.3 更新Leader、ISR、leader_epoch、controller_epoch：寫入/brokers/topics/[topic]/partitions/[partition]/state
直接通過RPC向set_p相關(guān)的Broker發(fā)送LeaderAndISRRequest命令。Controller可以在一個RPC操作中發(fā)送多個命令從而提高效率。
Controller掛掉
每個 broker 都會在 zookeeper 的臨時節(jié)點 "/controller" 注冊 watcher，當(dāng) controller 宕機(jī)時 "/controller" 會消失，觸發(fā)broker的watch，每個 broker 都嘗試創(chuàng)建新的 controller path，只有一個競選成功并當(dāng)選為 controller。
使用Kafka如何保證冪等性
不丟消息
- 首先kafka保證了對已提交消息的at least保證
- Sender有重試機(jī)制
- producer業(yè)務(wù)方在使用producer發(fā)送消息時，注冊回調(diào)函數(shù)。在onError方法中重發(fā)消息
- consumer 拉取到消息后，處理完畢再commit，保證commit的消息一定被處理完畢
不重復(fù)
consumer拉取到消息先保存，commit成功后刪除緩存數(shù)據(jù)
Kafka高性能
- partition提升了并發(fā)
- zero-copy
- 順序?qū)懭?/li>
- 消息聚集batch
- 頁緩存
業(yè)務(wù)方對 Kafka producer的優(yōu)化
- 增大producer數(shù)量
- ack配置
- batch
