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

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

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

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

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

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

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

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

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

基本流程：

ConsumerGroup中的Consumer向各自注冊的分區(qū)上進行消費消息

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

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

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

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

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

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

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

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

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

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

對于消息不一致的情況：

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

所謂leaderepoch實際上是一對值：（epoch,offset），epoch標識leader的版本號，從0開始，每變更一次leader，epoch+1；而offset對應于該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隨機寫入速度相差高達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ū)—>共享到用戶空間緩存，共享而不是復制Customer從broker讀取數(shù)據(jù)，采用sendfile，將磁盤文件讀到OS內(nèi)核緩沖區(qū)后，直接轉(zhuǎn)到socketbuffer進行網(wǎng)絡(luò)發(fā)送。sendfile()只是適用于應用程序地址空間不需要對所訪問數(shù)據(jù)進行處理的情況

硬盤—>內(nèi)核緩沖區(qū)—>內(nèi)核socket緩沖區(qū)—>協(xié)議引擎關(guān)鍵配置Broker配置名稱描述類型默認值配置示例log.dir保存日志數(shù)據(jù)的目錄（對log.dirs屬性的補充）string/tmp/kafka-logslog.dirs保存日志數(shù)據(jù)的目錄，如果未設(shè)置將使用log.dir的配置stringnulllog.dirs=/home/kafka1,/home/kafka2,/home/kafka3zookeeper.connectZookeeper主機地址stringzookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka1listeners監(jiān)聽器列表-使用逗號分隔URI列表和監(jiān)聽器名稱。如果偵聽器名稱不是安全協(xié)議，則還必須設(shè)置tocol.map。指定主機名為來綁定到所有接口。留空則綁定到默認接口上。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是否允許在服務器上自動創(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配置名稱描述類型默認值配置示例retention.ms規(guī)定了該topic消息被保存的時長long604800000retention.bytes規(guī)定了要為該Topic預留多大的磁盤空間long-1max.message.bytesKafkaKafka允許接收該topic最大消息大小int1000012=976KBConsumer配置名稱描述類型默認值配置示例er