震惊了！原来这才是 Kafka！（多图+深入）
簡(jiǎn)介
kafka是一個(gè)分布式消息隊(duì)列。具有高性能、持久化、多副本備份、橫向擴(kuò)展能力。生產(chǎn)者往隊(duì)列里寫消息，消費(fèi)者從隊(duì)列里取消息進(jìn)行業(yè)務(wù)邏輯。一般在架構(gòu)設(shè)計(jì)中起到解耦、削峰、異步處理的作用。
kafka對(duì)外使用topic的概念，生產(chǎn)者往topic里寫消息，消費(fèi)者從讀消息。為了做到水平擴(kuò)展，一個(gè)topic實(shí)際是由多個(gè)partition組成的，遇到瓶頸時(shí)，可以通過增加partition的數(shù)量來進(jìn)行橫向擴(kuò)容。單個(gè)parition內(nèi)是保證消息有序。
每新寫一條消息，kafka就是在對(duì)應(yīng)的文件append寫，所以性能非常高。
kafka的總體數(shù)據(jù)流是這樣的：
kafka data flow大概用法就是，Producers往Brokers里面的指定Topic中寫消息，Consumers從Brokers里面拉去指定Topic的消息，然后進(jìn)行業(yè)務(wù)處理。
圖中有兩個(gè)topic，topic 0有兩個(gè)partition，topic 1有一個(gè)partition，三副本備份。可以看到consumer gourp 1中的consumer 2沒有分到partition處理，這是有可能出現(xiàn)的，下面會(huì)講到。
關(guān)于broker、topics、partitions的一些元信息用zk來存，監(jiān)控和路由啥的也都會(huì)用到zk。
生產(chǎn)
基本流程是這樣的：
kafka sdk product flow.png創(chuàng)建一條記錄，記錄中一個(gè)要指定對(duì)應(yīng)的topic和value，key和partition可選。 先序列化，然后按照topic和partition，放進(jìn)對(duì)應(yīng)的發(fā)送隊(duì)列中。kafka produce都是批量請(qǐng)求，會(huì)積攢一批，然后一起發(fā)送，不是調(diào)send()就進(jìn)行立刻進(jìn)行網(wǎng)絡(luò)發(fā)包。
如果partition沒填，那么情況會(huì)是這樣的：
key有填
按照key進(jìn)行哈希，相同key去一個(gè)partition。（如果擴(kuò)展了partition的數(shù)量那么就不能保證了）
key沒填
round-robin來選partition
這些要發(fā)往同一個(gè)partition的請(qǐng)求按照配置，攢一波，然后由一個(gè)單獨(dú)的線程一次性發(fā)過去。
API
有high level api，替我們把很多事情都干了，offset，路由啥都替我們干了，用以來很簡(jiǎn)單。
還有simple api，offset啥的都是要我們自己記錄。
partition
當(dāng)存在多副本的情況下，會(huì)盡量把多個(gè)副本，分配到不同的broker上。kafka會(huì)為partition選出一個(gè)leader，之后所有該partition的請(qǐng)求，實(shí)際操作的都是leader，然后再同步到其他的follower。當(dāng)一個(gè)broker歇菜后，所有l(wèi)eader在該broker上的partition都會(huì)重新選舉，選出一個(gè)leader。（這里不像分布式文件存儲(chǔ)系統(tǒng)那樣會(huì)自動(dòng)進(jìn)行復(fù)制保持副本數(shù)）
然后這里就涉及兩個(gè)細(xì)節(jié)：怎么分配partition，怎么選leader。
關(guān)于partition的分配，還有l(wèi)eader的選舉，總得有個(gè)執(zhí)行者。在kafka中，這個(gè)執(zhí)行者就叫controller。kafka使用zk在broker中選出一個(gè)controller，用于partition分配和leader選舉。
partition的分配
將所有Broker（假設(shè)共n個(gè)Broker）和待分配的Partition排序
將第i個(gè)Partition分配到第（i mod n）個(gè)Broker上 （這個(gè)就是leader）
將第i個(gè)Partition的第j個(gè)Replica分配到第（(i + j) mode n）個(gè)Broker上
leader容災(zāi)
controller會(huì)在Zookeeper的/brokers/ids節(jié)點(diǎn)上注冊(cè)Watch，一旦有broker宕機(jī)，它就能知道。當(dāng)broker宕機(jī)后，controller就會(huì)給受到影響的partition選出新leader。controller從zk的/brokers/topics/[topic]/partitions/[partition]/state中，讀取對(duì)應(yīng)partition的ISR（in-sync replica已同步的副本）列表，選一個(gè)出來做leader。
選出leader后，更新zk，然后發(fā)送LeaderAndISRRequest給受影響的broker，讓它們改變知道這事。為什么這里不是使用zk通知，而是直接給broker發(fā)送rpc請(qǐng)求，我的理解可能是這樣做zk有性能問題吧。
如果ISR列表是空，那么會(huì)根據(jù)配置，隨便選一個(gè)replica做leader，或者干脆這個(gè)partition就是歇菜。如果ISR列表的有機(jī)器，但是也歇菜了，那么還可以等ISR的機(jī)器活過來。
多副本同步
這里的策略，服務(wù)端這邊的處理是follower從leader批量拉取數(shù)據(jù)來同步。但是具體的可靠性，是由生產(chǎn)者來決定的。
生產(chǎn)者生產(chǎn)消息的時(shí)候，通過request.required.acks參數(shù)來設(shè)置數(shù)據(jù)的可靠性。
| 0 | which means that the producer never waits for an acknowledgement from the broker.發(fā)過去就完事了，不關(guān)心broker是否處理成功，可能丟數(shù)據(jù)。 |
| 1 | which means that the producer gets an acknowledgement after the leader replica has received the data. 當(dāng)寫Leader成功后就返回,其他的replica都是通過fetcher去同步的,所以kafka是異步寫，主備切換可能丟數(shù)據(jù)。 |
| -1 | which means that the producer gets an acknowledgement after all in-sync replicas have received the data. 要等到isr里所有機(jī)器同步成功，才能返回成功，延時(shí)取決于最慢的機(jī)器。強(qiáng)一致，不會(huì)丟數(shù)據(jù)。 |
在acks=-1的時(shí)候，如果ISR少于min.insync.replicas指定的數(shù)目，那么就會(huì)返回不可用。
這里ISR列表中的機(jī)器是會(huì)變化的，根據(jù)配置replica.lag.time.max.ms，多久沒同步，就會(huì)從ISR列表中剔除。以前還有根據(jù)落后多少條消息就踢出ISR，在1.0版本后就去掉了，因?yàn)檫@個(gè)值很難取，在高峰的時(shí)候很容易出現(xiàn)節(jié)點(diǎn)不斷的進(jìn)出ISR列表。
從ISA中選出leader后，follower會(huì)從把自己日志中上一個(gè)高水位后面的記錄去掉，然后去和leader拿新的數(shù)據(jù)。因?yàn)樾碌膌eader選出來后，follower上面的數(shù)據(jù)，可能比新leader多，所以要截取。這里高水位的意思，對(duì)于partition和leader，就是所有ISR中都有的最新一條記錄。消費(fèi)者最多只能讀到高水位；
從leader的角度來說高水位的更新會(huì)延遲一輪，例如寫入了一條新消息，ISR中的broker都fetch到了，但是ISR中的broker只有在下一輪的fetch中才能告訴leader。
也正是由于這個(gè)高水位延遲一輪，在一些情況下，kafka會(huì)出現(xiàn)丟數(shù)據(jù)和主備數(shù)據(jù)不一致的情況，0.11開始，使用leader epoch來代替高水位。（https://cwiki.apache.org/confluence/display/KAFKA/KIP-101+-+Alter+Replication+Protocol+to+use+Leader+Epoch+rather+than+High+Watermark+for+Truncation#KIP-101-AlterReplicationProtocoltouseLeaderEpochratherthanHighWatermarkforTruncation-Scenario1:HighWatermarkTruncationfollowedbyImmediateLeaderElection）
思考：
當(dāng)acks=-1時(shí)
是follwers都來fetch就返回成功，還是等follwers第二輪fetch？
leader已經(jīng)寫入本地，但是ISR中有些機(jī)器失敗，那么怎么處理呢？
消費(fèi)
訂閱topic是以一個(gè)消費(fèi)組來訂閱的，一個(gè)消費(fèi)組里面可以有多個(gè)消費(fèi)者。同一個(gè)消費(fèi)組中的兩個(gè)消費(fèi)者，不會(huì)同時(shí)消費(fèi)一個(gè)partition。換句話來說，就是一個(gè)partition，只能被消費(fèi)組里的一個(gè)消費(fèi)者消費(fèi)，但是可以同時(shí)被多個(gè)消費(fèi)組消費(fèi)。因此，如果消費(fèi)組內(nèi)的消費(fèi)者如果比partition多的話，那么就會(huì)有個(gè)別消費(fèi)者一直空閑。
API
訂閱topic時(shí)，可以用正則表達(dá)式，如果有新topic匹配上，那能自動(dòng)訂閱上。
offset的保存
一個(gè)消費(fèi)組消費(fèi)partition，需要保存offset記錄消費(fèi)到哪，以前保存在zk中，由于zk的寫性能不好，以前的解決方法都是consumer每隔一分鐘上報(bào)一次。這里zk的性能嚴(yán)重影響了消費(fèi)的速度，而且很容易出現(xiàn)重復(fù)消費(fèi)。
在0.10版本后，kafka把這個(gè)offset的保存，從zk總剝離，保存在一個(gè)名叫__consumeroffsets topic的topic中。寫進(jìn)消息的key由groupid、topic、partition組成，value是偏移量offset。topic配置的清理策略是compact?？偸潜Ａ糇钚碌膋ey，其余刪掉。一般情況下，每個(gè)key的offset都是緩存在內(nèi)存中，查詢的時(shí)候不用遍歷partition，如果沒有緩存，第一次就會(huì)遍歷partition建立緩存，然后查詢返回。
確定consumer group位移信息寫入__consumers_offsets的哪個(gè)partition，具體計(jì)算公式：
__consumers_offsets?partition?=Math.abs(groupId.hashCode()?%?groupMetadataTopicPartitionCount)??? //groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定，默認(rèn)是50個(gè)分區(qū)。思考：
如果正在跑的服務(wù)，修改了offsets.topic.num.partitions，那么offset的保存是不是就亂套了？
分配partition--reblance
生產(chǎn)過程中broker要分配partition，消費(fèi)過程這里，也要分配partition給消費(fèi)者。類似broker中選了一個(gè)controller出來，消費(fèi)也要從broker中選一個(gè)coordinator，用于分配partition。
下面從頂向下，分別闡述一下
怎么選coordinator。
交互流程。
reblance的流程。
選coordinator
看offset保存在那個(gè)partition
該partition leader所在的broker就是被選定的coordinator
這里我們可以看到，consumer group的coordinator，和保存consumer group offset的partition leader是同一臺(tái)機(jī)器。
交互流程
把coordinator選出來之后，就是要分配了
整個(gè)流程是這樣的：
consumer啟動(dòng)、或者coordinator宕機(jī)了，consumer會(huì)任意請(qǐng)求一個(gè)broker，發(fā)送ConsumerMetadataRequest請(qǐng)求，broker會(huì)按照上面說的方法，選出這個(gè)consumer對(duì)應(yīng)coordinator的地址。
consumer 發(fā)送heartbeat請(qǐng)求給coordinator，返回IllegalGeneration的話，就說明consumer的信息是舊的了，需要重新加入進(jìn)來，進(jìn)行reblance。返回成功，那么consumer就從上次分配的partition中繼續(xù)執(zhí)行。
reblance流程
consumer給coordinator發(fā)送JoinGroupRequest請(qǐng)求。
這時(shí)其他consumer發(fā)heartbeat請(qǐng)求過來時(shí)，coordinator會(huì)告訴他們，要reblance了。
其他consumer發(fā)送JoinGroupRequest請(qǐng)求。
所有記錄在冊(cè)的consumer都發(fā)了JoinGroupRequest請(qǐng)求之后，coordinator就會(huì)在這里consumer中隨便選一個(gè)leader。然后回JoinGroupRespone，這會(huì)告訴consumer你是follower還是leader，對(duì)于leader，還會(huì)把follower的信息帶給它，讓它根據(jù)這些信息去分配partition
當(dāng)partition或者消費(fèi)者的數(shù)量發(fā)生變化時(shí)，都得進(jìn)行reblance。
列舉一下會(huì)reblance的情況：
增加partition
增加消費(fèi)者
消費(fèi)者主動(dòng)關(guān)閉
消費(fèi)者宕機(jī)了
coordinator自己也宕機(jī)了
消息投遞語義
kafka支持3種消息投遞語義
At most once：最多一次，消息可能會(huì)丟失，但不會(huì)重復(fù)
At least once：最少一次，消息不會(huì)丟失，可能會(huì)重復(fù)
Exactly once：只且一次，消息不丟失不重復(fù)，只且消費(fèi)一次（0.11中實(shí)現(xiàn)，僅限于下游也是kafka）
在業(yè)務(wù)中，常常都是使用At least once的模型，如果需要可重入的話，往往是業(yè)務(wù)自己實(shí)現(xiàn)。
At least once
先獲取數(shù)據(jù)，再進(jìn)行業(yè)務(wù)處理，業(yè)務(wù)處理成功后commit offset。
1、生產(chǎn)者生產(chǎn)消息異常，消息是否成功寫入不確定，重做，可能寫入重復(fù)的消息
2、消費(fèi)者處理消息，業(yè)務(wù)處理成功后，更新offset失敗，消費(fèi)者重啟的話，會(huì)重復(fù)消費(fèi)
At most once
先獲取數(shù)據(jù)，再commit offset，最后進(jìn)行業(yè)務(wù)處理。
Exactly once
思路是這樣的，首先要保證消息不丟，再去保證不重復(fù)。所以盯著At least once的原因來搞。 首先想出來的：
生產(chǎn)者重做導(dǎo)致重復(fù)寫入消息----生產(chǎn)保證冪等性
消費(fèi)者重復(fù)消費(fèi)---消滅重復(fù)消費(fèi)，或者業(yè)務(wù)接口保證冪等性重復(fù)消費(fèi)也沒問題
由于業(yè)務(wù)接口是否冪等，不是kafka能保證的，所以kafka這里提供的exactly once是有限制的，消費(fèi)者的下游也必須是kafka。所以一下討論的，沒特殊說明，消費(fèi)者的下游系統(tǒng)都是kafka（注:使用kafka conector，它對(duì)部分系統(tǒng)做了適配，實(shí)現(xiàn)了exactly once）。
生產(chǎn)者冪等性好做，沒啥問題。
解決重復(fù)消費(fèi)有兩個(gè)方法：
下游系統(tǒng)保證冪等性，重復(fù)消費(fèi)也不會(huì)導(dǎo)致多條記錄。
把commit offset和業(yè)務(wù)處理綁定成一個(gè)事務(wù)。
本來exactly once實(shí)現(xiàn)第1點(diǎn)就ok了。
但是在一些使用場(chǎng)景下，我們的數(shù)據(jù)源可能是多個(gè)topic，處理后輸出到多個(gè)topic，這時(shí)我們會(huì)希望輸出時(shí)要么全部成功，要么全部失敗。這就需要實(shí)現(xiàn)事務(wù)性。既然要做事務(wù)，那么干脆把重復(fù)消費(fèi)的問題從根源上解決，把commit offset和輸出到其他topic綁定成一個(gè)事務(wù)。
生產(chǎn)冪等性
思路是這樣的，為每個(gè)producer分配一個(gè)pid，作為該producer的唯一標(biāo)識(shí)。producer會(huì)為每一個(gè)維護(hù)一個(gè)單調(diào)遞增的seq。類似的，broker也會(huì)為每個(gè)記錄下最新的seq。當(dāng)req_seq == broker_seq+1時(shí)，broker才會(huì)接受該消息。因?yàn)?#xff1a;
消息的seq比broker的seq大超過時(shí)，說明中間有數(shù)據(jù)還沒寫入，即亂序了。
消息的seq不比broker的seq小，那么說明該消息已被保存。
事務(wù)性/原子性廣播
場(chǎng)景是這樣的：
先從多個(gè)源topic中獲取數(shù)據(jù)。
做業(yè)務(wù)處理，寫到下游的多個(gè)目的topic。
更新多個(gè)源topic的offset。
其中第2、3點(diǎn)作為一個(gè)事務(wù)，要么全成功，要么全失敗。這里得益與offset實(shí)際上是用特殊的topic去保存，這兩點(diǎn)都?xì)w一為寫多個(gè)topic的事務(wù)性處理。
基本思路是這樣的：
引入tid（transaction id），和pid不同，這個(gè)id是應(yīng)用程序提供的，用于標(biāo)識(shí)事務(wù)，和producer是誰并沒關(guān)系。就是任何producer都可以使用這個(gè)tid去做事務(wù)，這樣進(jìn)行到一半就死掉的事務(wù)，可以由另一個(gè)producer去恢復(fù)。
同時(shí)為了記錄事務(wù)的狀態(tài)，類似對(duì)offset的處理，引入transaction coordinator用于記錄transaction log。在集群中會(huì)有多個(gè)transaction coordinator，每個(gè)tid對(duì)應(yīng)唯一一個(gè)transaction coordinator。
注：transaction log刪除策略是compact，已完成的事務(wù)會(huì)標(biāo)記成null，compact后不保留。
做事務(wù)時(shí)，先標(biāo)記開啟事務(wù)，寫入數(shù)據(jù)，全部成功就在transaction log中記錄為prepare commit狀態(tài)，否則寫入prepare abort的狀態(tài)。之后再去給每個(gè)相關(guān)的partition寫入一條marker（commit或者abort）消息，標(biāo)記這個(gè)事務(wù)的message可以被讀取或已經(jīng)廢棄。成功后在transaction log記錄下commit/abort狀態(tài)，至此事務(wù)結(jié)束。
數(shù)據(jù)流：
Kafka Transactions Data Flow首先使用tid請(qǐng)求任意一個(gè)broker（代碼中寫的是負(fù)載最小的broker），找到對(duì)應(yīng)的transaction coordinator。
請(qǐng)求transaction coordinator獲取到對(duì)應(yīng)的pid，和pid對(duì)應(yīng)的epoch，這個(gè)epoch用于防止僵死進(jìn)程復(fù)活導(dǎo)致消息錯(cuò)亂，當(dāng)消息的epoch比當(dāng)前維護(hù)的epoch小時(shí)，拒絕掉。tid和pid有一一對(duì)應(yīng)的關(guān)系，這樣對(duì)于同一個(gè)tid會(huì)返回相同的pid。
client先請(qǐng)求transaction coordinator記錄的事務(wù)狀態(tài)，初始狀態(tài)是BEGIN，如果是該事務(wù)中第一個(gè)到達(dá)的，同時(shí)會(huì)對(duì)事務(wù)進(jìn)行計(jì)時(shí)；client輸出數(shù)據(jù)到相關(guān)的partition中；client再請(qǐng)求transaction coordinator記錄offset的事務(wù)狀態(tài)；client發(fā)送offset commit到對(duì)應(yīng)offset partition。
client發(fā)送commit請(qǐng)求，transaction coordinator記錄prepare commit/abort，然后發(fā)送marker給相關(guān)的partition。全部成功后，記錄commit/abort的狀態(tài)，最后這個(gè)記錄不需要等待其他replica的ack，因?yàn)閜repare不丟就能保證最終的正確性了。
這里prepare的狀態(tài)主要是用于事務(wù)恢復(fù)，例如給相關(guān)的partition發(fā)送控制消息，沒發(fā)完就宕機(jī)了，備機(jī)起來后，producer發(fā)送請(qǐng)求獲取pid時(shí)，會(huì)把未完成的事務(wù)接著完成。
當(dāng)partition中寫入commit的marker后，相關(guān)的消息就可被讀取。所以kafka事務(wù)在prepare commit到commit這個(gè)時(shí)間段內(nèi)，消息是逐漸可見的，而不是同一時(shí)刻可見。
詳細(xì)細(xì)節(jié)可看：https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-TransactionalGuarantees
消費(fèi)事務(wù)
前面都是從生產(chǎn)的角度看待事務(wù)。還需要從消費(fèi)的角度去考慮一些問題。
消費(fèi)時(shí)，partition中會(huì)存在一些消息處于未commit狀態(tài)，即業(yè)務(wù)方應(yīng)該看不到的消息，需要過濾這些消息不讓業(yè)務(wù)看到，kafka選擇在消費(fèi)者進(jìn)程中進(jìn)行過來，而不是在broker中過濾，主要考慮的還是性能。kafka高性能的一個(gè)關(guān)鍵點(diǎn)是zero copy，如果需要在broker中過濾，那么勢(shì)必需要讀取消息內(nèi)容到內(nèi)存，就會(huì)失去zero copy的特性。
文件組織
kafka的數(shù)據(jù)，實(shí)際上是以文件的形式存儲(chǔ)在文件系統(tǒng)的。topic下有partition，partition下有segment，segment是實(shí)際的一個(gè)個(gè)文件，topic和partition都是抽象概念。
在目錄/
不能識(shí)別此Latex公式: {topicName}-{partitionid}/下，存儲(chǔ)著實(shí)際的log文件（即segment），還有對(duì)應(yīng)的索引文件。
每個(gè)segment文件大小相等，文件名以這個(gè)segment中最小的offset命名，文件擴(kuò)展名是.log；segment對(duì)應(yīng)的索引的文件名字一樣，擴(kuò)展名是.index。有兩個(gè)index文件，一個(gè)是offset index用于按offset去查message，一個(gè)是time index用于按照時(shí)間去查，其實(shí)這里可以優(yōu)化合到一起，下面只說offset index。總體的組織是這樣的：
kafka 文件組織為了減少索引文件的大小，降低空間使用，方便直接加載進(jìn)內(nèi)存中，這里的索引使用稀疏矩陣，不會(huì)每一個(gè)message都記錄下具體位置，而是每隔一定的字節(jié)數(shù)，再建立一條索引。 索引包含兩部分，分別是baseOffset，還有position。
baseOffset：意思是這條索引對(duì)應(yīng)segment文件中的第幾條message。這樣做方便使用數(shù)值壓縮算法來節(jié)省空間。例如kafka使用的是varint。
position：在segment中的絕對(duì)位置。
查找offset對(duì)應(yīng)的記錄時(shí)，會(huì)先用二分法，找出對(duì)應(yīng)的offset在哪個(gè)segment中，然后使用索引，在定位出offset在segment中的大概位置，再遍歷查找message。
常用配置項(xiàng)
broker配置
| broker.id | broker的唯一標(biāo)識(shí) |
| auto.create.topics.auto | 設(shè)置成true，就是遇到?jīng)]有的topic自動(dòng)創(chuàng)建topic。 |
| log.dirs | log的目錄數(shù)，目錄里面放partition，當(dāng)生成新的partition時(shí)，會(huì)挑目錄里partition數(shù)最少的目錄放。 |
topic配置
| num.partitions | 新建一個(gè)topic，會(huì)有幾個(gè)partition。 |
| log.retention.ms | 對(duì)應(yīng)的還有minutes，hours的單位。日志保留時(shí)間，因?yàn)閯h除是文件維度而不是消息維度，看的是日志文件的mtime。 |
| log.retention.bytes | partion最大的容量，超過就清理老的。注意這個(gè)是partion維度，就是說如果你的topic有8個(gè)partition，配置1G，那么平均分配下，topic理論最大值8G。 |
| log.segment.bytes | 一個(gè)segment的大小。超過了就滾動(dòng)。 |
| log.segment.ms | 一個(gè)segment的打開時(shí)間，超過了就滾動(dòng)。 |
| message.max.bytes | message最大多大 |
關(guān)于日志清理，默認(rèn)當(dāng)前正在寫的日志，是怎么也不會(huì)清理掉的。
還有0.10之前的版本，時(shí)間看的是日志文件的mtime，但這個(gè)指是不準(zhǔn)確的，有可能文件被touch一下，mtime就變了。因此在0.10版本開始，改為使用該文件最新一條消息的時(shí)間來判斷。
按大小清理這里也要注意，Kafka在定時(shí)任務(wù)中嘗試比較當(dāng)前日志量總大小是否超過閾值至少一個(gè)日志段的大小。如果超過但是沒超過一個(gè)日志段，那么就不會(huì)刪除。
總結(jié)
以上是生活随笔為你收集整理的震惊了！原来这才是 Kafka！（多图+深入）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 并不是所有的 Github 项目写在简历
- 下一篇： Java 并发框架全览，这个牛逼！