第一章 初識Kafka

Kafka定位為一個分散式流處理平臺，它以高吞吐、可持久化、可水平擴充套件、支援流資料處理等多種特性而被廣泛使用。

一個典型的Kafka體系架構包含若干Producer、若干Broker、若干Consumer，以及一個Zookeeper叢集。

Zookeeper用來負責叢集元資料的管理、控制器的選舉等操作的，Producer將訊息傳送到Broker，Broker負責將收到的訊息持久化到磁碟中，而Consumer負責從Broker訂閱並消費訊息。

在Kafka中還有兩個特別重要的概念---主題(Topic)和分割槽(Partition)。Kafka訊息以Topic為單位進行歸類，Producer負責將訊息傳送到特定的Topic，而消費者負責訂閱Topic並進行消費。

Topic是一個邏輯上的概念，它細分為多個Partition，一個Partition只屬於單個Topic，同一Topic下不同Partition包含的訊息是不同的，Partition在儲存層可以看作是一個可追加的日誌檔案，訊息被追加到Partition日誌檔案的時候會分配一個特定的偏移量(offset)。offset是訊息在Partition上的唯一標識，Kafka通過它來保證訊息在Partition內的順序性，注意的是，Kafka只能保證Partition有序而不是Topic有序。(如果保證Topic有序只能讓Topic包含一個Partition)。

每一個訊息被髮送到Broker之前，會根據Partition規則選擇儲存到哪個具體的Partition。如果Partition規則設定得合理，所有的訊息都可以均勻地分配到不同的Partition裡，如果一個Topic只對應一個檔案，那麼這個檔案所在的機器I/O將會成為這個Topic的效能瓶頸，而Partition解決了這個問題。

Kafka為Partition引入了多副本機制，通過增加副本數量可以提升容災能力。同一Partition的不同副本中儲存的是相同的訊息，副本之間是“一主多從”的關係，其中leader副本負責讀寫請求，follower副本只負責與leader副本的訊息同步，很多時候follower副本中的訊息相對leader副本有一定的滯後。副本處於不同的Broker中，當leader副本出現故障時，從follower副本中重新選舉一個新的leader副本對外提供服務。Kafka通過多副本機制實現了故障的自動轉移，當Kafka叢集中某個Broker失效時仍然能保證服務可用。

Kafka消費端也具備一定的容災能力。Consumer使用拉模式從服務端拉取訊息，並且儲存訊息的具體位置，當消費者宕機後恢復上線時可以根據之前儲存的訊息位置重新拉取需要的訊息進行消費，這樣就不會造成訊息丟失。

Partition中的所有副本統稱為AR(Assigned Replicas)。所有與leader副本保持一定程度同步的副本(包括leader副本)組成ISR(In-Sync Replicas)，ISR時AR的一個子集。與leader副本同步滯後過多的副本組成OSR(Out-Of-Sync Replicas)。正常情況下，應該AR=ISR，OSR=0。

leader副本負責維護和跟蹤ISR集合中所有follower副本的滯後狀態，當follower副本落後太多或失效時，leader副本會把它從ISR集合中剔除，如果OSR中有follower副本追上了leader副本，那麼leader副本會把它從OSR集合轉移到ISR集合。

預設情況下，當leader副本發生故障時，只有ISR集合裡的副本才有資格被選舉為新的leader。

ISR與HW和LEO也有緊密的關係。HW即High Watermark的縮寫，俗稱高水位，標識了一個特定的訊息偏移量，消費者只能拉取到這個偏移量之前的訊息。LEO是Log End Offset的縮寫，標識當前日誌檔案中下一條待寫入訊息的偏移量，LEO的大小相當於當前日誌Partition中最後一條訊息的偏移量的值加1.ISR集合中的每個副本都會維護自身的LEO，而ISR集合中最小的LEO即為Partition的HW，對消費者而言只能消費HW之前的訊息。

第二章 生產者

生產者客戶端的整體架構如下所示：

整個生產者有兩個執行緒協調執行，這兩個執行緒分別為主執行緒和Sender執行緒。主執行緒中由KafkaProducer建立訊息，然後 通過可能的攔截器、序列化器和分割槽器的作用滯後快取到訊息累加器(RecordAccumulator)。Sender執行緒負責從訊息累加器中獲取訊息並將其傳送到Kafka中。

訊息累加器主要用來快取訊息以便Sender可以批量傳送，進而減少網路傳輸的資源消耗以提升效能。如果生產者傳送訊息的速度超過傳送到伺服器的速度，則會導致生產者空間不足，這個時候KafkaProducer的send傳送呼叫要麼被阻塞，要麼丟擲異常，這個取決於引數max.block.ms的配置(預設為60s)。

在訊息累加器的內部為每個Partition都維護了一個雙端佇列，佇列中的內容是ProducerBatch。訊息寫入快取時，追加到雙端佇列的尾部。Sender從雙端佇列的頭部讀取訊息。ProducerBatch是訊息批次，包含多條ProducerRecord。

訊息在網路上都是以位元組的形式傳輸的，在傳送之前需要建立一塊記憶體區域來儲存對應的訊息。當一條訊息流入訊息累加器時，會先尋找與訊息分割槽對應的雙端佇列，再從這個雙端佇列的尾部獲取一個ProducerBatch，檢視ProducerBatch中是否還可以寫入這個ProducerRecord，可以則寫入，否則建立一個新的ProducerBatch。

Sender從訊息累加器中獲取快取的訊息之後，會將<Partition, Deque<ProducerBatch>>的形式轉變成<Node, List<ProducerBatch>的形式，Node表示Kafka叢集的Broker節點，對於網路連線來說，生產者客戶端與具體的Broker節點建立連線，而不關心訊息屬於哪個Partition，所以在這裡需要一個應用邏輯層面到網路I/O層面的轉換。

轉換成<Node, List<ProducerBatch>的形式後，Sender還會進一步封裝成<Node, Request>的形式，這樣就可以將Request傳送到各個Node了。Request在從Sender傳送到Kafka之前還會儲存到InFlightRequest中，InFlightRequest儲存物件的具體形式為Map<NodeId, Deque<Request>>，它的主要作用時快取了已經發出去但還沒收到響應的請求。