[TOC] https://mp.weixin.qq.com/s/pTl8RofxRNA6-FjEACdD3g ## kafka 是什么？有什么作用？ Kafka 是一個分布式的流式處理平臺，它以高吞吐、可持久化、可水平擴展、支持流數據處理等多種特性而被廣泛使用 主要功能體現于三點： * **消息系統**：kafka與傳統的消息中間件都具備**系統解耦、冗余存儲、流量削峰、緩沖、異步通信、擴展性、可恢復性**等功能。與此同時，kafka還提供了大多數消息系統難以實現的消息順序性保障及回溯性消費的功能。 * **存儲系統**：kafka把**消息持久化到磁盤**，相比于其他基于內存存儲的系統而言，有效的降低了消息丟失的風險。這得益于其消息持久化和多副本機制。也可以將kafka作為長期的存儲系統來使用，只需要把對應的數據保留策略設置為“永久”或啟用主題日志壓縮功能。 * **流式處理平臺**：kafka為流行的流式處理框架提供了可靠的數據來源，還提供了一個完整的流式處理框架，比如窗口、連接、變換和聚合等各類操作。 ## kafka 的架構是怎么樣的？  Producer 將消息發送到 Broker，Broker 負責將受到的消息存儲到磁盤中，而 Consumer 負責從 Broker 訂閱并消費消息。 Kafka 基本概念： * **Producer**：生產者，負責將消息發送到 Broker * **Consumer**：消費者，從 Broker 接收消息 * **Consumer Group**：消費者組，由**多個 Consumer 組成**。消費者組內每個消費者負責消費不同分區的數據，**一個分區只能由一個組內消費者消費**；消費者組之間互不影響。所有的消費者都屬于某個消費者組，即消費者組是邏輯上的一個訂閱者。 * **Broker**：可以看做一個獨立的**Kafka 服務節點或 Kafka 服務實例**。如果一臺服務器上只部署了一個 Kafka 實例，那么我們也可以將 Broker 看做一臺 Kafka 服務器。 * **Topic**：一個邏輯上的概念，包含很多 Partition，**同一個 Topic 下的 Partiton 的消息內容是不相同的**。 * **Partition**：為了實現擴展性，一個非常大的 topic**可以分布到多個 broker 上，一個 topic 可以分為多個 partition**，每個 partition 是一個有序的隊列。 * **Replica**：副本，**同一分區的不同副本保存的是相同的消息**，為保證集群中的某個節點發生故障時，該節點上的 partition 數據不丟失，且 kafka 仍然能夠繼續工作，kafka 提供了副本機制，一個 topic 的每個分區都有若干個副本，一個 leader 和若干個 follower。 * **Leader**：每個分區的多個副本中的"主副本"，**生產者以及消費者只與 Leader 交互**。 * **Follower**：每個分區的多個副本中的"從副本"，**負責實時從 Leader 中同步數據，保持和 Leader 數據的同步**。Leader 發生故障時，從 Follower 副本中重新選舉新的 Leader 副本對外提供服務。 ## Kafka Replicas是怎么管理的？  * AR:分區中的**所有 Replica 統稱為 AR** * ISR:所有與 Leader 副本**保持一定程度同步**的Replica(包括 Leader 副本在內)組成 ISR * OSR:與 Leader 副本**同步滯后過多的**Replica 組成了 OSR Leader 負責維護和跟蹤 ISR 集合中所有 Follower 副本的滯后狀態，當 Follower 副本落后過多時，就會將其放入 OSR 集合，當 Follower 副本追上了 Leader 的進度時，就會將其放入 ISR 集合。 默認情況下，**只有 ISR 中的副本才有資格晉升為 Leader**。 ## 如何確定當前能讀到哪一條消息？ 分區相當于一個日志文件，我們先簡單介紹幾個概念  如上圖是一個分區日志文件 * 標識**共有7條消息**，offset (消息偏移量)分別是0~6 * 0 代表這個日志文件的**開始** * HW(High Watermark) 為4，0~3 代表這個日志文件**可以消費的區間**，消費者只能消費到這四條消息 * LEO 代表即將要寫入消息的偏移量 offset **分區 ISR 集合中的每個副本都會維護自己的 LEO，而 ISR 集合中最小的LEO 即為分區的 HW**  如上圖: 三個分區副本都是 ISR集合當中的，最小的 LEO 為 3，就代表分區的 HW 為3，所以當前分區只能消費到 0~2 之間的三條數據，如下圖  ## 生產者發送消息有哪些模式？ 總共有三種模式 * 1.**發后即忘**（fire-and-forget） * 它只管往 Kafka 里面發送消息，但是**不關心消息是否正確到達**，這種方式的效率最高，但是可靠性也最差，比如當發生某些不可充實異常的時候會造成消息的丟失 * 2.**同步**（sync） * producer.send()返回一個Future對象，調用get()方法變回進行同步等待，就知道消息是否發送成功，**發送一條消息需要等上個消息發送成功后才可以繼續發送** * 3.**異步**（async） * Kafka支持 producer.send() 傳入一個回調函數，消息不管成功或者失敗都會調用這個回調函數，這樣就算是異步發送，我們也知道消息的發送情況，然后再回調函數中選擇記錄日志還是重試都取決于調用方 * ## 發送消息的分區策略有哪些？  * 1.輪詢：**依次**將消息發送該topic下的所有分區，如果在創建消息的時候 key 為 null，Kafka 默認采用這種策略。 * 2.key 指定分區：在創建消息是 key 不為空，并且使用默認分區器，Kafka 會將 key 進行 hash，然后**根據hash值映射到指定的分區上**。這樣的好處是 key 相同的消息會在一個分區下，Kafka 并不能保證全局有序，但是在每個分區下的消息是有序的，按照順序存儲，按照順序消費。在保證同一個 key 的消息是有序的，這樣基本能滿足消息的順序性的需求。但是**如果 partation 數量發生變化，那就很難保證 key 與分區之間的映射關系了**。 * 3.自定義策略：實現 Partitioner 接口就能自定義分區策略。 * 4.指定 Partiton 發送 ## Kafka 支持讀寫分離嗎？為什么？ Kafka 是**不支持讀寫分離**的，那么讀寫分離的好處是什么？主要就是讓一個節點去承擔另一個節點的負載壓力，也就是能做到一定程度的負載均衡，而且 Kafka 不通過讀寫分離也可以一定程度上去實現負載均衡。  * 1.數據不一致的問題:讀寫分離必然涉及到數據的同步，只要是**不同節點之間的數據同步**，必然**會有數據不一致的問題**存在。 * 2.延時問題:由于 Kafka 獨特的數據處理方式，導致如果將數據從一個節點同步到另一個節點必然會經過**主節點磁盤和從節點磁盤**，對一些延時性要求較高的應用來說，并不太適用 ## Kafka 是怎么去實現負載均衡的？ Kafka 的負責均衡主要是**通過分區來實現**的，我們知道 Kafka 是**主寫主讀**的架構，如下圖:  共三個 broker ，里面各有三個副本，總共有三個 partation， 深色的是 leader，淺色的是 follower，上下灰色分別代表生產者和消費者，虛線代表 follower 從 leader 拉取消息。 我們從這張圖就可以很明顯的看出來，**每個 broker 都有消費者拉取消息，每個 broker 也都有生產者發送消息，每個 broker 上的讀寫負載都是一樣的**，這也說明了 kafka 獨特的架構方式可以通過主寫主讀來實現負載均衡。 ## Kafka 的負責均衡會有什么問題呢？ kafka的負載均衡在絕對理想的狀況下可以實現，但是會有某些情況出現一定程度上的負載不均衡  * 1.**broker 端分配不均**:當創建 topic 的時候可能會出現某些 broker 分配到的分區數多，而有些 broker 分配的分區少，這就導致了 leader 多副本不均。 * 2.**生產者寫入消息不均**:生產者可能只對某些 broker 中的 leader 副本進行大量的寫入操作，而對其他的 leader 副本不聞不問。 * 3.**消費者消費不均**:消費者可能只對某些 broker 中的 leader 副本進行大量的拉取操作，而對其他的 leader 副本不聞不問。 * 4.**leader 副本切換不均**：當主從副本切換或者分區副本進行了重分配后，可能會導致各個 broker 中的 leader 副本分配不均勻。 ## Kafka 的可靠性是怎么保證的？  **1.acks** 這個參數用來指定分區中有多少個副本收到這條消息，生產者才認為這條消息是寫入成功的，這個參數有三個值： * 1.acks = 1，默認為1。生產者發送消息，**只要 leader 副本成功寫入消息，就代表成功**。這種方案的問題在于，當返回成功后，如果 leader 副本和 follower 副本還**沒有來得及同步**，leader 就崩潰了，那么在選舉后新的 leader 就沒有這條**消息，也就丟失了**。 * 2.acks = 0。生產者發送消息后直接算寫入成功，不需要等待響應。這個方案的問題很明顯，**只要服務端寫消息時出現任何問題，都會導致消息丟失**。 * 3.acks = -1 或 acks = all。生產者發送消息后，需要等待 ISR 中的所有副本都成功寫入消息后才能收到服務端的響應。毫無疑問這種方案的**可靠性是最高**的，但是如果 ISR 中只有leader 副本，那么就和 acks = 1 毫無差別了。 **2.消息發送的方式** 第6問中我們提到了生產者發送消息有三種方式，發完即忘，同步和異步。我們可以**通過同步或者異步**獲取響應結果，**失敗做重試**來保證消息的可靠性。 **3.手動提交位移** 默認情況下，當消費者消費到消息后，就會自動提交位移。但是如果消費者消費出錯，沒有進入真正的業務處理，那么就可能會導致這條消息消費失敗，從而丟失。我們可以開啟手動提交位移，等待業務正常處理完成后，再提交offset。 **4.通過副本 LEO 來確定分區 HW** ## Kafka 的消息消費方式有哪些？ 一般消息消費有兩種模式，推和拉。Kafka的消費是屬于**拉模式**的，而此模式的消息消費方式有**兩種，點對點和發布訂閱**。  * 1.**點對點**:如果所有消費者屬于同一個消費組，那么所有的消息都會被均勻的投遞給每一個消費者，**每條消息只會被其中一個消費者消費**。  * 2.**發布訂閱**:如果所有消費者屬于不同的消費組，那么所有的消息都會被投遞給每一個消費者，**每個消費者都會收到該消息**。 ## 副本 leader 是怎么選舉的？ 優先副本就是說在 AR 集合中的第一個副本。比如分區 2 的 AR 為 0，1，2，那么分區 2 的優先副本就為0。理想情況下優先副本就是 leader 副本。優先副本選舉就是促使優先副本成為 leader 副本，從而維護集群的負載均衡。 ## 如何增強消費者的消費能力？ * 1.可以考慮增加 topic 的分區數，并且同時提升消費組的消費者數量，消費者數=分區數。 * 2.如果是消費者消費不及時，可以采用多線程的方式進行消費，并且優化業務方法流程，同樣的分區數 ## kafka 控制器是什么？有什么作用 在 Kafka 集群中會有一個或多個 broker，其中有一個 broker 會被選舉為控制器，**它負責管理整個集群中所有分區和副本的狀態**，kafka 集群中**只能有一個控制器**。 * 當某個分區的 leader 副本出現故障時，由控制器負責**為該分區選舉新的 leader 副本**。 * 當檢測到某個分區的ISR集合發生變化時，由控制器負責**通知所有 broker 更新其元數據信息**。 * 當為某個 topic 增加分區數量時，由控制器**負責分區的重新分配**。 ## kafka 控制器是怎么進行選舉的？ 所謂控制器就是一個Borker，在一個kafka集群中，有多個broker節點，但是它們之間需要選舉出一個leader，其他的broker充當follower角色。集群中第一個啟動的broker會通過在zookeeper中創建臨時節點`/controller`來讓自己成為控制器，其他broker啟動時也會在zookeeper中創建臨時節點，但是發現節點已經存在，所以它們會收到一個異常，意識到控制器已經存在，那么就會在zookeeper中創建watch對象，便于它們收到控制器變更的通知。 　　那么如果控制器由于網絡原因與zookeeper斷開連接或者異常退出，那么其他broker通過watch收到控制器變更的通知，就會去嘗試創建臨時節點`/controller`，如果有一個broker創建成功，那么其他broker就會收到創建異常通知，也就意味著集群中已經有了控制器，其他broker只需創建watch對象即可。 　　如果集群中有一個broker發生異常退出了，那么控制器就會檢查這個broker是否有分區的副本leader，如果有那么這個分區就需要一個新的leader，此時控制器就會去遍歷其他副本，決定哪一個成為新的leader，同時更新分區的ISR集合。 　　如果有一個broker加入集群中，那么控制器就會通過Broker ID去判斷新加入的broker中是否含有現有分區的副本，如果有，就會從分區副本中去同步數據。 　　集群中每選舉一次控制器，就會通過zookeeper創建一個`controller epoch`，每一個選舉都會創建一個更大，包含最新信息的`epoch`，如果有broker收到比這個`epoch`舊的數據，就會忽略它們，kafka也通過這個`epoch`來防止集群產生“腦裂”。 ## kafka 為什么這么快？  * **1.順序讀寫** 磁盤分為順序讀寫與隨機讀寫，基于磁盤的隨機讀寫確實很慢，但磁盤的順序讀寫性能卻很高，kafka 這里采用的就是順序讀寫。 * **2.Page Cache** 為了優化讀寫性能，Kafka 利用了操作**系統本身的 Page Cache**，就是利用操作系統自身的內存而不是JVM空間內存。 * **3.零拷貝** Kafka使用了零拷貝技術，也就是**直接將數據從內核空間的讀緩沖區直接拷貝到內核空間的 socket 緩沖區**，然后再寫入到 NIC 緩沖區，避免了在內核空間和用戶空間之間穿梭。 * **4.分區分段+索引** Kafka 的 message 是按 topic分 類存儲的，topic 中的數據又是按照一個一個的 partition 即分區存儲到不同 broker 節點。每個 partition 對應了操作系統上的一個文件夾，partition 實際上又是按照segment分段存儲的。 通過這種分區分段的設計，Kafka 的 message 消息實際上是分布式存儲在一個一個小的 segment 中的，每次文件操作也是直接操作的 segment。為了進一步的查詢優化，Kafka 又默認為分段后的數據文件建立了索引文件，就是文件系統上的.index文件。這種分區分段+索引的設計，不僅提升了數據讀取的效率，同時也提高了數據操作的并行度。 * **5.批量讀寫** Kafka**數據讀寫也是批量的而不是單條的**,這樣可以避免在網絡上頻繁傳輸單個消息帶來的延遲和帶寬開銷。假設網絡帶寬為10MB/S，一次性傳輸10MB的消息比傳輸1KB的消息10000萬次顯然要快得多。 * **6.批量壓縮** Kafka 把所有的消息都變成一個**批量的文件**，并且進行合理的**批量壓縮**，減少網絡 IO 損耗，通過 mmap 提高 I/O 速度，寫入數據的時候由于單個Partion是末尾添加所以速度最優；讀取數據的時候配合 sendfile 進行直接讀取。 ## 什么情況下 kafka 會丟失消息？ Kafka 有三次消息傳遞的過程：生產者發消息給 Broker，Broker 同步消息和持久化消息，Broker 將消息傳遞給消費者。  * 1.生產者發送數據: 在第 11 問中的 acks中有說到 * 當 acks 為 0，**只要服務端寫消息時出現任何問題，都會導致消息丟失**。 * 當 acks 配置為 1 時，生產者發送消息，只要 leader 副本成功寫入消息，就代表成功。這種方案的問題在于，當返回成功后，**如果 leader 副本和 follower 副本還沒有來得及同步，leader 就崩潰了，那么在選舉后新的 leader 就沒有這條消息，也就丟失了**。 * 2.Broker 存儲數據:kafka 通過 Page Cache 將數據寫入磁盤。 * Page Cache 就是當往磁盤文件寫入的時候，系統會先將數據流寫入緩存中，但是**什么時候將緩存的數據寫入文件中是由操作系統自行決定**。所以**如果此時機器突然掛了，也是會丟失消息的**。 * 3.消費者消費數據:在開啟**自動提交 offset**時，只要消費者消費到消息，那么就會自動提交偏移量，**如果業務還沒有來得及處理，那么消息就會丟失**。