Kafka-設(shè)計架構(gòu)原理詳細(xì)解析

Kafka設(shè)計架構(gòu)原理詳細(xì)解析

什么是Kafka？ApacheKafka是一個開放源代碼的分布式事件流平臺，成千上萬的公司使用它來實現(xiàn)高性

能數(shù)據(jù)管道，流分析，數(shù)據(jù)集成和關(guān)鍵任務(wù)等相關(guān)的應(yīng)用程序。Kafka的應(yīng)用場景構(gòu)造實時流數(shù)據(jù)管道，它可以在系統(tǒng)或應(yīng)用之間可靠地獲取數(shù)據(jù)(相當(dāng)于messagequeue)，特別是在集群情況下，多個服務(wù)器需要建立交流構(gòu)建實時流式應(yīng)用程序，對這些流數(shù)據(jù)進(jìn)行轉(zhuǎn)換或者影響。(就是流處理，通過kafkastreamtopic和topic之間內(nèi)部進(jìn)行變化)Kafka架構(gòu)設(shè)計

Producer：生產(chǎn)者可以將數(shù)據(jù)發(fā)布到所選擇的topic（主題）中。生成者負(fù)責(zé)將記錄分配到topic的哪一個分區(qū)（partition）中，這里可以使用對多個partition循環(huán)發(fā)送來實現(xiàn)多個server負(fù)載均衡Broker：日志的分區(qū)（partition）分布在Kafka集群的服務(wù)器上。每個服務(wù)器處理數(shù)據(jù)和請求時，共享這些分區(qū)。每一個分區(qū)都會在以配置的服務(wù)器上進(jìn)行備份，確保容錯性。

其中，每個分區(qū)都有一臺server作為leader，零臺或墮胎server作為follows。leaderserver處理一切對分區(qū)的讀寫請求，而follwers只需被動的同步leader上的數(shù)據(jù)。當(dāng)leader宕機(jī)了，followers中的一臺server會自動成為新的eader，每臺server都會成為某些分區(qū)的leader和某些分區(qū)的follower，因此集群的負(fù)載是均衡的Consumer：消費(fèi)者使用一個group（消費(fèi)組）名稱來表示，發(fā)布到topic中的每條記錄將被分配到訂閱消費(fèi)組中的其中一個消費(fèi)者示例。消費(fèi)者實例可以分布在多個進(jìn)程中或多個機(jī)器上

這里有兩個注意的地方：如果所有的消費(fèi)者實例在同一個消費(fèi)組中，消息記錄會負(fù)載均衡到消費(fèi)組中的每一個消費(fèi)者實例如果所有的消費(fèi)者實例在不同的消費(fèi)組中，則會將每條消息記錄廣播到所有的消費(fèi)組或消費(fèi)者進(jìn)程中

如圖中所示，這個Kafka集群中有兩臺server，四個分區(qū)（p0-p3）和兩個消費(fèi)組。這時分區(qū)中的消息記錄會廣播到所有的消費(fèi)者組中Kafka生產(chǎn)者架構(gòu)

基本流程：主線程Producer中會經(jīng)過攔截器、序列化器、分區(qū)器，然后將處理好的消息發(fā)送到消息累加器中消息累加器每個分區(qū)會對應(yīng)一個隊列，在收到消息后，將消息放到隊列中使用ProducerBatch批量的進(jìn)行消息發(fā)送到Sender線程處理（這里為了提高發(fā)送效率，減少帶寬），ProducerBatch中就是我們需要發(fā)送的消息，其中消息累加器中可以使用Buffer.memory配置，默認(rèn)為32MBSender線程會從隊列的隊頭部開始讀取消息，然后創(chuàng)建request后會經(jīng)過會被緩存，然后提交到Selector，Selector發(fā)送消息到Kafka集群對于一些還沒收到Kafka集群ack響應(yīng)的消息，會將未響應(yīng)接收消息的請求進(jìn)行緩存，當(dāng)收到Kafka集群ack響應(yīng)后，會將request請求在緩存中清除并同時移除消息累加器中的消息Kafka消費(fèi)者架構(gòu)

基本流程：

ConsumerGroup中的Consumer向各自注冊的分區(qū)上進(jìn)行消費(fèi)消息

Consumer消費(fèi)消息后會將當(dāng)前標(biāo)注的消費(fèi)位移信息以消息的方式提交到位移主題中記錄，一個ConsumerGroup中多個Consumer會做負(fù)載均衡，如果一個Consumer宕機(jī)，會自動切換到組內(nèi)別的Consumer進(jìn)行消費(fèi)關(guān)鍵的點(diǎn)：

ConsumerGroup：組內(nèi)多個的Consumer可以公用一個ConsumerId，組內(nèi)所有的Consumer只能注冊到一個分區(qū)上去消費(fèi)，一個ConsumerGroup只能到一個Topic上去消費(fèi)位移主題：位移主題的主要作用是保存Kafka消費(fèi)者的位移信息Kafka老版本之前:

在Kafka老版本之前處理方式是自動或手動地將位移數(shù)據(jù)提交到Zookeeper進(jìn)行保存，Consumer重啟后，自動從Zookeeper中讀取消費(fèi)位移信息，從而在上次的offset地方繼續(xù)消費(fèi)

優(yōu)點(diǎn)：KafkaBroker中不需要保存位移數(shù)據(jù)，減少了Broker端需要持有的狀態(tài)信息，有利于動態(tài)擴(kuò)展

缺點(diǎn)：每一個Consumer消費(fèi)后需要發(fā)送位移信息到Zookeeper，而Zooker不適用于這種高頻的寫操作Kafka最新版本中位移主題的處理方式：

Consumer的位移信息offset會當(dāng)作一條條普通消息提交到位移主題（_consumer_offsets）中。Kafka文件存儲架構(gòu)

window文件系統(tǒng)中的文件列表：

這里比較好理解：一個Topic分別存儲在不同的partition中一個partitioin對應(yīng)著多個replica備份一個relica對應(yīng)著一個Log一個Log對應(yīng)多個LogSegment而在LogSegment中存儲著log文件、索引文件、其它文件Kafka如何保證數(shù)據(jù)有序性？一些場景需要保證多個消息的消費(fèi)順序，比如訂單，但在kafka中一個消息可能被發(fā)到多個partition中多個線程處理，被多個消費(fèi)者消費(fèi)，無法保證消息的消費(fèi)順序解決方案：將需要順序消費(fèi)的消息發(fā)送的時候設(shè)置將某個topic發(fā)送到指定的partition（也可以根據(jù)key的hash與分區(qū)進(jìn)行運(yùn)算），則在partition中的消息也是有序的，消費(fèi)的時候?qū)⒁唤M同hash的key放到同一個queue中保證同一個消費(fèi)者下的同一個線程對此queue進(jìn)行消費(fèi)?？偨Y(jié)：一個producer->一個partition->一個queue->一個comsumer->一個線程

當(dāng)對于需要順序消費(fèi)的消息數(shù)量大的時候，無法保證吞吐量Kafka如何保證數(shù)據(jù)可靠性？AR（AssignedReplicas）：分區(qū)中的所有副本統(tǒng)稱為AR。所有消息會先發(fā)送到leader副本，然后follower副本才能從leader中拉取消息進(jìn)行同步。但是在同步期間，follower對于leader而言會有一定程度的滯后，這個時候follower和leader并非完全同步狀態(tài)OSR（OutSyncReplicas）：follower副本與leader副本沒有完全同步或滯后的副本集合ISR（InSyncReplicas）：AR中的一個子集，ISR中的副本都是與leader保持完全同步的副本，如果某個在ISR中的follower副本落后于leader副本太多，則會被從ISR中移除，否則如果完全同步，會從OSR中移至ISR集合。在默認(rèn)情況下，當(dāng)leader副本發(fā)生故障時，只有在ISR集合中的follower副本才有資格被選舉為新leader，而OSR中的副本沒有機(jī)會（可以通過unclean.leader.election.enable進(jìn)行配置）HW（HighWatermark）：高水位，它標(biāo)識了一個特定的消息偏移量（offset），消費(fèi)者只能拉取到這個水位offset之前的消息LEO（LogEndOffset）：標(biāo)識當(dāng)前日志文件中下一條待寫入的消息的offset。在ISR集合中的每個副本都會維護(hù)自身的LEO，且HW==LEO。

圖中，HW就是8，Consumer只能拉去0~7的消息，LEO就是15，代表消息還沒有同步到follower下面通過一個例子來說明下ISR、HW、LEO之間的關(guān)系：

假設(shè)由一個leader副本，它有兩個follower副本，這時候producer向leader寫入3、4兩條消息，我們來觀察下他們是如何同步的

這個時候?qū)懭雰蓷l消息到leader，這個時候LEO變?yōu)?，然后follower開始同步leader數(shù)據(jù)

由于網(wǎng)絡(luò)或其它原因，follower2同步效率較低，還沒有完成同步，這個時候HW的offset為4，在此offset之前的消息Consumer都可見

在一定的延遲后，follower2也完成了隊leader副本的同步，這時HW為5，LEO為5，且兩個follower副本都在ISR集合中，在leader或follower宕機(jī)后，會在ISR集合的副本中選舉一個來當(dāng)新的leader副本HW高水位的弊端：高水位更新需要一輪額外的拉取請求leader和follower之間同步會有時間差，可能導(dǎo)致數(shù)據(jù)不一致或數(shù)據(jù)丟失

接下來通過一個例子來進(jìn)行詳細(xì)說明消息1消息丟失的過程（min.insync.replicas=1）：

對于消息不一致的情況：

就是leader、follower同時宕機(jī)，然后由follower先恢復(fù)且寫入消息1，HW=1，leader恢復(fù)啟之后發(fā)現(xiàn)HW相等，則不進(jìn)行同步，但實際上他們的消息1不是同一個消息，導(dǎo)致消息不一致在kafka版本中引入LeaderEpoch來解決使用高水位導(dǎo)致的數(shù)據(jù)丟失和數(shù)據(jù)不一致的問題

所謂leaderepoch實際上是一對值：（epoch,offset），epoch標(biāo)識leader的版本號，從0開始，每變更一次leader，epoch+1；而offset對應(yīng)于該epoch版本的leader寫入第一條消息（成為leader后的首條消息）的位移

（0，0）、（1，120）表示第一個leader從位移0開始寫入消息，共寫了120條，第二個leader版本號為1，從位移120處開始寫入消息規(guī)避數(shù)據(jù)丟失（圖片來源網(wǎng)絡(luò)）：

規(guī)避數(shù)據(jù)不一致（圖片來源網(wǎng)絡(luò)）：

Kafka高性能原因分析順序?qū)懭耄喉樞驅(qū)懭肱c隨機(jī)寫入速度相差高達(dá)6000倍批量處理：使用消息累加器僅多個消息批量發(fā)送，既節(jié)省帶寬有提高了發(fā)送速度消息壓縮：kafka支持隊消息壓縮，支持格式有：gzip、snapply、lz4，可以使用compression.type配置頁緩存：在消息發(fā)送后，并沒有等到消息寫入磁盤后才返回，而是到pageCache中就返回。pageCache與文件系統(tǒng)的寫入由操作系統(tǒng)自動完成零拷貝（zero-copy）：Kafka兩個重要過程都使用了零拷貝技術(shù)，且都是操作系統(tǒng)層面的狹義零拷貝，一是Producer生產(chǎn)的數(shù)據(jù)存到broker，二是Consumer從broker讀取數(shù)據(jù)。

正常的非零拷貝的數(shù)據(jù)拷貝過程：

硬盤—>內(nèi)核緩沖區(qū)—>用戶緩沖區(qū)—>內(nèi)核socket緩沖區(qū)—>協(xié)議引擎Producer生產(chǎn)的數(shù)據(jù)持久化到broker，采用mmap文件映射，實現(xiàn)順序的快速寫入；

硬盤—>內(nèi)核緩沖區(qū)—>共享到用戶空間緩存，共享而不是復(fù)制Customer從broker讀取數(shù)據(jù)，采用sendfile，將磁盤文件讀到OS內(nèi)核緩沖區(qū)后，直接轉(zhuǎn)到socketbuffer進(jìn)行網(wǎng)絡(luò)發(fā)送。sendfile()只是適用于應(yīng)用程序地址空間不需要對所訪問數(shù)據(jù)進(jìn)行處理的情況

硬盤—>內(nèi)核緩沖區(qū)—>內(nèi)核socket緩沖區(qū)—>協(xié)議引擎關(guān)鍵配置Broker配置名稱描述類型默認(rèn)值配置示例log.dir保存日志數(shù)據(jù)的目錄（對log.dirs屬性的補(bǔ)充）string/tmp/kafka-logslog.dirs保存日志數(shù)據(jù)的目錄，如果未設(shè)置將使用log.dir的配置stringnulllog.dirs=/home/kafka1,/home/kafka2,/home/kafka3zookeeper.connectZookeeper主機(jī)地址stringzookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka1listeners監(jiān)聽器列表-使用逗號分隔URI列表和監(jiān)聽器名稱。如果偵聽器名稱不是安全協(xié)議，則還必須設(shè)置tocol.map。指定主機(jī)名為來綁定到所有接口。留空則綁定到默認(rèn)接口上。stringnull合法監(jiān)聽器列表的示例：PLAINTEXT://myhost:9092，SSL://:9091CLIENT://:9092，REPLICATION://localhost:9093tocol.map偵聽器名稱和安全協(xié)議之間的映射。stringPLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLauto.create.topics.enable是否允許在服務(wù)器上自動創(chuàng)建topicbooleantrue推薦為falseunclean.leader.election.enable指定副本是否能夠不再ISR中被選舉為leader，即使這樣可能會丟失數(shù)據(jù)booleanfalse推薦為trueauto.leader.rebalance.enable是否允許leader平衡。后臺線程會定期檢查并觸發(fā)leader平衡booleantrue推薦為truelog.retention.{hoursminutesms}日志刪除的時間閾值（時、分、毫秒）intlog.rentention.bytes日志刪除的大小閾值long-1-1，表示沒有限制message.max.byteskafka允許的最大的一個批次消息大小int1000012=976KBTopic配置名稱描述類型默認(rèn)值配置示例retention.ms規(guī)定了該topic消息被保存的時長long604800000retention.bytes規(guī)定了要為該Topic預(yù)留多大的磁盤空間long-1max.message.bytesKafkaKafka允許接收該topic最大消息大小int1000012=976KBConsumer配置名稱描述類型默認(rèn)值配置示例er