Kafka史上最详细总结
Kafka
Kafka是最初由Linkedin公司開發(fā)，是一個分布式、支持分區(qū)的（partition）、多副本的（replica），基于zookeeper協(xié)調(diào)的分布式消息系統(tǒng)，它的最大的特性就是可以實時的處理大量數(shù)據(jù)以滿足各種需求場景：比如基于hadoop的批處理系統(tǒng)、低延遲的實時系統(tǒng)、storm/Spark流式處理引擎，web/nginx日志、訪問日志，消息服務等等，用scala語言編寫，Linkedin于2010年貢獻給了Apache基金會并成為頂級開源 項目。
?
1.前言
消息隊列的性能好壞，其文件存儲機制設(shè)計是衡量一個消息隊列服務技術(shù)水平和最關(guān)鍵指標之一。下面將從Kafka文件存儲機制和物理結(jié)構(gòu)角度，分析Kafka是如何實現(xiàn)高效文件存儲，及實際應用效果。
?
?1.1 ?Kafka的特性:
- 高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒，每個topic可以分多個partition, consumer group 對partition進行consume操作。
- 可擴展性：kafka集群支持熱擴展
- 持久性、可靠性：消息被持久化到本地磁盤，并且支持數(shù)據(jù)備份防止數(shù)據(jù)丟失
- 容錯性：允許集群中節(jié)點失敗（若副本數(shù)量為n,則允許n-1個節(jié)點失敗）
- 高并發(fā)：支持數(shù)千個客戶端同時讀寫
?
1.2 ? Kafka的使用場景：
- 日志收集：一個公司可以用Kafka可以收集各種服務的log，通過kafka以統(tǒng)一接口服務的方式開放給各種consumer，例如hadoop、Hbase、Solr等。
- 消息系統(tǒng)：解耦和生產(chǎn)者和消費者、緩存消息等。
- 用戶活動跟蹤：Kafka經(jīng)常被用來記錄web用戶或者app用戶的各種活動，如瀏覽網(wǎng)頁、搜索、點擊等活動，這些活動信息被各個服務器發(fā)布到kafka的topic中，然后訂閱者通過訂閱這些topic來做實時的監(jiān)控分析，或者裝載到hadoop、數(shù)據(jù)倉庫中做離線分析和挖掘。
- 運營指標：Kafka也經(jīng)常用來記錄運營監(jiān)控數(shù)據(jù)。包括收集各種分布式應用的數(shù)據(jù)，生產(chǎn)各種操作的集中反饋，比如報警和報告。
- 流式處理：比如spark streaming和storm
- 事件源
?
1.3 ?Kakfa的設(shè)計思想
-?Kakfa Broker Leader的選舉：Kakfa Broker集群受Zookeeper管理。所有的Kafka Broker節(jié)點一起去Zookeeper上注冊一個臨時節(jié)點，因為只有一個Kafka Broker會注冊成功，其他的都會失敗，所以這個成功在Zookeeper上注冊臨時節(jié)點的這個Kafka Broker會成為Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。（這個過程叫Controller在ZooKeeper注冊Watch）。這個Controller會監(jiān)聽其他的Kafka Broker的所有信息，如果這個kafka broker controller宕機了，在zookeeper上面的那個臨時節(jié)點就會消失，此時所有的kafka broker又會一起去Zookeeper上注冊一個臨時節(jié)點，因為只有一個Kafka Broker會注冊成功，其他的都會失敗，所以這個成功在Zookeeper上注冊臨時節(jié)點的這個Kafka Broker會成為Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。例如：一旦有一個broker宕機了，這個kafka broker controller會讀取該宕機broker上所有的partition在zookeeper上的狀態(tài)，并選取ISR列表中的一個replica作為partition leader（如果ISR列表中的replica全掛，選一個幸存的replica作為leader; 如果該partition的所有的replica都宕機了，則將新的leader設(shè)置為-1，等待恢復，等待ISR中的任一個Replica“活”過來，并且選它作為Leader；或選擇第一個“活”過來的Replica（不一定是ISR中的）作為Leader），這個broker宕機的事情，kafka controller也會通知zookeeper，zookeeper就會通知其他的kafka broker。
這里曾經(jīng)發(fā)生過一個bug，TalkingData使用Kafka0.8.1的時候，kafka controller在Zookeeper上注冊成功后，它和Zookeeper通信的timeout時間是6s，也就是如果kafka controller如果有6s中沒有和Zookeeper做心跳，那么Zookeeper就認為這個kafka controller已經(jīng)死了，就會在Zookeeper上把這個臨時節(jié)點刪掉，那么其他Kafka就會認為controller已經(jīng)沒了，就會再次搶著注冊臨時節(jié)點，注冊成功的那個kafka broker成為controller，然后，之前的那個kafka controller就需要各種shut down去關(guān)閉各種節(jié)點和事件的監(jiān)聽。但是當kafka的讀寫流量都非常巨大的時候，TalkingData的一個bug是，由于網(wǎng)絡等原因，kafka controller和Zookeeper有6s中沒有通信，于是重新選舉出了一個新的kafka controller，但是原來的controller在shut down的時候總是不成功，這個時候producer進來的message由于Kafka集群中存在兩個kafka controller而無法落地。導致數(shù)據(jù)淤積。
這里曾經(jīng)還有一個bug，TalkingData使用Kafka0.8.1的時候，當ack=0的時候，表示producer發(fā)送出去message，只要對應的kafka broker topic partition leader接收到的這條message，producer就返回成功，不管partition leader 是否真的成功把message真正存到kafka。當ack=1的時候，表示producer發(fā)送出去message，同步的把message存到對應topic的partition的leader上，然后producer就返回成功，partition leader異步的把message同步到其他partition replica上。當ack=all或-1，表示producer發(fā)送出去message，同步的把message存到對應topic的partition的leader和對應的replica上之后，才返回成功。但是如果某個kafka controller?切換的時候，會導致partition leader的切換（老的?kafka controller上面的partition leader會選舉到其他的kafka broker上）,但是這樣就會導致丟數(shù)據(jù)。
-??Consumergroup：各個consumer（consumer 線程）可以組成一個組（Consumer group?），partition中的每個message只能被組（Consumer group?）中的一個consumer（consumer 線程）消費，如果一個message可以被多個consumer（consumer 線程）消費的話，那么這些consumer必須在不同的組。Kafka不支持一個partition中的message由兩個或兩個以上的同一個consumer group下的consumer thread來處理，除非再啟動一個新的consumer group。所以如果想同時對一個topic做消費的話，啟動多個consumer group就可以了，但是要注意的是，這里的多個consumer的消費都必須是順序讀取partition里面的message，新啟動的consumer默認從partition隊列最頭端最新的地方開始阻塞的讀message。它不能像AMQ那樣可以多個BET作為consumer去互斥的（for update悲觀鎖）并發(fā)處理message，這是因為多個BET去消費一個Queue中的數(shù)據(jù)的時候，由于要保證不能多個線程拿同一條message，所以就需要行級別悲觀所（for update）,這就導致了consume的性能下降，吞吐量不夠。而kafka為了保證吞吐量，只允許同一個consumer group下的一個consumer線程去訪問一個partition。如果覺得效率不高的時候，可以加partition的數(shù)量來橫向擴展，那么再加新的consumer thread去消費。如果想多個不同的業(yè)務都需要這個topic的數(shù)據(jù)，起多個consumer group就好了，大家都是順序的讀取message，offsite的值互不影響。這樣沒有鎖競爭，充分發(fā)揮了橫向的擴展性，吞吐量極高。這也就形成了分布式消費的概念。
? ? 當啟動一個consumer group去消費一個topic的時候，無論topic里面有多個少個partition，無論我們consumer group里面配置了多少個consumer thread，這個consumer group下面的所有consumer thread一定會消費全部的partition；即便這個consumer group下只有一個consumer thread，那么這個consumer thread也會去消費所有的partition。因此，最優(yōu)的設(shè)計就是，consumer group下的consumer thread的數(shù)量等于partition數(shù)量，這樣效率是最高的。
? ? 同一partition的一條message只能被同一個Consumer Group內(nèi)的一個Consumer消費。不能夠一個consumer group的多個consumer同時消費一個partition。
? ? 一個consumer group下，無論有多少個consumer，這個consumer group一定回去把這個topic下所有的partition都消費了。當consumer group里面的consumer數(shù)量小于這個topic下的partition數(shù)量的時候，如下圖groupA,groupB，就會出現(xiàn)一個conusmer thread消費多個partition的情況，總之是這個topic下的partition都會被消費。如果consumer group里面的consumer數(shù)量等于這個topic下的partition數(shù)量的時候，如下圖groupC，此時效率是最高的，每個partition都有一個consumer thread去消費。當consumer group里面的consumer數(shù)量大于這個topic下的partition數(shù)量的時候，如下圖GroupD，就會有一個consumer thread空閑。因此，我們在設(shè)定consumer group的時候，只需要指明里面有幾個consumer數(shù)量即可，無需指定對應的消費partition序號，consumer會自動進行rebalance。
? ? 多個Consumer Group下的consumer可以消費同一條message，但是這種消費也是以o（1）的方式順序的讀取message去消費,，所以一定會重復消費這批message的，不能向AMQ那樣多個BET作為consumer消費（對message加鎖，消費的時候不能重復消費message）
-?Consumer Rebalance的觸發(fā)條件：（1）Consumer增加或刪除會觸發(fā) Consumer Group的Rebalance（2）Broker的增加或者減少都會觸發(fā) Consumer Rebalance
-?Consumer：?Consumer處理partition里面的message的時候是o（1）順序讀取的。所以必須維護著上一次讀到哪里的offsite信息。high level API,offset存于Zookeeper中，low level API的offset由自己維護。一般來說都是使用high level api的。Consumer的delivery gurarantee，默認是讀完message先commmit再處理message，autocommit默認是true，這時候先commit就會更新offsite+1，一旦處理失敗，offsite已經(jīng)+1，這個時候就會丟message；也可以配置成讀完消息處理再commit，這種情況下consumer端的響應就會比較慢的，需要等處理完才行。
一般情況下，一定是一個consumer group處理一個topic的message。Best Practice是這個consumer group里面consumer的數(shù)量等于topic里面partition的數(shù)量，這樣效率是最高的，一個consumer thread處理一個partition。如果這個consumer group里面consumer的數(shù)量小于topic里面partition的數(shù)量，就會有consumer thread同時處理多個partition（這個是kafka自動的機制，我們不用指定），但是總之這個topic里面的所有partition都會被處理到的。。如果這個consumer group里面consumer的數(shù)量大于topic里面partition的數(shù)量，多出的consumer thread就會閑著啥也不干，剩下的是一個consumer thread處理一個partition，這就造成了資源的浪費，因為一個partition不可能被兩個consumer thread去處理。所以我們線上的分布式多個service服務，每個service里面的kafka consumer數(shù)量都小于對應的topic的partition數(shù)量，但是所有服務的consumer數(shù)量只和等于partition的數(shù)量，這是因為分布式service服務的所有consumer都來自一個consumer group，如果來自不同的consumer group就會處理重復的message了（同一個consumer group下的consumer不能處理同一個partition，不同的consumer group可以處理同一個topic，那么都是順序處理message，一定會處理重復的。一般這種情況都是兩個不同的業(yè)務邏輯，才會啟動兩個consumer group來處理一個topic）。
?
如果producer的流量增大，當前的topic的parition數(shù)量=consumer數(shù)量，這時候的應對方式就是很想擴展：增加topic下的partition，同時增加這個consumer group下的consumer。
??? ? ? ? ? ? ??
-?Delivery Mode :?Kafka producer 發(fā)送message不用維護message的offsite信息，因為這個時候，offsite就相當于一個自增id，producer就盡管發(fā)送message就好了。而且Kafka與AMQ不同，AMQ大都用在處理業(yè)務邏輯上，而Kafka大都是日志，所以Kafka的producer一般都是大批量的batch發(fā)送message，向這個topic一次性發(fā)送一大批message，load balance到一個partition上，一起插進去，offsite作為自增id自己增加就好。但是Consumer端是需要維護這個partition當前消費到哪個message的offsite信息的，這個offsite信息，high level api是維護在Zookeeper上，low level api是自己的程序維護。（Kafka管理界面上只能顯示high level api的consumer部分，因為low level api的partition offsite信息是程序自己維護，kafka是不知道的，無法在管理界面上展示 ）當使用high level api的時候，先拿message處理，再定時自動commit offsite+1（也可以改成手動）, 并且kakfa處理message是沒有鎖操作的。因此如果處理message失敗，此時還沒有commit offsite+1，當consumer thread重啟后會重復消費這個message。但是作為高吞吐量高并發(fā)的實時處理系統(tǒng)，at least once的情況下，至少一次會被處理到，是可以容忍的。如果無法容忍，就得使用low level api來自己程序維護這個offsite信息，那么想什么時候commit offsite+1就自己搞定了。
?
-?Topic & Partition：Topic相當于傳統(tǒng)消息系統(tǒng)MQ中的一個隊列queue，producer端發(fā)送的message必須指定是發(fā)送到哪個topic，但是不需要指定topic下的哪個partition，因為kafka會把收到的message進行l(wèi)oad balance，均勻的分布在這個topic下的不同的partition上（ hash(message) % [broker數(shù)量] ?）。物理上存儲上，這個topic會分成一個或多個partition，每個partiton相當于是一個子queue。在物理結(jié)構(gòu)上，每個partition對應一個物理的目錄（文件夾），文件夾命名是[topicname]_[partition]_[序號]，一個topic可以有無數(shù)多的partition，根據(jù)業(yè)務需求和數(shù)據(jù)量來設(shè)置。在kafka配置文件中可隨時更高num.partitions參數(shù)來配置更改topic的partition數(shù)量，在創(chuàng)建Topic時通過參數(shù)指定parittion數(shù)量。Topic創(chuàng)建之后通過Kafka提供的工具也可以修改partiton數(shù)量。
? ?一般來說，（1）一個Topic的Partition數(shù)量大于等于Broker的數(shù)量，可以提高吞吐率。（2）同一個Partition的Replica盡量分散到不同的機器，高可用。
??當add a new partition的時候，partition里面的message不會重新進行分配，原來的partition里面的message數(shù)據(jù)不會變，新加的這個partition剛開始是空的，隨后進入這個topic的message就會重新參與所有partition的load balance
-?Partition Replica：每個partition可以在其他的kafka broker節(jié)點上存副本，以便某個kafka broker節(jié)點宕機不會影響這個kafka集群。存replica副本的方式是按照kafka broker的順序存。例如有5個kafka broker節(jié)點，某個topic有3個partition，每個partition存2個副本，那么partition1存broker1,broker2，partition2存broker2,broker3。。。以此類推（replica副本數(shù)目不能大于kafka broker節(jié)點的數(shù)目，否則報錯。這里的replica數(shù)其實就是partition的副本總數(shù)，其中包括一個leader，其他的就是copy副本）。這樣如果某個broker宕機，其實整個kafka內(nèi)數(shù)據(jù)依然是完整的。但是，replica副本數(shù)越高，系統(tǒng)雖然越穩(wěn)定，但是回來帶資源和性能上的下降；replica副本少的話，也會造成系統(tǒng)丟數(shù)據(jù)的風險。
? （1）怎樣傳送消息：producer先把message發(fā)送到partition leader，再由leader發(fā)送給其他partition follower。（如果讓producer發(fā)送給每個replica那就太慢了）
? （2）在向Producer發(fā)送ACK前需要保證有多少個Replica已經(jīng)收到該消息：根據(jù)ack配的個數(shù)而定
? （3）怎樣處理某個Replica不工作的情況：如果這個部工作的partition replica不在ack列表中，就是producer在發(fā)送消息到partition leader上，partition leader向partition follower發(fā)送message沒有響應而已，這個不會影響整個系統(tǒng)，也不會有什么問題。如果這個不工作的partition replica在ack列表中的話，producer發(fā)送的message的時候會等待這個不工作的partition replca寫message成功，但是會等到time out，然后返回失敗因為某個ack列表中的partition replica沒有響應，此時kafka會自動的把這個部工作的partition replica從ack列表中移除，以后的producer發(fā)送message的時候就不會有這個ack列表下的這個部工作的partition replica了。?
? （4）怎樣處理Failed Replica恢復回來的情況：如果這個partition replica之前不在ack列表中，那么啟動后重新受Zookeeper管理即可，之后producer發(fā)送message的時候，partition leader會繼續(xù)發(fā)送message到這個partition follower上。如果這個partition replica之前在ack列表中，此時重啟后，需要把這個partition replica再手動加到ack列表中。（ack列表是手動添加的，出現(xiàn)某個部工作的partition replica的時候自動從ack列表中移除的）
-?Partition leader與follower：partition也有l(wèi)eader和follower之分。leader是主partition，producer寫kafka的時候先寫partition leader，再由partition leader push給其他的partition follower。partition leader與follower的信息受Zookeeper控制，一旦partition leader所在的broker節(jié)點宕機，zookeeper會沖其他的broker的partition follower上選擇follower變?yōu)閜arition leader。
-?Topic分配partition和partition replica的算法：（1）將Broker（size=n）和待分配的Partition排序。（2）將第i個Partition分配到第（i%n）個Broker上。（3）將第i個Partition的第j個Replica分配到第（(i + j) % n）個Broker上
?
-?消息投遞可靠性
一個消息如何算投遞成功，Kafka提供了三種模式：
- 第一種是啥都不管，發(fā)送出去就當作成功，這種情況當然不能保證消息成功投遞到broker；
- 第二種是Master-Slave模型，只有當Master和所有Slave都接收到消息時，才算投遞成功，這種模型提供了最高的投遞可靠性，但是損傷了性能；
- 第三種模型，即只要Master確認收到消息就算投遞成功；實際使用時，根據(jù)應用特性選擇，絕大多數(shù)情況下都會中和可靠性和性能選擇第三種模型
? 消息在broker上的可靠性，因為消息會持久化到磁盤上，所以如果正常stop一個broker，其上的數(shù)據(jù)不會丟失；但是如果不正常stop，可能會使存在頁面緩存來不及寫入磁盤的消息丟失，這可以通過配置flush頁面緩存的周期、閾值緩解，但是同樣會頻繁的寫磁盤會影響性能，又是一個選擇題，根據(jù)實際情況配置。
? 消息消費的可靠性，Kafka提供的是“At least once”模型，因為消息的讀取進度由offset提供，offset可以由消費者自己維護也可以維護在zookeeper里，但是當消息消費后consumer掛掉，offset沒有即時寫回，就有可能發(fā)生重復讀的情況，這種情況同樣可以通過調(diào)整commit offset周期、閾值緩解，甚至消費者自己把消費和commit offset做成一個事務解決，但是如果你的應用不在乎重復消費，那就干脆不要解決，以換取最大的性能。
?
-?Partition ack：當ack=1，表示producer寫partition leader成功后，broker就返回成功，無論其他的partition follower是否寫成功。當ack=2，表示producer寫partition leader和其他一個follower成功的時候，broker就返回成功，無論其他的partition follower是否寫成功。當ack=-1[parition的數(shù)量]的時候，表示只有producer全部寫成功的時候，才算成功，kafka broker才返回成功信息。這里需要注意的是，如果ack=1的時候，一旦有個broker宕機導致partition的follower和leader切換，會導致丟數(shù)據(jù)。
??
-?message狀態(tài)：在Kafka中，消息的狀態(tài)被保存在consumer中，broker不會關(guān)心哪個消息被消費了被誰消費了，只記錄一個offset值（指向partition中下一個要被消費的消息位置），這就意味著如果consumer處理不好的話，broker上的一個消息可能會被消費多次。
-?message持久化：Kafka中會把消息持久化到本地文件系統(tǒng)中，并且保持o(1)極高的效率。我們眾所周知IO讀取是非常耗資源的性能也是最慢的，這就是為了數(shù)據(jù)庫的瓶頸經(jīng)常在IO上，需要換SSD硬盤的原因。但是Kafka作為吞吐量極高的MQ，卻可以非常高效的message持久化到文件。這是因為Kafka是順序?qū)懭雘（1）的時間復雜度，速度非?？臁Ｒ彩歉咄掏铝康脑?。由于message的寫入持久化是順序?qū)懭氲?#xff0c;因此message在被消費的時候也是按順序被消費的，保證partition的message是順序消費的。一般的機器,單機每秒100k條數(shù)據(jù)。
-?message有效期：Kafka會長久保留其中的消息，以便consumer可以多次消費，當然其中很多細節(jié)是可配置的。
-?Produer :?Producer向Topic發(fā)送message，不需要指定partition，直接發(fā)送就好了。kafka通過partition ack來控制是否發(fā)送成功并把信息返回給producer，producer可以有任意多的thread，這些kafka服務器端是不care的。Producer端的delivery guarantee默認是At least once的。也可以設(shè)置Producer異步發(fā)送實現(xiàn)At most once。Producer可以用主鍵冪等性實現(xiàn)Exactly once
-?Kafka高吞吐量： Kafka的高吞吐量體現(xiàn)在讀寫上，分布式并發(fā)的讀和寫都非?？?#xff0c;寫的性能體現(xiàn)在以o(1)的時間復雜度進行順序?qū)懭搿Ｗx的性能體現(xiàn)在以o(1)的時間復雜度進行順序讀取，?對topic進行partition分區(qū)，consume group中的consume線程可以以很高能性能進行順序讀。
- Kafka delivery guarantee(message傳送保證)：（1）At most once消息可能會丟，絕對不會重復傳輸；（2）At least once 消息絕對不會丟，但是可能會重復傳輸；（3）Exactly once每條信息肯定會被傳輸一次且僅傳輸一次，這是用戶想要的。
-?批量發(fā)送：Kafka支持以消息集合為單位進行批量發(fā)送，以提高push效率。
-?push-and-pull?: Kafka中的Producer和consumer采用的是push-and-pull模式，即Producer只管向broker push消息，consumer只管從broker pull消息，兩者對消息的生產(chǎn)和消費是異步的。
-?Kafka集群中broker之間的關(guān)系：不是主從關(guān)系，各個broker在集群中地位一樣，我們可以隨意的增加或刪除任何一個broker節(jié)點。
-?負載均衡方面： Kafka提供了一個 metadata API來管理broker之間的負載（對Kafka0.8.x而言，對于0.7.x主要靠zookeeper來實現(xiàn)負載均衡）。
-?同步異步：Producer采用異步push方式，極大提高Kafka系統(tǒng)的吞吐率（可以通過參數(shù)控制是采用同步還是異步方式）。
-?分區(qū)機制partition：Kafka的broker端支持消息分區(qū)partition，Producer可以決定把消息發(fā)到哪個partition，在一個partition?中message的順序就是Producer發(fā)送消息的順序，一個topic中可以有多個partition，具體partition的數(shù)量是可配置的。partition的概念使得kafka作為MQ可以橫向擴展，吞吐量巨大。partition可以設(shè)置replica副本，replica副本存在不同的kafka broker節(jié)點上，第一個partition是leader,其他的是follower，message先寫到partition leader上，再由partition leader push到parition follower上。所以說kafka可以水平擴展，也就是擴展partition。
-?離線數(shù)據(jù)裝載：Kafka由于對可拓展的數(shù)據(jù)持久化的支持，它也非常適合向Hadoop或者數(shù)據(jù)倉庫中進行數(shù)據(jù)裝載。
-?實時數(shù)據(jù)與離線數(shù)據(jù)：kafka既支持離線數(shù)據(jù)也支持實時數(shù)據(jù)，因為kafka的message持久化到文件，并可以設(shè)置有效期，因此可以把kafka作為一個高效的存儲來使用，可以作為離線數(shù)據(jù)供后面的分析。當然作為分布式實時消息系統(tǒng)，大多數(shù)情況下還是用于實時的數(shù)據(jù)處理的，但是當cosumer消費能力下降的時候可以通過message的持久化在淤積數(shù)據(jù)在kafka。
-?插件支持：現(xiàn)在不少活躍的社區(qū)已經(jīng)開發(fā)出不少插件來拓展Kafka的功能，如用來配合Storm、Hadoop、flume相關(guān)的插件。
-?解耦: ?相當于一個MQ，使得Producer和Consumer之間異步的操作，系統(tǒng)之間解耦
-?冗余: ?replica有多個副本，保證一個broker node宕機后不會影響整個服務
-?擴展性: ?broker節(jié)點可以水平擴展，partition也可以水平增加，partition replica也可以水平增加
-?峰值: ?在訪問量劇增的情況下，kafka水平擴展, 應用仍然需要繼續(xù)發(fā)揮作用
-?可恢復性: ?系統(tǒng)的一部分組件失效時，由于有partition的replica副本，不會影響到整個系統(tǒng)。
-?順序保證性：由于kafka的producer的寫message與consumer去讀message都是順序的讀寫，保證了高效的性能。
-?緩沖：由于producer那面可能業(yè)務很簡單，而后端consumer業(yè)務會很復雜并有數(shù)據(jù)庫的操作，因此肯定是producer會比consumer處理速度快，如果沒有kafka，producer直接調(diào)用consumer，那么就會造成整個系統(tǒng)的處理速度慢，加一層kafka作為MQ，可以起到緩沖的作用。
-?異步通信：作為MQ，Producer與Consumer異步通信
2.Kafka文件存儲機制
2.1 Kafka部分名詞解釋如下：
?
?????Kafka中發(fā)布訂閱的對象是topic。我們可以為每類數(shù)據(jù)創(chuàng)建一個topic，把向topic發(fā)布消息的客戶端稱作producer，從topic訂閱消息的客戶端稱作consumer。Producers和consumers可以同時從多個topic讀寫數(shù)據(jù)。一個kafka集群由一個或多個broker服務器組成，它負責持久化和備份具體的kafka消息。
- Broker：Kafka節(jié)點，一個Kafka節(jié)點就是一個broker，多個broker可以組成一個Kafka集群。
- Topic：一類消息，消息存放的目錄即主題，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能夠同時負責多個topic的分發(fā)。
- Partition：topic物理上的分組，一個topic可以分為多個partition，每個partition是一個有序的隊列
- Segment：partition物理上由多個segment組成，每個Segment存著message信息
- Producer?: 生產(chǎn)message發(fā)送到topic
- Consumer?: 訂閱topic消費message, consumer作為一個線程來消費
- Consumer Group：一個Consumer Group包含多個consumer, 這個是預先在配置文件中配置好的。各個consumer（consumer 線程）可以組成一個組（Consumer group ），partition中的每個message只能被組（Consumer group ） 中的一個consumer（consumer 線程 ）消費，如果一個message可以被多個consumer（consumer 線程 ） 消費的話，那么這些consumer必須在不同的組。Kafka不支持一個partition中的message由兩個或兩個以上的consumer thread來處理，即便是來自不同的consumer group的也不行。它不能像AMQ那樣可以多個BET作為consumer去處理message，這是因為多個BET去消費一個Queue中的數(shù)據(jù)的時候，由于要保證不能多個線程拿同一條message，所以就需要行級別悲觀所（for update）,這就導致了consume的性能下降，吞吐量不夠。而kafka為了保證吞吐量，只允許一個consumer線程去訪問一個partition。如果覺得效率不高的時候，可以加partition的數(shù)量來橫向擴展，那么再加新的consumer thread去消費。這樣沒有鎖競爭，充分發(fā)揮了橫向的擴展性，吞吐量極高。這也就形成了分布式消費的概念。
? - 2.2 kafka一些原理概念
1.持久化
kafka使用文件存儲消息(append only log),這就直接決定kafka在性能上嚴重依賴文件系統(tǒng)的本身特性.且無論任何OS下,對文件系統(tǒng)本身的優(yōu)化是非常艱難的.文件緩存/直接內(nèi)存映射等是常用的手段.因為kafka是對日志文件進行append操作,因此磁盤檢索的開支是較小的;同時為了減少磁盤寫入的次數(shù),broker會將消息暫時buffer起來,當消息的個數(shù)(或尺寸)達到一定閥值時,再flush到磁盤,這樣減少了磁盤IO調(diào)用的次數(shù).對于kafka而言,較高性能的磁盤,將會帶來更加直接的性能提升.
?
2.性能
除磁盤IO之外,我們還需要考慮網(wǎng)絡IO,這直接關(guān)系到kafka的吞吐量問題.kafka并沒有提供太多高超的技巧;對于producer端,可以將消息buffer起來,當消息的條數(shù)達到一定閥值時,批量發(fā)送給broker;對于consumer端也是一樣,批量fetch多條消息.不過消息量的大小可以通過配置文件來指定.對于kafka broker端,似乎有個sendfile系統(tǒng)調(diào)用可以潛在的提升網(wǎng)絡IO的性能:將文件的數(shù)據(jù)映射到系統(tǒng)內(nèi)存中,socket直接讀取相應的內(nèi)存區(qū)域即可,而無需進程再次copy和交換(這里涉及到"磁盤IO數(shù)據(jù)"/"內(nèi)核內(nèi)存"/"進程內(nèi)存"/"網(wǎng)絡緩沖區(qū)",多者之間的數(shù)據(jù)copy).
其實對于producer/consumer/broker三者而言,CPU的開支應該都不大,因此啟用消息壓縮機制是一個良好的策略;壓縮需要消耗少量的CPU資源,不過對于kafka而言,網(wǎng)絡IO更應該需要考慮.可以將任何在網(wǎng)絡上傳輸?shù)南⒍冀?jīng)過壓縮.kafka支持gzip/snappy等多種壓縮方式.
?
3.負載均衡
kafka集群中的任何一個broker,都可以向producer提供metadata信息,這些metadata中包含"集群中存活的servers列表"/"partitions leader列表"等信息(請參看zookeeper中的節(jié)點信息). 當producer獲取到metadata信息之后, producer將會和Topic下所有partition leader保持socket連接;消息由producer直接通過socket發(fā)送到broker,中間不會經(jīng)過任何"路由層".
異步發(fā)送，將多條消息暫且在客戶端buffer起來,并將他們批量發(fā)送到broker;小數(shù)據(jù)IO太多,會拖慢整體的網(wǎng)絡延遲,批量延遲發(fā)送事實上提升了網(wǎng)絡效率;不過這也有一定的隱患,比如當producer失效時,那些尚未發(fā)送的消息將會丟失。
?
4.Topic模型
其他JMS實現(xiàn),消息消費的位置是有prodiver保留,以便避免重復發(fā)送消息或者將沒有消費成功的消息重發(fā)等,同時還要控制消息的狀態(tài).這就要求JMS broker需要太多額外的工作.在kafka中,partition中的消息只有一個consumer在消費,且不存在消息狀態(tài)的控制,也沒有復雜的消息確認機制,可見kafka broker端是相當輕量級的.當消息被consumer接收之后,consumer可以在本地保存最后消息的offset,并間歇性的向zookeeper注冊offset.由此可見,consumer客戶端也很輕量級。
kafka中consumer負責維護消息的消費記錄,而broker則不關(guān)心這些,這種設(shè)計不僅提高了consumer端的靈活性,也適度的減輕了broker端設(shè)計的復雜度;這是和眾多JMS prodiver的區(qū)別.此外,kafka中消息ACK的設(shè)計也和JMS有很大不同,kafka中的消息是批量(通常以消息的條數(shù)或者chunk的尺寸為單位)發(fā)送給consumer,當消息消費成功后,向zookeeper提交消息的offset,而不會向broker交付ACK.或許你已經(jīng)意識到,這種"寬松"的設(shè)計,將會有"丟失"消息/"消息重發(fā)"的危險.
?
5.消息傳輸一致
Kafka提供3種消息傳輸一致性語義：最多1次，最少1次，恰好1次。
最少1次：可能會重傳數(shù)據(jù)，有可能出現(xiàn)數(shù)據(jù)被重復處理的情況;
最多1次：可能會出現(xiàn)數(shù)據(jù)丟失情況;
恰好1次：并不是指真正只傳輸1次，只不過有一個機制。確保不會出現(xiàn)“數(shù)據(jù)被重復處理”和“數(shù)據(jù)丟失”的情況。
?
at most once: 消費者fetch消息,然后保存offset,然后處理消息;當client保存offset之后,但是在消息處理過程中consumer進程失效(crash),導致部分消息未能繼續(xù)處理.那么此后可能其他consumer會接管,但是因為offset已經(jīng)提前保存,那么新的consumer將不能fetch到offset之前的消息(盡管它們尚沒有被處理),這就是"at most once".
at least once: 消費者fetch消息,然后處理消息,然后保存offset.如果消息處理成功之后,但是在保存offset階段zookeeper異?；蛘遚onsumer失效,導致保存offset操作未能執(zhí)行成功,這就導致接下來再次fetch時可能獲得上次已經(jīng)處理過的消息,這就是"at least once".
"Kafka Cluster"到消費者的場景中可以采取以下方案來得到“恰好1次”的一致性語義：
最少1次＋消費者的輸出中額外增加已處理消息最大編號：由于已處理消息最大編號的存在，不會出現(xiàn)重復處理消息的情況。
?
6.副本
kafka中,replication策略是基于partition,而不是topic;kafka將每個partition數(shù)據(jù)復制到多個server上,任何一個partition有一個leader和多個follower(可以沒有);備份的個數(shù)可以通過broker配置文件來設(shè)定。leader處理所有的read-write請求,follower需要和leader保持同步.Follower就像一個"consumer",消費消息并保存在本地日志中;leader負責跟蹤所有的follower狀態(tài),如果follower"落后"太多或者失效,leader將會把它從replicas同步列表中刪除.當所有的follower都將一條消息保存成功,此消息才被認為是"committed",那么此時consumer才能消費它,這種同步策略,就要求follower和leader之間必須具有良好的網(wǎng)絡環(huán)境.即使只有一個replicas實例存活,仍然可以保證消息的正常發(fā)送和接收,只要zookeeper集群存活即可.
選擇follower時需要兼顧一個問題,就是新leader server上所已經(jīng)承載的partition leader的個數(shù),如果一個server上有過多的partition leader,意味著此server將承受著更多的IO壓力.在選舉新leader,需要考慮到"負載均衡",partition leader較少的broker將會更有可能成為新的leader.
?
7.log
每個log entry格式為"4個字節(jié)的數(shù)字N表示消息的長度" + "N個字節(jié)的消息內(nèi)容";每個日志都有一個offset來唯一的標記一條消息,offset的值為8個字節(jié)的數(shù)字,表示此消息在此partition中所處的起始位置..每個partition在物理存儲層面,有多個log file組成(稱為segment).segment file的命名為"最小offset".kafka.例如"00000000000.kafka";其中"最小offset"表示此segment中起始消息的offset.
獲取消息時,需要指定offset和最大chunk尺寸,offset用來表示消息的起始位置,chunk size用來表示最大獲取消息的總長度(間接的表示消息的條數(shù)).根據(jù)offset,可以找到此消息所在segment文件,然后根據(jù)segment的最小offset取差值,得到它在file中的相對位置,直接讀取輸出即可.
?
8.分布式
kafka使用zookeeper來存儲一些meta信息,并使用了zookeeper watch機制來發(fā)現(xiàn)meta信息的變更并作出相應的動作(比如consumer失效,觸發(fā)負載均衡等)
Broker node registry: 當一個kafka broker啟動后,首先會向zookeeper注冊自己的節(jié)點信息(臨時znode),同時當broker和zookeeper斷開連接時,此znode也會被刪除.
Broker Topic Registry: 當一個broker啟動時,會向zookeeper注冊自己持有的topic和partitions信息,仍然是一個臨時znode.
Consumer and Consumer group: 每個consumer客戶端被創(chuàng)建時,會向zookeeper注冊自己的信息;此作用主要是為了"負載均衡".一個group中的多個consumer可以交錯的消費一個topic的所有partitions;簡而言之,保證此topic的所有partitions都能被此group所消費,且消費時為了性能考慮,讓partition相對均衡的分散到每個consumer上.
Consumer id Registry: 每個consumer都有一個唯一的ID(host:uuid,可以通過配置文件指定,也可以由系統(tǒng)生成),此id用來標記消費者信息.
Consumer offset Tracking: 用來跟蹤每個consumer目前所消費的partition中最大的offset.此znode為持久節(jié)點,可以看出offset跟group_id有關(guān),以表明當group中一個消費者失效,其他consumer可以繼續(xù)消費.
Partition Owner registry: 用來標記partition正在被哪個consumer消費.臨時znode。此節(jié)點表達了"一個partition"只能被group下一個consumer消費,同時當group下某個consumer失效,那么將會觸發(fā)負載均衡(即:讓partitions在多個consumer間均衡消費,接管那些"游離"的partitions)
當consumer啟動時,所觸發(fā)的操作:
A) 首先進行"Consumer id Registry";
B) 然后在"Consumer id Registry"節(jié)點下注冊一個watch用來監(jiān)聽當前group中其他consumer的"leave"和"join";只要此znode path下節(jié)點列表變更,都會觸發(fā)此group下consumer的負載均衡.(比如一個consumer失效,那么其他consumer接管partitions).
C) 在"Broker id registry"節(jié)點下,注冊一個watch用來監(jiān)聽broker的存活情況;如果broker列表變更,將會觸發(fā)所有的groups下的consumer重新balance.
?
總結(jié):
1) Producer端使用zookeeper用來"發(fā)現(xiàn)"broker列表,以及和Topic下每個partition leader建立socket連接并發(fā)送消息.
2) Broker端使用zookeeper用來注冊broker信息,已經(jīng)監(jiān)測partition leader存活性.
3) Consumer端使用zookeeper用來注冊consumer信息,其中包括consumer消費的partition列表等,同時也用來發(fā)現(xiàn)broker列表,并和partition leader建立socket連接,并獲取消息。
?
9.Leader的選擇
Kafka的核心是日志文件，日志文件在集群中的同步是分布式數(shù)據(jù)系統(tǒng)最基礎(chǔ)的要素。
如果leaders永遠不會down的話我們就不需要followers了！一旦leader down掉了，需要在followers中選擇一個新的leader.但是followers本身有可能延時太久或者crash，所以必須選擇高質(zhì)量的follower作為leader.必須保證，一旦一個消息被提交了，但是leader down掉了，新選出的leader必須可以提供這條消息。大部分的分布式系統(tǒng)采用了多數(shù)投票法則選擇新的leader,對于多數(shù)投票法則，就是根據(jù)所有副本節(jié)點的狀況動態(tài)的選擇最適合的作為leader.Kafka并不是使用這種方法。
Kafka動態(tài)維護了一個同步狀態(tài)的副本的集合（a set of in-sync replicas），簡稱ISR，在這個集合中的節(jié)點都是和leader保持高度一致的，任何一條消息必須被這個集合中的每個節(jié)點讀取并追加到日志中了，才回通知外部這個消息已經(jīng)被提交了。因此這個集合中的任何一個節(jié)點隨時都可以被選為leader.ISR在ZooKeeper中維護。ISR中有f+1個節(jié)點，就可以允許在f個節(jié)點down掉的情況下不會丟失消息并正常提供服。ISR的成員是動態(tài)的，如果一個節(jié)點被淘汰了，當它重新達到“同步中”的狀態(tài)時，他可以重新加入ISR.這種leader的選擇方式是非常快速的，適合kafka的應用場景。
一個邪惡的想法：如果所有節(jié)點都down掉了怎么辦？Kafka對于數(shù)據(jù)不會丟失的保證，是基于至少一個節(jié)點是存活的，一旦所有節(jié)點都down了，這個就不能保證了。
實際應用中，當所有的副本都down掉時，必須及時作出反應。可以有以下兩種選擇:
1. 等待ISR中的任何一個節(jié)點恢復并擔任leader。
2. 選擇所有節(jié)點中（不只是ISR）第一個恢復的節(jié)點作為leader.
這是一個在可用性和連續(xù)性之間的權(quán)衡。如果等待ISR中的節(jié)點恢復，一旦ISR中的節(jié)點起不起來或者數(shù)據(jù)都是了，那集群就永遠恢復不了了。如果等待ISR意外的節(jié)點恢復，這個節(jié)點的數(shù)據(jù)就會被作為線上數(shù)據(jù)，有可能和真實的數(shù)據(jù)有所出入，因為有些數(shù)據(jù)它可能還沒同步到。Kafka目前選擇了第二種策略，在未來的版本中將使這個策略的選擇可配置，可以根據(jù)場景靈活的選擇。
這種窘境不只Kafka會遇到，幾乎所有的分布式數(shù)據(jù)系統(tǒng)都會遇到。
?
10.副本管理
以上僅僅以一個topic一個分區(qū)為例子進行了討論，但實際上一個Kafka將會管理成千上萬的topic分區(qū).Kafka盡量的使所有分區(qū)均勻的分布到集群所有的節(jié)點上而不是集中在某些節(jié)點上，另外主從關(guān)系也盡量均衡這樣每個幾點都會擔任一定比例的分區(qū)的leader.
優(yōu)化leader的選擇過程也是很重要的，它決定了系統(tǒng)發(fā)生故障時的空窗期有多久。Kafka選擇一個節(jié)點作為“controller”,當發(fā)現(xiàn)有節(jié)點down掉的時候它負責在游泳分區(qū)的所有節(jié)點中選擇新的leader,這使得Kafka可以批量的高效的管理所有分區(qū)節(jié)點的主從關(guān)系。如果controller down掉了，活著的節(jié)點中的一個會備切換為新的controller.
?
11.Leader與副本同步
對于某個分區(qū)來說，保存正分區(qū)的"broker"為該分區(qū)的"leader"，保存?zhèn)浞莘謪^(qū)的"broker"為該分區(qū)的"follower"。備份分區(qū)會完全復制正分區(qū)的消息，包括消息的編號等附加屬性值。為了保持正分區(qū)和備份分區(qū)的內(nèi)容一致，Kafka采取的方案是在保存?zhèn)浞莘謪^(qū)的"broker"上開啟一個消費者進程進行消費，從而使得正分區(qū)的內(nèi)容與備份分區(qū)的內(nèi)容保持一致。一般情況下，一個分區(qū)有一個“正分區(qū)”和零到多個“備份分區(qū)”。可以配置“正分區(qū)+備份分區(qū)”的總數(shù)量，關(guān)于這個配置，不同主題可以有不同的配置值。注意，生產(chǎn)者，消費者只與保存正分區(qū)的"leader"進行通信。
?
Kafka允許topic的分區(qū)擁有若干副本，這個數(shù)量是可以配置的，你可以為每個topic配置副本的數(shù)量。Kafka會自動在每個副本上備份數(shù)據(jù)，所以當一個節(jié)點down掉時數(shù)據(jù)依然是可用的。
Kafka的副本功能不是必須的，你可以配置只有一個副本，這樣其實就相當于只有一份數(shù)據(jù)。
創(chuàng)建副本的單位是topic的分區(qū)，每個分區(qū)都有一個leader和零或多個followers.所有的讀寫操作都由leader處理，一般分區(qū)的數(shù)量都比broker的數(shù)量多的多，各分區(qū)的leader均勻的分布在brokers中。所有的followers都復制leader的日志，日志中的消息和順序都和leader中的一致。followers向普通的consumer那樣從leader那里拉取消息并保存在自己的日志文件中。
許多分布式的消息系統(tǒng)自動的處理失敗的請求，它們對一個節(jié)點是否著（alive）”有著清晰的定義。Kafka判斷一個節(jié)點是否活著有兩個條件：
1. 節(jié)點必須可以維護和ZooKeeper的連接，Zookeeper通過心跳機制檢查每個節(jié)點的連接。
2. 如果節(jié)點是個follower,他必須能及時的同步leader的寫操作，延時不能太久。
符合以上條件的節(jié)點準確的說應該是“同步中的（in sync）”，而不是模糊的說是“活著的”或是“失敗的”。Leader會追蹤所有“同步中”的節(jié)點，一旦一個down掉了，或是卡住了，或是延時太久，leader就會把它移除。至于延時多久算是“太久”，是由參數(shù)replica.lag.max.messages決定的，怎樣算是卡住了，怎是由參數(shù)replica.lag.time.max.ms決定的。
只有當消息被所有的副本加入到日志中時，才算是“committed”，只有committed的消息才會發(fā)送給consumer，這樣就不用擔心一旦leader down掉了消息會丟失。Producer也可以選擇是否等待消息被提交的通知，這個是由參數(shù)acks決定的。
Kafka保證只要有一個“同步中”的節(jié)點，“committed”的消息就不會丟失。
?
?
- 2.3 ?kafka拓撲結(jié)構(gòu)
? ? ? ?一個典型的Kafka集群中包含若干Producer（可以是web前端FET，或者是服務器日志等），若干broker（Kafka支持水平擴展，一般broker數(shù)量越多，集群吞吐率越高），若干ConsumerGroup，以及一個Zookeeper集群。Kafka通過Zookeeper管理Kafka集群配置：選舉Kafka broker的leader，以及在Consumer Group發(fā)生變化時進行rebalance，因為consumer消費kafka topic的partition的offsite信息是存在Zookeeper的。Producer使用push模式將消息發(fā)布到broker，Consumer使用pull模式從broker訂閱并消費消息。
?
分析過程分為以下4個步驟：
- topic中partition存儲分布
- partiton中文件存儲方式 (partition在linux服務器上就是一個目錄（文件夾）)
- partiton中segment文件存儲結(jié)構(gòu)
- 在partition中如何通過offset查找message
通過上述4過程詳細分析，我們就可以清楚認識到kafka文件存儲機制的奧秘。
?
2.3 topic中partition存儲分布
假設(shè)實驗環(huán)境中Kafka集群只有一個broker，xxx/message-folder為數(shù)據(jù)文件存儲根目錄，在Kafka broker中server.properties文件配置(參數(shù)log.dirs=xxx/message-folder)，例如創(chuàng)建2個topic名 稱分別為report_push、launch_info, partitions數(shù)量都為partitions=4
存儲路徑和目錄規(guī)則為：
xxx/message-folder
? |--report_push-0
? |--report_push-1
? |--report_push-2
? |--report_push-3
? |--launch_info-0
? |--launch_info-1
? |--launch_info-2
? |--launch_info-3
?
在Kafka文件存儲中，同一個topic下有多個不同partition，每個partition為一個目錄，partiton命名規(guī)則為topic名稱+有序序號，第一個partiton序號從0開始，序號最大值為partitions數(shù)量減1。
消息發(fā)送時都被發(fā)送到一個topic，其本質(zhì)就是一個目錄，而topic由是由一些Partition組成,其組織結(jié)構(gòu)如下圖所示：
?
我們可以看到，Partition是一個Queue的結(jié)構(gòu)，每個Partition中的消息都是有序的，生產(chǎn)的消息被不斷追加到Partition上，其中的每一個消息都被賦予了一個唯一的offset值。
?
Kafka集群會保存所有的消息，不管消息有沒有被消費；我們可以設(shè)定消息的過期時間，只有過期的數(shù)據(jù)才會被自動清除以釋放磁盤空間。比如我們設(shè)置消息過期時間為2天，那么這2天內(nèi)的所有消息都會被保存到集群中，數(shù)據(jù)只有超過了兩天才會被清除。
?
Kafka只維護在Partition中的offset值，因為這個offsite標識著這個partition的message消費到哪條了。Consumer每消費一個消息，offset就會加1。其實消息的狀態(tài)完全是由Consumer控制的，Consumer可以跟蹤和重設(shè)這個offset值，這樣的話Consumer就可以讀取任意位置的消息。
?
把消息日志以Partition的形式存放有多重考慮，第一，方便在集群中擴展，每個Partition可以通過調(diào)整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應任意大小的數(shù)據(jù)了；第二就是可以提高并發(fā)，因為可以以Partition為單位讀寫了。
?
通過上面介紹的我們可以知道，kafka中的數(shù)據(jù)是持久化的并且能夠容錯的。Kafka允許用戶為每個topic設(shè)置副本數(shù)量，副本數(shù)量決定了有幾個broker來存放寫入的數(shù)據(jù)。如果你的副本數(shù)量設(shè)置為3，那么一份數(shù)據(jù)就會被存放在3臺不同的機器上，那么就允許有2個機器失敗。一般推薦副本數(shù)量至少為2，這樣就可以保證增減、重啟機器時不會影響到數(shù)據(jù)消費。如果對數(shù)據(jù)持久化有更高的要求，可以把副本數(shù)量設(shè)置為3或者更多。
?
Kafka中的topic是以partition的形式存放的，每一個topic都可以設(shè)置它的partition數(shù)量，Partition的數(shù)量決定了組成topic的message的數(shù)量。Producer在生產(chǎn)數(shù)據(jù)時，會按照一定規(guī)則（這個規(guī)則是可以自定義的）把消息發(fā)布到topic的各個partition中。上面將的副本都是以partition為單位的，不過只有一個partition的副本會被選舉成leader作為讀寫用。
?
關(guān)于如何設(shè)置partition值需要考慮的因素。一個partition只能被一個消費者消費（一個消費者可以同時消費多個partition），因此，如果設(shè)置的partition的數(shù)量小于consumer的數(shù)量，就會有消費者消費不到數(shù)據(jù)。所以，推薦partition的數(shù)量一定要大于同時運行的consumer的數(shù)量。另外一方面，建議partition的數(shù)量大于集群broker的數(shù)量，這樣leader partition就可以均勻的分布在各個broker中，最終使得集群負載均衡。在Cloudera,每個topic都有上百個partition。需要注意的是，kafka需要為每個partition分配一些內(nèi)存來緩存消息數(shù)據(jù)，如果partition數(shù)量越大，就要為kafka分配更大的heap space。
2.4 partiton中文件存儲方式
- 每個partion(目錄)相當于一個巨型文件被平均分配到多個大小相等segment(段)數(shù)據(jù)文件中。但每個段segment file消息數(shù)量不一定相等，這種特性方便old segment file快速被刪除。
- 每個partiton只需要支持順序讀寫就行了，segment文件生命周期由服務端配置參數(shù)決定。
這樣做的好處就是能快速刪除無用文件，有效提高磁盤利用率。
2.5 partiton中segment文件存儲結(jié)構(gòu)
producer發(fā)message到某個topic，message會被均勻的分布到多個partition上（隨機或根據(jù)用戶指定的回調(diào)函數(shù)進行分布），kafka broker收到message往對應partition的最后一個segment上添加該消息，當某個segment上的消息條數(shù)達到配置值或消息發(fā)布時間超過閾值時，segment上的消息會被flush到磁盤，只有flush到磁盤上的消息consumer才能消費，segment達到一定的大小后將不會再往該segment寫數(shù)據(jù)，broker會創(chuàng)建新的segment。
?
每個part在內(nèi)存中對應一個index，記錄每個segment中的第一條消息偏移。
- segment file組成：由2大部分組成，分別為index file和data file，此2個文件一一對應，成對出現(xiàn)，后綴".index"和“.log”分別表示為segment索引文件、數(shù)據(jù)文件.
- segment文件命名規(guī)則：partion全局的第一個segment從0開始，后續(xù)每個segment文件名為上一個全局partion的最大offset(偏移message數(shù))。數(shù)值最大為64位long大小，19位數(shù)字字符長度，沒有數(shù)字用0填充。
?
每個segment中存儲很多條消息，消息id由其邏輯位置決定，即從消息id可直接定位到消息的存儲位置，避免id到位置的額外映射。
下面文件列表是筆者在Kafka broker上做的一個實驗，創(chuàng)建一個topicXXX包含1 partition，設(shè)置每個segment大小為500MB,并啟動producer向Kafka broker寫入大量數(shù)據(jù),如下圖2所示segment文件列表形象說明了上述2個規(guī)則：
以上述圖2中一對segment file文件為例，說明segment中index<—->data file對應關(guān)系物理結(jié)構(gòu)如下：
上述圖3中索引文件存儲大量元數(shù)據(jù)，數(shù)據(jù)文件存儲大量消息，索引文件中元數(shù)據(jù)指向?qū)獢?shù)據(jù)文件中message的物理偏移地址。其中以索引文件中 元數(shù)據(jù)3,497為例，依次在數(shù)據(jù)文件中表示第3個message(在全局partiton表示第368772個message)、以及該消息的物理偏移 地址為497。
從上述圖3了解到segment data file由許多message組成，下面詳細說明message物理結(jié)構(gòu)如下：
參數(shù)說明：
| 8 byte offset | 在parition(分區(qū))內(nèi)的每條消息都有一個有序的id號，這個id號被稱為偏移(offset),它可以唯一確定每條消息在parition(分區(qū))內(nèi)的位置。即offset表示partiion的第多少message |
| 4 byte message size | message大小 |
| 4 byte CRC32 | 用crc32校驗message |
| 1 byte “magic" | 表示本次發(fā)布Kafka服務程序協(xié)議版本號 |
| 1 byte “attributes" | 表示為獨立版本、或標識壓縮類型、或編碼類型。 |
| 4 byte key length | 表示key的長度,當key為-1時，K byte key字段不填 |
| K byte key | 可選 |
| value bytes payload | 表示實際消息數(shù)據(jù)。 |
?
2.6 在partition中如何通過offset查找message
例如讀取offset=368776的message，需要通過下面2個步驟查找。
-
第一步查找segment file
上述圖2為例，其中00000000000000000000.index表示最開始的文件，起始偏移量(offset)為0.第二個文件 00000000000000368769.index的消息量起始偏移量為368770 = 368769 + 1.同樣，第三個文件00000000000000737337.index的起始偏移量為737338=737337 + 1，其他后續(xù)文件依次類推，以起始偏移量命名并排序這些文件，只要根據(jù)offset **二分查找**文件列表，就可以快速定位到具體文件。
當offset=368776時定位到00000000000000368769.index|log
-
第二步通過segment file查找message通過第一步定位到segment file，當offset=368776時，依次定位到00000000000000368769.index的元數(shù)據(jù)物理位置和 00000000000000368769.log的物理偏移地址，然后再通過00000000000000368769.log順序查找直到 offset=368776為止。
segment index file采取稀疏索引存儲方式，它減少索引文件大小，通過mmap可以直接內(nèi)存操作，稀疏索引為數(shù)據(jù)文件的每個對應message設(shè)置一個元數(shù)據(jù)指針,它 比稠密索引節(jié)省了更多的存儲空間，但查找起來需要消耗更多的時間。
?
kafka會記錄offset到zk中。但是，zk client api對zk的頻繁寫入是一個低效的操作。0.8.2 kafka引入了native offset storage，將offset管理從zk移出，并且可以做到水平擴展。其原理就是利用了kafka的compacted topic，offset以consumer group,topic與partion的組合作為key直接提交到compacted topic中。同時Kafka又在內(nèi)存中維護了的三元組來維護最新的offset信息，consumer來取最新offset信息的時候直接內(nèi)存里拿即可。當然，kafka允許你快速的checkpoint最新的offset信息到磁盤上。
?
3.Partition Replication原則
Kafka高效文件存儲設(shè)計特點
- Kafka把topic中一個parition大文件分成多個小文件段，通過多個小文件段，就容易定期清除或刪除已經(jīng)消費完文件，減少磁盤占用。
- 通過索引信息可以快速定位message和確定response的最大大小。
- 通過index元數(shù)據(jù)全部映射到memory，可以避免segment file的IO磁盤操作。
- 通過索引文件稀疏存儲，可以大幅降低index文件元數(shù)據(jù)占用空間大小。
?
?
1. Kafka集群partition?replication默認自動分配分析
下面以一個Kafka集群中4個Broker舉例，創(chuàng)建1個topic包含4個Partition，2 Replication；數(shù)據(jù)Producer流動如圖所示：
(1)
?
?
(2)當集群中新增2節(jié)點，Partition增加到6個時分布情況如下：
?
副本分配邏輯規(guī)則如下：
- 在Kafka集群中，每個Broker都有均等分配Partition的Leader機會。
- 上述圖Broker Partition中，箭頭指向為副本，以Partition-0為例:broker1中parition-0為Leader，Broker2中Partition-0為副本。
- 上述圖種每個Broker(按照BrokerId有序)依次分配主Partition,下一個Broker為副本，如此循環(huán)迭代分配，多副本都遵循此規(guī)則。
?
副本分配算法如下：
- 將所有N Broker和待分配的i個Partition排序.
- 將第i個Partition分配到第(i mod n)個Broker上.
- 將第i個Partition的第j個副本分配到第((i + j) mod n)個Broker上.
?
4.Kafka Broker一些特性
4.1 無狀態(tài)的Kafka Broker :
1. Broker沒有副本機制，一旦broker宕機，該broker的消息將都不可用。
2. Broker不保存訂閱者的狀態(tài)，由訂閱者自己保存。
3. 無狀態(tài)導致消息的刪除成為難題（可能刪除的消息正在被訂閱），kafka采用基于時間的SLA(服務水平保證)，消息保存一定時間（通常為7天）后會被刪除。
4. 消息訂閱者可以rewind back到任意位置重新進行消費，當訂閱者故障時，可以選擇最小的offset進行重新讀取消費消息。
?
4.2 message的交付與生命周期 ：
1. 不是嚴格的JMS， 因此kafka對消息的重復、丟失、錯誤以及順序型沒有嚴格的要求。（這是與AMQ最大的區(qū)別）
2. kafka提供at-least-once delivery,即當consumer宕機后，有些消息可能會被重復delivery。
3. 因每個partition只會被consumer group內(nèi)的一個consumer消費，故kafka保證每個partition內(nèi)的消息會被順序的訂閱。
4. Kafka為每條消息為每條消息計算CRC校驗，用于錯誤檢測，crc校驗不通過的消息會直接被丟棄掉。
?
4.3 壓縮
?
Kafka支持以集合（batch）為單位發(fā)送消息，在此基礎(chǔ)上，Kafka還支持對消息集合進行壓縮，Producer端可以通過GZIP或Snappy格式對消息集合進行壓縮。Producer端進行壓縮之后，在Consumer端需進行解壓。壓縮的好處就是減少傳輸?shù)臄?shù)據(jù)量，減輕對網(wǎng)絡傳輸?shù)膲毫?#xff0c;在對大數(shù)據(jù)處理上，瓶頸往往體現(xiàn)在網(wǎng)絡上而不是CPU。
?
那么如何區(qū)分消息是壓縮的還是未壓縮的呢，Kafka在消息頭部添加了一個描述壓縮屬性字節(jié)，這個字節(jié)的后兩位表示消息的壓縮采用的編碼，如果后兩位為0，則表示消息未被壓縮。
?
4.4 消息可靠性
?
在消息系統(tǒng)中，保證消息在生產(chǎn)和消費過程中的可靠性是十分重要的，在實際消息傳遞過程中，可能會出現(xiàn)如下三中情況：
?
- 一個消息發(fā)送失敗
?
- 一個消息被發(fā)送多次
?
- 最理想的情況：exactly-once ,一個消息發(fā)送成功且僅發(fā)送了一次
?
有許多系統(tǒng)聲稱它們實現(xiàn)了exactly-once，但是它們其實忽略了生產(chǎn)者或消費者在生產(chǎn)和消費過程中有可能失敗的情況。比如雖然一個Producer成功發(fā)送一個消息，但是消息在發(fā)送途中丟失，或者成功發(fā)送到broker，也被consumer成功取走，但是這個consumer在處理取過來的消息時失敗了。
?
從Producer端看：Kafka是這么處理的，當一個消息被發(fā)送后，Producer會等待broker成功接收到消息的反饋（可通過參數(shù)控制等待時間），如果消息在途中丟失或是其中一個broker掛掉，Producer會重新發(fā)送（我們知道Kafka有備份機制，可以通過參數(shù)控制是否等待所有備份節(jié)點都收到消息）。
?
從Consumer端看：前面講到過partition，broker端記錄了partition中的一個offset值，這個值指向Consumer下一個即將消費message。當Consumer收到了消息，但卻在處理過程中掛掉，此時Consumer可以通過這個offset值重新找到上一個消息再進行處理。Consumer還有權(quán)限控制這個offset值，對持久化到broker端的消息做任意處理。
?
4.5 備份機制
?
備份機制是Kafka0.8版本的新特性，備份機制的出現(xiàn)大大提高了Kafka集群的可靠性、穩(wěn)定性。有了備份機制后，Kafka允許集群中的節(jié)點掛掉后而不影響整個集群工作。一個備份數(shù)量為n的集群允許n-1個節(jié)點失敗。在所有備份節(jié)點中，有一個節(jié)點作為lead節(jié)點，這個節(jié)點保存了其它備份節(jié)點列表，并維持各個備份間的狀體同步。下面這幅圖解釋了Kafka的備份機制:
?
4.6 Kafka高效性相關(guān)設(shè)計
?
4.6.1 消息的持久化
Kafka高度依賴文件系統(tǒng)來存儲和緩存消息(AMQ的nessage是持久化到mysql數(shù)據(jù)庫中的)，因為一般的人認為磁盤是緩慢的，這導致人們對持久化結(jié)構(gòu)具有競爭性持懷疑態(tài)度。其實，磁盤的快或者慢，這決定于我們?nèi)绾问褂么疟P。因為磁盤線性寫的速度遠遠大于隨機寫。線性讀寫在大多數(shù)應用場景下是可以預測的。
4.6.2 常數(shù)時間性能保證
每個Topic的Partition的是一個大文件夾，里面有無數(shù)個小文件夾segment，但partition是一個隊列，隊列中的元素是segment,消費的時候先從第0個segment開始消費，新來message存在最后一個消息隊列中。對于segment也是對隊列，隊列元素是message,有對應的offsite標識是哪個message。消費的時候先從這個segment的第一個message開始消費，新來的message存在segment的最后。
?
消息系統(tǒng)的持久化隊列可以構(gòu)建在對一個文件的讀和追加上，就像一般情況下的日志解決方案。它有一個優(yōu)點，所有的操作都是常數(shù)時間，并且讀寫之間不會相互阻塞。這種設(shè)計具有極大的性能優(yōu)勢：最終系統(tǒng)性能和數(shù)據(jù)大小完全無關(guān)，服務器可以充分利用廉價的硬盤來提供高效的消息服務。
?
事實上還有一點，磁盤空間的無限增大而不影響性能這點，意味著我們可以提供一般消息系統(tǒng)無法提供的特性。比如說，消息被消費后不是立馬被刪除，我們可以將這些消息保留一段相對比較長的時間（比如一個星期）。
?
5.Kafka 生產(chǎn)者-消費者
?????消息系統(tǒng)通常都會由生產(chǎn)者，消費者，Broker三大部分組成，生產(chǎn)者會將消息寫入到Broker，消費者會從Broker中讀取出消息，不同的MQ實現(xiàn)的Broker實現(xiàn)會有所不同，不過Broker的本質(zhì)都是要負責將消息落地到服務端的存儲系統(tǒng)中。具體步驟如下：
生產(chǎn)者客戶端應用程序產(chǎn)生消息：
客戶端連接對象將消息包裝到請求中發(fā)送到服務端
服務端的入口也有一個連接對象負責接收請求，并將消息以文件的形式存儲起來
服務端返回響應結(jié)果給生產(chǎn)者客戶端
消費者客戶端應用程序消費消息：
客戶端連接對象將消費信息也包裝到請求中發(fā)送給服務端
服務端從文件存儲系統(tǒng)中取出消息
服務端返回響應結(jié)果給消費者客戶端
客戶端將響應結(jié)果還原成消息并開始處理消息
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? 圖4-1?客戶端和服務端交互
?
5.1 ?Producers
?
Producers直接發(fā)送消息到broker上的leader partition，不需要經(jīng)過任何中介或其他路由轉(zhuǎn)發(fā)。為了實現(xiàn)這個特性，kafka集群中的每個broker都可以響應producer的請求，并返回topic的一些元信息，這些元信息包括哪些機器是存活的，topic的leader partition都在哪，現(xiàn)階段哪些leader partition是可以直接被訪問的。
?
Producer客戶端自己控制著消息被推送到哪些partition。實現(xiàn)的方式可以是隨機分配、實現(xiàn)一類隨機負載均衡算法，或者指定一些分區(qū)算法。Kafka提供了接口供用戶實現(xiàn)自定義的partition，用戶可以為每個消息指定一個partitionKey，通過這個key來實現(xiàn)一些hash分區(qū)算法。比如，把userid作為partitionkey的話，相同userid的消息將會被推送到同一個partition。
?
以Batch的方式推送數(shù)據(jù)可以極大的提高處理效率，kafka Producer 可以將消息在內(nèi)存中累計到一定數(shù)量后作為一個batch發(fā)送請求。Batch的數(shù)量大小可以通過Producer的參數(shù)控制，參數(shù)值可以設(shè)置為累計的消息的數(shù)量（如500條）、累計的時間間隔（如100ms）或者累計的數(shù)據(jù)大小(64KB)。通過增加batch的大小，可以減少網(wǎng)絡請求和磁盤IO的次數(shù)，當然具體參數(shù)設(shè)置需要在效率和時效性方面做一個權(quán)衡。
?
Producers可以異步的并行的向kafka發(fā)送消息，但是通常producer在發(fā)送完消息之后會得到一個future響應，返回的是offset值或者發(fā)送過程中遇到的錯誤。這其中有個非常重要的參數(shù)“acks”,這個參數(shù)決定了producer要求leader partition 收到確認的副本個數(shù)，如果acks設(shè)置數(shù)量為0，表示producer不會等待broker的響應，所以，producer無法知道消息是否發(fā)送成功，這樣有可能會導致數(shù)據(jù)丟失，但同時，acks值為0會得到最大的系統(tǒng)吞吐量。
?
若acks設(shè)置為1，表示producer會在leader partition收到消息時得到broker的一個確認，這樣會有更好的可靠性，因為客戶端會等待直到broker確認收到消息。若設(shè)置為-1，producer會在所有備份的partition收到消息時得到broker的確認，這個設(shè)置可以得到最高的可靠性保證。
?
Kafka 消息有一個定長的header和變長的字節(jié)數(shù)組組成。因為kafka消息支持字節(jié)數(shù)組，也就使得kafka可以支持任何用戶自定義的序列號格式或者其它已有的格式如Apache Avro、protobuf等。Kafka沒有限定單個消息的大小，但我們推薦消息大小不要超過1MB,通常一般消息大小都在1~10kB之前。
?
發(fā)布消息時，kafka client先構(gòu)造一條消息，將消息加入到消息集set中（kafka支持批量發(fā)布，可以往消息集合中添加多條消息，一次行發(fā)布），send消息時，producer client需指定消息所屬的topic。
?
5.2 ?Consumers
Kafka提供了兩套consumer api，分為high-level api和sample-api。Sample-api 是一個底層的API，它維持了一個和單一broker的連接，并且這個API是完全無狀態(tài)的，每次請求都需要指定offset值，因此，這套API也是最靈活的。
?
在kafka中，當前讀到哪條消息的offset值是由consumer來維護的，因此，consumer可以自己決定如何讀取kafka中的數(shù)據(jù)。比如，consumer可以通過重設(shè)offset值來重新消費已消費過的數(shù)據(jù)。不管有沒有被消費，kafka會保存數(shù)據(jù)一段時間，這個時間周期是可配置的，只有到了過期時間，kafka才會刪除這些數(shù)據(jù)。（這一點與AMQ不一樣，AMQ的message一般來說都是持久化到mysql中的，消費完的message會被delete掉）
?
High-level API封裝了對集群中一系列broker的訪問，可以透明的消費一個topic。它自己維持了已消費消息的狀態(tài)，即每次消費的都是下一個消息。
?
High-level API還支持以組的形式消費topic，如果consumers有同一個組名，那么kafka就相當于一個隊列消息服務，而各個consumer均衡的消費相應partition中的數(shù)據(jù)。若consumers有不同的組名，那么此時kafka就相當與一個廣播服務，會把topic中的所有消息廣播到每個consumer。
?
High level api和Low level api是針對consumer而言的，和producer無關(guān)。
?
High level api是consumer讀的partition的offsite是存在zookeeper上。High level api?會啟動另外一個線程去每隔一段時間，offsite自動同步到zookeeper上。換句話說，如果使用了High level api， 每個message只能被讀一次，一旦讀了這條message之后，無論我consumer的處理是否ok。High level api的另外一個線程會自動的把offiste+1同步到zookeeper上。如果consumer讀取數(shù)據(jù)出了問題，offsite也會在zookeeper上同步。因此，如果consumer處理失敗了，會繼續(xù)執(zhí)行下一條。這往往是不對的行為。因此，Best Practice是一旦consumer處理失敗，直接讓整個conusmer group拋Exception終止，但是最后讀的這一條數(shù)據(jù)是丟失了，因為在zookeeper里面的offsite已經(jīng)+1了。等再次啟動conusmer group的時候，已經(jīng)從下一條開始讀取處理了。
?
Low level api是consumer讀的partition的offsite在consumer自己的程序中維護。不會同步到zookeeper上。但是為了kafka manager能夠方便的監(jiān)控，一般也會手動的同步到zookeeper上。這樣的好處是一旦讀取某個message的consumer失敗了，這條message的offsite我們自己維護，我們不會+1。下次再啟動的時候，還會從這個offsite開始讀。這樣可以做到exactly once對于數(shù)據(jù)的準確性有保證。
?
?
對于Consumer group：
1. 允許consumer group（包含多個consumer，如一個集群同時消費）對一個topic進行消費，不同的consumer group之間獨立消費。
2. 為了對減小一個consumer group中不同consumer之間的分布式協(xié)調(diào)開銷，指定partition為最小的并行消費單位，即一個group內(nèi)的consumer只能消費不同的partition。
?
Consumer與Partition的關(guān)系：
- 如果consumer比partition多，是浪費，因為kafka的設(shè)計是在一個partition上是不允許并發(fā)的，所以consumer數(shù)不要大于partition數(shù)
- 如果consumer比partition少，一個consumer會對應于多個partitions，這里主要合理分配consumer數(shù)和partition數(shù)，否則會導致partition里面的數(shù)據(jù)被取的不均勻
- 如果consumer從多個partition讀到數(shù)據(jù)，不保證數(shù)據(jù)間的順序性，kafka只保證在一個partition上數(shù)據(jù)是有序的，但多個partition，根據(jù)你讀的順序會有不同
- 增減consumer，broker，partition會導致rebalance，所以rebalance后consumer對應的partition會發(fā)生變化
- High-level接口中獲取不到數(shù)據(jù)的時候是會block的
?
負載低的情況下可以每個線程消費多個partition。但負載高的情況下，Consumer 線程數(shù)最好和Partition數(shù)量保持一致。如果還是消費不過來，應該再開 Consumer 進程，進程內(nèi)線程數(shù)同樣和分區(qū)數(shù)一致。
?
消費消息時，kafka client需指定topic以及partition number（每個partition對應一個邏輯日志流，如topic代表某個產(chǎn)品線，partition代表產(chǎn)品線的日志按天切分的結(jié)果），consumer client訂閱后，就可迭代讀取消息，如果沒有消息，consumer client會阻塞直到有新的消息發(fā)布。consumer可以累積確認接收到的消息，當其確認了某個offset的消息，意味著之前的消息也都已成功接收到，此時broker會更新zookeeper上地offset registry。
?
5.3??高效的數(shù)據(jù)傳輸
1.??發(fā)布者每次可發(fā)布多條消息（將消息加到一個消息集合中發(fā)布），?consumer每次迭代消費一條消息。
2.??不創(chuàng)建單獨的cache，使用系統(tǒng)的page cache。發(fā)布者順序發(fā)布，訂閱者通常比發(fā)布者滯后一點點，直接使用Linux的page cache效果也比較后，同時減少了cache管理及垃圾收集的開銷。
3.??使用sendfile優(yōu)化網(wǎng)絡傳輸，減少一次內(nèi)存拷貝。
?
6.Kafka 與 Zookeeper
?
6.1 Zookeeper?協(xié)調(diào)控制
1.?管理broker與consumer的動態(tài)加入與離開。(Producer不需要管理，隨便一臺計算機都可以作為Producer向Kakfa Broker發(fā)消息)
2.?觸發(fā)負載均衡，當broker或consumer加入或離開時會觸發(fā)負載均衡算法，使得一
? ?個consumer group內(nèi)的多個consumer的消費負載平衡。（因為一個comsumer消費一個或多個partition，一個partition只能被一個consumer消費）
3.??維護消費關(guān)系及每個partition的消費信息。
?
6.2 Zookeeper上的細節(jié)：
1.?每個broker啟動后會在zookeeper上注冊一個臨時的broker registry，包含broker的ip地址和端口號，所存儲的topics和partitions信息。
2.?每個consumer啟動后會在zookeeper上注冊一個臨時的consumer registry：包含consumer所屬的consumer group以及訂閱的topics。
3.?每個consumer group關(guān)聯(lián)一個臨時的owner registry和一個持久的offset registry。對于被訂閱的每個partition包含一個owner registry，內(nèi)容為訂閱這個partition的consumer id；同時包含一個offset registry，內(nèi)容為上一次訂閱的offset。
總結(jié)
以上是生活随笔為你收集整理的Kafka史上最详细总结的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 万能makefile深入浅出 - 第四篇
- 下一篇： C语言rand(),srand()函数真