kafka原理

Kafka是最初由Linkedin公司開發，是一個分佈式、支持分區的（partition）、多副本的（replica），基於zookeeper協調的分佈式消息系統，它的最大的特性就是能夠實時的處理大量數據以知足各類需求場景：好比基於hadoop的批處理系統、低延遲的實時系統、storm/Spark流式處理引擎，web/nginx日誌、訪問日誌，消息服務等等，用scala語言編寫，Linkedin於2010年貢獻給了Apache基金會併成爲頂級開源項目。

1.前言
消息隊列的性能好壞，其文件存儲機制設計是衡量一個消息隊列服務技術水平和最關鍵指標之一。下面將從Kafka文件存儲機制和物理結構角度，分析Kafka是如何實現高效文件存儲，及實際應用效果。

1.1 Kafka的特性:

• 高吞吐量、低延遲：kafka每秒能夠處理幾十萬條消息，它的延遲最低只有幾毫秒，每一個topic能夠分多個partition, consumer group 對partition進行consume操做。
• 可擴展性：kafka集羣支持熱擴展
• 持久性、可靠性：消息被持久化到本地磁盤，而且支持數據備份防止數據丟失
• 容錯性：容許集羣中節點失敗（若副本數量爲n,則容許n-1個節點失敗）
• 高併發：支持數千個客戶端同時讀寫

1.2 Kafka的使用場景：

• 日誌收集：一個公司能夠用Kafka能夠收集各類服務的log，經過kafka以統一接口服務的方式開放給各類consumer，例如hadoop、Hbase、Solr等。
• 消息系統：解耦和生產者和消費者、緩存消息等。
• 用戶活動跟蹤：Kafka常常被用來記錄web用戶或者app用戶的各類活動，如瀏覽網頁、搜索、點擊等活動，這些活動信息被各個服務器發佈到kafka的topic中，而後訂閱者經過訂閱這些topic來作實時的監控分析，或者裝載到hadoop、數據倉庫中作離線分析和挖掘。
• 運營指標：Kafka也常常用來記錄運營監控數據。包括收集各類分佈式應用的數據，生產各類操做的集中反饋，好比報警和報告。
• 流式處理：好比spark streaming和storm

1.3 Kakfa的設計思想

• Kakfa Broker Leader的選舉：Kakfa Broker集羣受Zookeeper管理。全部的Kafka Broker節點一塊兒去Zookeeper上註冊一個臨時節點，由於只有一個Kafka Broker會註冊成功，其餘的都會失敗，因此這個成功在Zookeeper上註冊臨時節點的這個Kafka Broker會成爲Kafka Broker Controller，其餘的Kafka broker叫Kafka Broker follower。（這個過程叫Controller在ZooKeeper註冊Watch）。這個Controller會監聽其餘的Kafka Broker的全部信息，若是這個kafka broker controller宕機了，在zookeeper上面的那個臨時節點就會消失，此時全部的kafka broker又會一塊兒去Zookeeper上註冊一個臨時節點，由於只有一個Kafka Broker會註冊成功，其餘的都會失敗，因此這個成功在Zookeeper上註冊臨時節點的這個Kafka Broker會成爲Kafka Broker Controller，其餘的Kafka broker叫Kafka Broker follower。例如：一旦有一個broker宕機了，這個kafka broker controller會讀取該宕機broker上全部的partition在zookeeper上的狀態，並選取ISR列表中的一個replica做爲partition leader（若是ISR列表中的replica全掛，選一個倖存的replica做爲leader; 若是該partition的全部的replica都宕機了，則將新的leader設置爲-1，等待恢復，等待ISR中的任一個Replica「活」過來，而且選它做爲Leader；或選擇第一個「活」過來的Replica（不必定是ISR中的）做爲Leader），這個broker宕機的事情，kafka controller也會通知zookeeper，zookeeper就會通知其餘的kafka broker。
• 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。 當啓動一個consumer group去消費一個topic的時候，不管topic裏面有多個少個partition，不管咱們consumer group裏面配置了多少個consumer thread，這個consumer group下面的全部consumer thread必定會消費所有的partition；即使這個consumer group下只有一個consumer thread，那麼這個consumer thread也會去消費全部的partition。所以，最優的設計就是，consumer group下的consumer thread的數量等於partition數量，這樣效率是最高的。 同一partition的一條message只能被同一個Consumer Group內的一個Consumer消費。不可以一個consumer group的多個consumer同時消費一個partition。
• Consumer Rebalance的觸發條件：(1): Consumer增長或刪除會觸發 Consumer Group的Rebalance（2）Broker的增長或者減小都會觸發 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已經+1，這個時候就會丟message；也能夠配置成讀完消息處理再commit，這種狀況下consumer端的響應就會比較慢的，須要等處理完才行。若是producer的流量增大，當前的topic的parition數量=consumer數量，這時候的應對方式就是很想擴展：增長topic下的partition，同時增長這個consumer group下的consumer。
  
• Delivery Mode: Kafka producer 發送message不用維護message的offsite信息，由於這個時候，offsite就至關於一個自增id，producer就儘管發送message就行了。可是Consumer端是須要維護這個partition當前消費到哪一個message的offsite信息的，這個offsite信息，high level api是維護在Zookeeper上，low level api是本身的程序維護。當使用high level api的時候，先拿message處理，再定時自動commit offsite+1（也能夠改爲手動）, 而且kakfa處理message是沒有鎖操做的。所以若是處理message失敗，此時尚未commit offsite+1，當consumer thread重啓後會重複消費這個message。可是做爲高吞吐量高併發的實時處理系統，at least once的狀況下，至少一次會被處理到，是能夠容忍的。若是沒法容忍，就得使用low level api來本身程序維護這個offsite信息，那麼想何時commit offsite+1就本身搞定了。
• Topic & Partition：Topic至關於傳統消息系統MQ中的一個隊列queue，producer端發送的message必須指定是發送到哪一個topic，可是不須要指定topic下的哪一個partition，由於kafka會把收到的message進行load balance，均勻的分佈在這個topic下的不一樣的partition上（ hash(message) % [broker數量] ）。 在物理結構上，每一個partition對應一個物理的目錄（文件夾），文件夾命名是[topicname]_[partition]_[序號]，一個topic能夠有無數多的partition，根據業務需求和數據量來設置。 在kafka配置文件中可隨時更高num.partitions參數來配置更改topic的partition數量，在建立Topic時經過參數指定parittion數量。Topic建立以後經過Kafka提供的工具也能夠修改partiton數量。 通常來講，（1）一個Topic的Partition數量大於等於Broker的數量，能夠提升吞吐率。（2）同一個Partition的Replica儘可能分散到不一樣的機器，高可用。 當add a new partition的時候，partition裏面的message不會從新進行分配，原來的partition裏面的message數據不會變，新加的這個partition剛開始是空的，隨後進入這個topic的message就會從新參與全部partition的load balance。
• Partition Replica：每一個partition能夠在其餘的kafka broker節點上存副本，以便某個kafka broker節點宕機不會影響這個kafka集羣。存replica副本的方式是按照kafka broker的順序存。例若有5個kafka broker節點，某個topic有3個partition，每一個partition存2個副本，那麼partition1存broker1,broker2，partition2存broker2,broker3。。。以此類推（replica副本數目不能大於kafka broker節點的數目，不然報錯。這裏的replica數其實就是partition的副本總數，其中包括一個leader，其餘的就是copy副本）。這樣若是某個broker宕機，其實整個kafka內數據依然是完整的。可是，replica副本數越高，系統雖然越穩定，可是回來帶資源和性能上的降低；replica副本少的話，也會形成系統丟數據的風險。
  （1）怎樣傳送消息：producer先把message發送到partition leader，再由leader發送給其餘partition follower。
  （2）在向Producer發送ACK前須要保證有多少個Replica已經收到該消息：根據ack配的個數而定。
  （3）怎樣處理某個Replica不工做的狀況：若是這個部工做的partition replica不在ack列表中，就是producer在發送消息到partition leader上，partition leader向partition follower發送message沒有響應而已，這個不會影響整個系統，也不會有什麼問題。若是這個不工做的partition replica在ack列表中的話，producer發送的message的時候會等待這個不工做的partition replca寫message成功，可是會等到time out，而後返回失敗由於某個ack列表中的partition replica沒有響應，此時kafka會自動的把這個部工做的partition replica從ack列表中移除，之後的producer發送message的時候就不會有這個ack列表下的這個部工做的partition replica了。
  （4）怎樣處理Failed Replica恢復回來的狀況：若是這個partition replica以前不在ack列表中，那麼啓動後從新受Zookeeper管理便可，以後producer發送message的時候，partition leader會繼續發送message到這個partition follower上。若是這個partition replica以前在ack列表中，此時重啓後，須要把這個partition replica再手動加到ack列表中。（ack列表是手動添加的，出現某個部工做的partition replica的時候自動從ack列表中移除的）
• Partition leader與follower：partition也有leader和follower之分。leader是主partition，producer寫kafka的時候先寫partition leader，再由partition leader push給其餘的partition follower。partition leader與follower的信息受Zookeeper控制，一旦partition leader所在的broker節點宕機，zookeeper會衝其餘的broker的partition follower上選擇follower變爲parition 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上。
• Partition ack：當ack=1，表示producer寫partition leader成功後，broker就返回成功，不管其餘的partition follower是否寫成功。當ack=2，表示producer寫partition leader和其餘一個follower成功的時候，broker就返回成功，不管其餘的partition follower是否寫成功。當ack=-1[parition的數量]的時候，表示只有producer所有寫成功的時候，纔算成功，kafka broker才返回成功信息。這裏須要注意的是，若是ack=1的時候，一旦有個broker宕機致使partition的follower和leader切換，會致使丟數據。
  
• message狀態：在Kafka中，消息的狀態被保存在consumer中，broker不會關心哪一個消息被消費了被誰消費了，只記錄一個offset值（指向partition中下一個要被消費的消息位置），這就意味着若是consumer處理很差的話，broker上的一個消息可能會被消費屢次。
• message持久化：Kafka中會把消息持久化到本地文件系統中，而且保持o(1)極高的效率。咱們衆所周知IO讀取是很是耗資源的性能也是最慢的，這就是爲了數據庫的瓶頸常常在IO上，須要換SSD硬盤的緣由。可是Kafka做爲吞吐量極高的MQ，卻能夠很是高效的message持久化到文件。這是由於Kafka是順序寫入o（1）的時間複雜度，速度很是快。也是高吞吐量的緣由。因爲message的寫入持久化是順序寫入的，所以message在被消費的時候也是按順序被消費的，保證partition的message是順序消費的。通常的機器,單機每秒100k條數據。
• message有效期：不一樣的版本不同，目前默認保存7天。
• Produer：Producer向Topic發送message，不須要指定partition，直接發送就行了。kafka經過partition ack來控制是否發送成功並把信息返回給producer，producer能夠有任意多的thread，這些kafka服務器端是不care的。Producer端的delivery guarantee默認是At least once的。也能夠設置Producer異步發送實現At most once。Producer能夠用主鍵冪等性實現Exactly once。
• Kafka高吞吐量: Kafka的高吞吐量體如今讀寫上，分佈式併發的讀和寫都很是快，寫的性能體如今以o(1)的時間複雜度進行順序寫入。讀的性能體如今以o(1)的時間複雜度進行順序讀取， 對topic進行partition分區，consume group中的consume線程能夠以很高能性能進行順序讀。
• Kafka delivery guarantee(message傳送保證)：（1）At most once消息可能會丟，絕對不會重複傳輸；（2）At least once 消息絕對不會丟，可是可能會重複傳輸；（3）Exactly once每條信息確定會被傳輸一次且僅傳輸一次，這是用戶想要的。
• 冗餘: replica有多個副本，保證一個broker node宕機後不會影響整個服務。
• 擴展性: broker節點能夠水平擴展，partition也能夠水平增長，partition replica也能夠水平增長。
• 峯值: 在訪問量劇增的狀況下，kafka水平擴展, 應用仍然須要繼續發揮做用。
• 可恢復性: 系統的一部分組件失效時，因爲有partition的replica副本，不會影響到整個系統。
• 順序保證性：因爲kafka的producer的寫message與consumer去讀message都是順序的讀寫，保證了高效的性能。
• 緩衝：因爲producer那面可能業務很簡單，然後端consumer業務會很複雜並有數據庫的操做，所以確定是producer會比consumer處理速度快，若是沒有kafka，producer直接調用consumer，那麼就會形成整個系統的處理速度慢，加一層kafka做爲MQ，能夠起到緩衝的做用。
• 異步通訊：做爲MQ，Producer與Consumer異步通訊。

---

2：kafka一些原理概念

• 持久化：kafka使用文件存儲消息(append only log),這就直接決定kafka在性能上嚴重依賴文件系統的自己特性.且不管任何OS下,對文件系統自己的優化是很是艱難的.文件緩存/直接內存映射等是經常使用的手段.由於kafka是對日誌文件進行append操做,所以磁盤檢索的開支是較小的;同時爲了減小磁盤寫入的次數,broker會將消息暫時buffer起來,當消息的個數(或尺寸)達到必定閥值時,再flush到磁盤,這樣減小了磁盤IO調用的次數.對於kafka而言,較高性能的磁盤,將會帶來更加直接的性能提高.
• 性能：除磁盤IO以外,咱們還須要考慮網絡IO,這直接關係到kafka的吞吐量問題.kafka並無提供太多高超的技巧;對於producer端,能夠將消息buffer起來,當消息的條數達到必定閥值時,批量發送給broker;對於consumer端也是同樣,批量fetch多條消息.不過消息量的大小能夠經過配置文件來指定.對於kafka broker端,彷佛有個sendfile系統調用能夠潛在的提高網絡IO的性能:將文件的數據映射到系統內存中,socket直接讀取相應的內存區域便可,而無需進程再次copy和交換(這裏涉及到"磁盤IO數據"/"內核內存"/"進程內存"/"網絡緩衝區",多者之間的數據copy)。
  其實對於producer/consumer/broker三者而言,CPU的開支應該都不大,所以啓用消息壓縮機制是一個良好的策略;壓縮須要消耗少許的CPU資源,不過對於kafka而言,網絡IO更應該須要考慮.能夠將任何在網絡上傳輸的消息都通過壓縮.kafka支持gzip/snappy等多種壓縮方式
• 負載均衡：kafka集羣中的任何一個broker,均可以向producer提供metadata信息,這些metadata中包含"集羣中存活的servers列表"/"partitions leader列表"等信息(請參看zookeeper中的節點信息). 當producer獲取到metadata信息以後, producer將會和Topic下全部partition leader保持socket鏈接;消息由producer直接經過socket發送到broker,中間不會通過任何"路由層".
  異步發送，將多條消息暫且在客戶端buffer起來,並將他們批量發送到broker;小數據IO太多,會拖慢總體的網絡延遲,批量延遲發送事實上提高了網絡效率;不過這也有必定的隱患,好比當producer失效時,那些還沒有發送的消息將會丟失。
• Topic模型：其餘JMS實現,消息消費的位置是有prodiver保留,以便避免重複發送消息或者將沒有消費成功的消息重發等,同時還要控制消息的狀態.這就要求JMS broker須要太多額外的工做.在kafka中,partition中的消息只有一個consumer在消費,且不存在消息狀態的控制,也沒有複雜的消息確認機制,可見kafka broker端是至關輕量級的.當消息被consumer接收以後,consumer能夠在本地保存最後消息的offset,並間歇性的向zookeeper註冊offset.因而可知,consumer客戶端也很輕量級。
  kafka中consumer負責維護消息的消費記錄,而broker則不關心這些,這種設計不只提升了consumer端的靈活性,也適度的減輕了broker端設計的複雜度;這是和衆多JMS prodiver的區別.此外,kafka中消息ACK的設計也和JMS有很大不一樣,kafka中的消息是批量(一般以消息的條數或者chunk的尺寸爲單位)發送給consumer,當消息消費成功後,向zookeeper提交消息的offset,而不會向broker交付ACK.或許你已經意識到,這種"寬鬆"的設計,將會有"丟失"消息/"消息重發"的危險。
• 消息傳輸一致：Kafka提供3種消息傳輸一致性語義：最多1次，最少1次，剛好1次。
  最少1次：可能會重傳數據，有可能出現數據被重複處理的狀況;
  最多1次：可能會出現數據丟失狀況;
  剛好1次：並非指真正只傳輸1次，只不過有一個機制。確保不會出現「數據被重複處理」和「數據丟失」的狀況。
  at most once: 消費者fetch消息,而後保存offset,而後處理消息;當client保存offset以後,可是在消息處理過程當中consumer進程失效(crash),致使部分消息未能繼續處理.那麼此後可能其餘consumer會接管,可是由於offset已經提早保存,那麼新的consumer將不能fetch到offset以前的消息(儘管它們尚沒有被處理),這就是"at most once".
  at least once: 消費者fetch消息,而後處理消息,而後保存offset.若是消息處理成功以後,可是在保存offset階段zookeeper異常或者consumer失效,致使保存offset操做未能執行成功,這就致使接下來再次fetch時可能得到上次已經處理過的消息,這就是"at least once".
  "Kafka Cluster"到消費者的場景中能夠採起如下方案來獲得「剛好1次」的一致性語義：最少1次＋消費者的輸出中額外增長已處理消息最大編號：因爲已處理消息最大編號的存在，不會出現重複處理消息的狀況。
• 副本：kafka中,replication策略是基於partition,而不是topic;kafka將每一個partition數據複製到多個server上,任何一個partition有一個leader和多個follower(能夠沒有);備份的個數能夠經過broker配置文件來設定。leader處理全部的read-write請求,follower須要和leader保持同步.Follower就像一個"consumer",消費消息並保存在本地日誌中;leader負責跟蹤全部的follower狀態,若是follower"落後"太多或者失效,leader將會把它從replicas同步列表中刪除.當全部的follower都將一條消息保存成功,此消息才被認爲是"committed",那麼此時consumer才能消費它,這種同步策略,就要求follower和leader之間必須具備良好的網絡環境.即便只有一個replicas實例存活,仍然能夠保證消息的正常發送和接收,只要zookeeper集羣存活便可.
  選擇follower時須要兼顧一個問題,就是新leader server上所已經承載的partition leader的個數,若是一個server上有過多的partition leader,意味着此server將承受着更多的IO壓力.在選舉新leader,須要考慮到"負載均衡",partition leader較少的broker將會更有可能成爲新的leader.
• 分佈式：kafka使用zookeeper來存儲一些meta信息,並使用了zookeeper watch機制來發現meta信息的變動並做出相應的動做(好比consumer失效,觸發負載均衡等)。
  Broker node registry: 當一個kafka broker啓動後,首先會向zookeeper註冊本身的節點信息(臨時znode),同時當broker和zookeeper斷開鏈接時,此znode也會被刪除.
  Broker Topic Registry: 當一個broker啓動時,會向zookeeper註冊本身持有的topic和partitions信息,仍然是一個臨時znode.
  Consumer and Consumer group: 每一個consumer客戶端被建立時,會向zookeeper註冊本身的信息;此做用主要是爲了"負載均衡".一個group中的多個consumer能夠交錯的消費一個topic的全部partitions;簡而言之,保證此topic的全部partitions都能被此group所消費,且消費時爲了性能考慮,讓partition相對均衡的分散到每一個consumer上.
  Consumer id Registry: 每一個consumer都有一個惟一的ID(host:uuid,能夠經過配置文件指定,也能夠由系統生成),此id用來標記消費者信息.
  Consumer offset Tracking: 用來跟蹤每一個consumer目前所消費的partition中最大的offset.此znode爲持久節點,能夠看出offset跟group_id有關,以代表當group中一個消費者失效,其餘consumer能夠繼續消費.
  Partition Owner registry: 用來標記partition正在被哪一個consumer消費.臨時znode。此節點表達了"一個partition"只能被group下一個consumer消費,同時當group下某個consumer失效,那麼將會觸發負載均衡(即:讓partitions在多個consumer間均衡消費,接管那些"遊離"的partitions)。
  當consumer啓動時,所觸發的操做:
  A) 首先進行"Consumer id Registry";
  B) 而後在"Consumer id Registry"節點下注冊一個watch用來監聽當前group中其餘consumer的"leave"和"join";只要此znode path下節點列表變動,都會觸發此group下consumer的負載均衡.(好比一個consumer失效,那麼其餘consumer接管partitions).
  C) 在"Broker id registry"節點下,註冊一個watch用來監聽broker的存活狀況;若是broker列表變動,將會觸發全部的groups下的consumer從新balance.
  總結:
  1) Producer端使用zookeeper用來"發現"broker列表,以及和Topic下每一個partition leader創建socket鏈接併發送消息.
  2) Broker端使用zookeeper用來註冊broker信息,已經監測partition leader存活性.
  3) Consumer端使用zookeeper用來註冊consumer信息,其中包括consumer消費的partition列表等,同時也用來發現broker列表,並和partition leader創建socket鏈接，並獲取消息。
• Leader的選擇：Kafka的核心是日誌文件，日誌文件在集羣中的同步是分佈式數據系統最基礎的要素。
  若是leaders永遠不會down的話咱們就不須要followers了！一旦leader down掉了，須要在followers中選擇一個新的leader.可是followers自己有可能延時過久或者crash，因此必須選擇高質量的follower做爲leader.必須保證，一旦一個消息被提交了，可是leader down掉了，新選出的leader必須能夠提供這條消息。大部分的分佈式系統採用了多數投票法則選擇新的leader,對於多數投票法則，就是根據全部副本節點的情況動態的選擇最適合的做爲leader.Kafka並非使用這種方法。
  Kafka動態維護了一個同步狀態的副本的集合（a set of in-sync replicas），簡稱ISR，在這個集合中的節點都是和leader保持高度一致的，任何一條消息必須被這個集合中的每一個節點讀取並追加到日誌中了，纔回通知外部這個消息已經被提交了。所以這個集合中的任何一個節點隨時均可以被選爲leader.ISR在ZooKeeper中維護。ISR中有f+1個節點，就能夠容許在f個節點down掉的狀況下不會丟失消息並正常提供服。ISR的成員是動態的，若是一個節點被淘汰了，當它從新達到「同步中」的狀態時，他能夠從新加入ISR.這種leader的選擇方式是很是快速的，適合kafka的應用場景。
  若是全部節點都down掉了怎麼辦？Kafka對於數據不會丟失的保證，是基於至少一個節點是存活的，一旦全部節點都down了，這個就不能保證了。
  實際應用中，當全部的副本都down掉時，必須及時做出反應。能夠有如下兩種選擇:
  1：等待ISR中的任何一個節點恢復並擔任leader。
  2：選擇全部節點中（不僅是ISR）第一個恢復的節點做爲leader.
  這是一個在可用性和連續性之間的權衡。若是等待ISR中的節點恢復，一旦ISR中的節點起不起來或者數據都是了，那集羣就永遠恢復不了了。若是等待ISR意外的節點恢復，這個節點的數據就會被做爲線上數據，有可能和真實的數據有所出入，由於有些數據它可能還沒同步到。Kafka目前選擇了第二種策略，在將來的版本中將使這個策略的選擇可配置，能夠根據場景靈活的選擇。
  這種窘境不僅Kafka會遇到，幾乎全部的分佈式數據系統都會遇到。
• 副本管理：以上僅僅以一個topic一個分區爲例子進行了討論，但實際上一個Kafka將會管理成千上萬的topic分區.Kafka儘可能的使全部分區均勻的分佈到集羣全部的節點上而不是集中在某些節點上，另外主從關係也儘可能均衡這樣每一個幾點都會擔任必定比例的分區的leader.
  優化leader的選擇過程也是很重要的，它決定了系統發生故障時的空窗期有多久。Kafka選擇一個節點做爲「controller」,當發現有節點down掉的時候它負責在游泳分區的全部節點中選擇新的leader,這使得Kafka能夠批量的高效的管理全部分區節點的主從關係。若是controller down掉了，活着的節點中的一個會備切換爲新的controller.
• Leader與副本同步：
  對於某個分區來講，保存正分區的"broker"爲該分區的"leader"，保存備份分區的"broker"爲該分區的"follower"。備份分區會徹底複製正分區的消息，包括消息的編號等附加屬性值。爲了保持正分區和備份分區的內容一致，Kafka採起的方案是在保存備份分區的"broker"上開啓一個消費者進程進行消費，從而使得正分區的內容與備份分區的內容保持一致。通常狀況下，一個分區有一個「正分區」和零到多個「備份分區」。能夠配置「正分區+備份分區」的總數量，關於這個配置，不一樣主題能夠有不一樣的配置值。注意，生產者，消費者只與保存正分區的"leader"進行通訊。
  Kafka容許topic的分區擁有若干副本，這個數量是能夠配置的，你能夠爲每一個topic配置副本的數量。Kafka會自動在每一個副本上備份數據，因此當一個節點down掉時數據依然是可用的。
  Kafka的副本功能不是必須的，你能夠配置只有一個副本，這樣其實就至關於只有一份數據。
  建立副本的單位是topic的分區，每一個分區都有一個leader和零或多個followers.全部的讀寫操做都由leader處理，通常分區的數量都比broker的數量多的多，各分區的leader均勻的分佈在brokers中。全部的followers都複製leader的日誌，日誌中的消息和順序都和leader中的一致。followers向普通的consumer那樣從leader那裏拉取消息並保存在本身的日誌文件中。
  許多分佈式的消息系統自動的處理失敗的請求，它們對一個節點是否着（alive）」有着清晰的定義。Kafka判斷一個節點是否活着有兩個條件：
• 節點必須能夠維護和ZooKeeper的鏈接，Zookeeper經過心跳機制檢查每一個節點的鏈接。
• 若是節點是個follower,他必須能及時的同步leader的寫操做，延時不能過久。符合以上條件的節點準確的說應該是「同步中的（in sync）」，而不是模糊的說是「活着的」或是「失敗的」。Leader會追蹤全部「同步中」的節點，一旦一個down掉了，或是卡住了，或是延時過久，leader就會把它移除。至於延時多久算是「過久」，是由參數replica.lag.max.messages決定的，怎樣算是卡住了，怎是由參數replica.lag.time.max.ms決定的。只有當消息被全部的副本加入到日誌中時，纔算是「committed」，只有committed的消息纔會發送給consumer，這樣就不用擔憂一旦leader down掉了消息會丟失。Producer也能夠選擇是否等待消息被提交的通知，這個是由參數acks決定的。Kafka保證只要有一個「同步中」的節點，「committed」的消息就不會丟失。