zookeeper簡介

Zookeeper是一種在分散式系統中被廣泛用來作為：分散式狀態管理、分散式協調管理、分散式配置管理、和分散式鎖服務的叢集。kafka增加和減少伺服器都會在Zookeeper節點上觸發相應的事件kafka系統會捕獲這些事件，進行新一輪的負載均衡，客戶端也會捕獲這些事件來進行新一輪的處理。

kafka簡介

生產者生產訊息、kafka叢集、消費者獲取訊息這樣一種架構，如下圖：

一些基本的概念：

Broker：Kafka叢集包含一個或多個伺服器，這種伺服器被稱為broker

Topic ：每條釋出到Kafka叢集的訊息都有一個類別，這個類別被稱為topic。（物理上不同topic的訊息分開儲存，邏輯上一個topic的訊息雖然保存於一個或多個broker上但使用者只需指定訊息的topic即可生產或消費資料而不必關心資料存於何處）

Partition ：parition是物理上的概念，每個topic包含一個或多個partition，建立topic時可指定parition數量。每個partition對應於一個資料夾，該資料夾下儲存該partition的資料和索引檔案。一個分割槽可以看作是一個FIFO的佇列。

Tips：kafka只保證同一個Partition中的訊息的順序性的。所以如果需要順序消費資料，可以根據key來消費。根據官方介紹：If a valid partition number is specified that partition will be used when sending the record. If no partition is specified but a key is present a partition will be chosen using a hash of the key. If neither key nor partition is present a partition will be assigned in a round-robin fashion.

Producer ： 負責釋出訊息到Kafka broker

Consumer ：消費訊息。每個consumer屬於一個特定的consuer group（可為每個consumer指定group name，若不指定group name則屬於預設的group）。使用consumer high level API時，同一topic的一條訊息只能被同一個consumer group內的一個consumer消費，但多個consumer group可同時消費這一訊息。

工作圖：

一個典型的kafka叢集中包含若干producer（可以是web前端產生的page view，或者是伺服器日誌，系統CPU、memory等），若干broker（Kafka支援水平擴充套件，一般broker數量越多，叢集吞吐率越高），若干consumer group，以及一個

Topic & Partition

Topic在邏輯上可以被認為是一個在的queue，每條消費都必須指定它的topic，可以簡單理解為必須指明把這條訊息放進哪個queue裡。為了使得Kafka的吞吐率可以水平擴充套件，物理上把topic分成一個或多個partition，每個partition在物理上對應一個資料夾，該資料夾下儲存這個partition的所有訊息和索引檔案。

如果partition規則設定的合理，所有訊息可以均勻分佈到不同的partition裡，這樣就實現了水平擴充套件。（如果一個topic對應一個檔案，那這個檔案所在的機器I/O將會成為這個topic的效能瓶頸，而partition解決了這個問題）。在建立topic時可以在$KAFKA_HOME/config/server.properties中指定這個partition的數量(如下所示)，當然也可以在topic建立之後去修改parition數量。

# The default number of log partitions per topic. More partitions allow greater# parallelism for consumption, but this will also result in more files across# the brokers.num.partitions=3

tips：對於傳統的message queue而言，一般會刪除已經被消費的訊息，而Kafka叢集會保留所有的訊息，無論其被消費與否。當然，因為磁碟限制，不可能永久保留所有資料（實際上也沒必要），因此Kafka提供兩種策略去刪除舊資料。一是基於時間，二是基於partition檔案大小。例如可以通過配置$KAFKA_HOME/config/server.properties，讓Kafka刪除一週前的資料，也可通過配置讓Kafka在partition檔案超過1GB時刪除舊資料。這裡要注意，因為Kafka讀取特定訊息的時間複雜度為O(1)，即與檔案大小無關，所以這裡刪除檔案與Kafka效能無關，選擇怎樣的刪除策略只與磁碟以及具體的需求有關。另外，Kafka會為每一個consumer group保留一些metadata資訊—當前消費的訊息的position，也即offset。這個offset由consumer控制。正常情況下consumer會在消費完一條訊息後線性增加這個offset。當然，consumer也可將offset設成一個較小的值，重新消費一些訊息。因為offet由consumer控制，所以Kafka broker是無狀態的，它不需要標記哪些訊息被哪些consumer過，不需要通過broker去保證同一個consumer group只有一個consumer能消費某一條訊息，因此也就不需要鎖機制，這也為Kafka的高吞吐率提供了有力保障。

Replication & Leader election

該 Replication與leader election配合提供了自動的failover機制。replication對Kafka的吞吐率是有一定影響的，但極大的增強了可用性。預設情況下，Kafka的replication數量為1。 每個partition都有一個唯一的leader，所有的讀寫操作都在leader上完成，follower批量從leader上pull資料。一般情況下partition的數量大於等於broker的數量，並且所有partition的leader均勻分佈在broker上。follower上的日誌和其leader上的完全一樣。

和大部分分散式系統一樣，Kakfa處理失敗需要明確定義一個broker是否alive。對於Kafka而言，Kafka存活包含兩個條件，一是它必須維護與Zookeeper的session(這個通過Zookeeper的heartbeat機制來實現)。二是follower必須能夠及時將leader的writing複製過來，不能“落後太多”。

leader會track“in sync”的node list。如果一個follower宕機，或者落後太多，leader將把它從”in sync” list中移除。這裡所描述的“落後太多”指follower複製的訊息落後於leader後的條數超過預定值。該值是server.properties檔案中的replica.lag.max.messages=4000

tips：Kafka只解決”fail/recover”，不處理“Byzantine”（“拜占庭”）問題。這裡應該就是CAP理論中的AP吧。一條訊息只有被“in sync” list裡的所有follower都從leader複製過去才會被認為已提交。這樣就避免了部分資料被寫進了leader，還沒來得及被任何follower複製就宕機了，而造成資料丟失（consumer無法消費這些資料）。而對於producer而言，它可以選擇是否等待訊息commit，這可以通過request.required.acks來設定。這種機制確保了只要“in sync” list有一個或以上的flollower，一條被commit的訊息就不會丟失。

這裡的複製機制即不是同步複製，也不是單純的非同步複製。事實上，同步複製要求“活著的”follower都複製完，這條訊息才會被認為commit，這種複製方式極大的影響了吞吐率。而非同步複製方式下，follower非同步的從leader複製資料，資料只要被leader寫入log就被認為已經commit，這種情況下如果follwer都落後於leader，而leader突然宕機，則會丟失資料。而Kafka的這種使用“in sync” list的方式則很好的均衡了確保資料不丟失以及吞吐率。follower可以批量的從leader複製資料，這樣極大的提高複製效能（批量寫磁碟），極大減少了follower與leader的差距（前文有說到，只要follower落後leader不太遠，則被認為在“in sync” list裡）。

上文說明了Kafka是如何做replication的，另外一個很重要的問題是當leader宕機了，怎樣在follower中選舉出新的leader。因為follower可能落後許多或者crash了，所以必須確保選擇“最新”的follower作為新的leader。一個基本的原則就是，如果leader不在了，新的leader必須擁有原來的leader commit的所有訊息。這就需要作一個折衷，如果leader在標明一條訊息被commit前等待更多的follower確認，那在它die之後就有更多的follower可以作為新的leader，但這也會造成吞吐率的下降。

一種非常常用的選舉leader的方式是“majority 靈秀”（“少數服從多數”），這種模式下，如果我們有2f+1個replica（包含leader和follower），那在commit之前必須保證有f+1個replica複製完訊息，為了保證正確選出新的leader，fail的replica不能超過f個。因為在剩下的任意f+1個replica裡，至少有一個replica包含有最新的所有訊息。這種方式有個很大的優勢，系統的latency只取決於最快的幾臺server，也就是說，如果replication factor是3，那latency就取決於最快的那個follower而非最慢那個。majority vote也有一些劣勢，為了保證leader election的正常進行，它所能容忍的fail的follower個數比較少。如果要容忍1個follower掛掉，必須要有3個以上的replica，如果要容忍2個follower掛掉，必須要有5個以上的replica。也就是說，在生產環境下為了保證較高的容錯程度，必須要有大量的replica，而大量的replica又會在大資料量下導致效能的急劇下降。這就是這種演算法更多用在Zookeeper這種共享叢集配置的系統中而很少在需要儲存大量資料的系統中使用的原因。

而kafka的方式是：Kafka在Zookeeper中動態維護了一個ISR（in-sync replicas） set，這個set裡的所有replica都跟上了leader，只有ISR裡的成員才有被選為leader的可能。在這種模式下，對於f+1個replica，一個Kafka topic能在保證不丟失已經commit的訊息的前提下容忍f個replica的失敗。