消息中間件-Kafka

一、kafka簡介

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

• Kafka具有近乎實時性的消息處理能力，即使面對海量消息也能夠高效地存儲消息和查詢消息。
  Kafka將消息保存在磁盤中，它以順序讀寫的方式訪問磁盤，從而避免了隨機讀寫磁盤導致的性能瓶頸。
• Kafka支持批量讀寫消息，并且會對消息進行批量壓縮
• Kafka支持消息分區，每個分區中的消息保證順序傳輸，而分區之間則可以并發操作，具備高并發能力
• Kafka也支持在線增加分區，支持在線水平擴展
• Kafka支持為每個分區創建多個副本，其中只會有一個Leader副本負責讀寫，其他副本只負責與Leader副本進行同步，Kafka會將Leader副本均勻地分布在集群中的服務器上，實現性能最大化，同時具備較強的容災能力

3、應用場景

• 在應用系統中可以將Kafka作為傳統的消息中間件，實現消息隊列和消息的發布/訂閱，在某些特定場景下其性能要更優于RabbitMQ、ActiveMQ等傳統的消息中間件
• Kafka也被用作系統中的數據總線，將其接入多個子系統中，子系統會將產生的數據發送到Kafka中保存，之后流轉到目的系統中
• Kafka還可以用作日志收集中心，多個系統產生的日志統一收集到Kafka中，然后由數據分析平臺進行統一處理。日志會被Kafka持久化到磁盤，所以同時支持離線數據處理和實時數據處理
• 事件溯源
  Kafka的持久化存儲和順序消息傳遞特性使其成為事件溯源的理想選擇。通過將系統的事件以消息的形式寫入Kafka的主題中，可以實現對系統狀態的完全恢復和追溯。這對于需要滿足合規性要求或實現事件溯源的系統非常重要，如金融交易系統、電子商務系統等。
• 流媒體處理
  Kafka在流媒體處理領域也有著廣泛的應用。流媒體處理要求系統能夠高效地處理大規模的音視頻數據流。Kafka的高吞吐量和低延遲特性使其成為一個理想的流媒體處理平臺。通過使用Kafka，可以構建高性能的音視頻處理系統，實現實時的流媒體傳輸、轉碼、存儲和分發。

4、數據持久化
在分布式系統中，各個組件是通過網路連接起來的。一般認為網絡傳輸是不可靠的，當數據在兩個組件之間進行傳遞的時候，傳輸過程可能會失敗。除非數據被持久化到磁盤，否則就可能造成消息的丟失。Kafka把數據以消息的形式持久化到磁盤，即使Kafka出現宕機，也可以保證數據不會丟失，通過這一方式規避了數據丟失風險。為了避免磁盤上的數據不斷增長，Kafka提供了日志清理、日志壓縮等功能，對過時的、已經處理完成的數據進行清除。在磁盤操作中，耗時最長的就是尋道時間，這是導致磁盤的隨機I/O性能很差的主要原因。為了提高消息持久化的性能，Kafka采用順序讀寫的方式訪問，實現了高吞吐量。
5、擴展與容災
Kafka的每個Topic（主題）都可以分為多個Partition（分區），每個分區都有多個Replica（副本），實現消息冗余備份。每個分區中的消息是不同的，這類似于數據庫中水平切分的思想，提高了并發讀寫的能力。而同一分區的不同副本中保存的是相同的消息，副本之間是一主多從的關系，其中Leader副本負責處理讀寫請求，Follower副本則只與Leader副本進行消息同步，當Leader副本出現故障時，則從Follower副本中重新選舉Leader副本對外提供服務。這樣，通過提高分區的數量，就可以實現水平擴展；通過提高副本的數量，就可以提高容災能力。
Kafka的容災能力不僅體現在服務端，在Consumer端也有相關設計。Consumer使用pull方式從服務端拉
取消息，并且在Consumer端保存消費的具體位置，當消費者宕機后恢復上線，可以根據自己保存的消費位
置重新拉取需要的消息進行消費，這就不會造成消息丟失。也就是說，Kafka不決定何時、如何消費消息，
而是Consumer自己決定何時、如何消費消息。
Kafka還支持Consumer的水平擴展能力。我們可以讓多個Consumer加入一個Consumer Group（消費組），在一個Consumer Group中，每個分區只能分配給一個Consumer消費，當Kafka服務端通過增加分區數量進行水平擴展后，我們可以向Consumer Group中增加新的Consumer來提高整個Consumer Group的消費能力。當Consumer Group中的一個Consumer出現故障下線時，會通過Rebalance操作將下線Consumer負責處理的分區分配給其他Consumer繼續處理；當下線Consumer重新上線加入Consumer Group時，會再進行一次Rebalance操作，重新分配分區。當然，一個Consumer Group可以訂閱很多不同的Topic，每個Consumer可以同時處理多個分區
6、分區數據的順序保證
Kafka保證一個分區內的消息的有序性，但是并不保證多個partition之間的數據有順序。
7、異步通信
Kafka為系統提供了異步處理能力。例如，兩個系統需要通過網絡進行數據交換，其中一端可以把一個消息放入Kafka中后立即返回繼續執行其他路基，不需要等待對端的響應。待后者將處理結果放入Kafka中之后，前者可以從其中獲取并解析響應。

二、Kfaka核心概念

• 消息
  消息是Kafka中最基本的數據單元。消息由一串字節構成，其中主要由key和value構成，key和value也都是byte數組。key的主要作用是根據一定的策略，將此消息路由到指定的分區中，這樣就可以保證包含同一key的消息全部寫入同一分區中，key可以是null。
• Topic/分區/Log
  Topic是用于存儲消息的邏輯概念，可以看作一個消息集合。每個Topic可以有多個生產者向其中推送（push）消息，也可以有任意多個消費者消費其中的消息。每個Topic可以劃分成多個分區（每個Topic都至少有一個分區），同一Topic下的不同分區包含的消息是不同的。每個消息在被添加到分區時，都會被分配一個offset，它是消息在此分區中的唯一編號，Kafka通過offset保證消息在分區內的順序，offset的順序性不跨分區，即Kafka只保證在同一個分區內的消息是有序的；同一Topic的多個分區內的消息，Kafka并不保證其順序性。同一Topic的不同分區會分配在不同的Broker上。分區是Kafka水平擴展性的基礎，我們可以通過增加服務器并在其上分配Partition的方式來增加Kafka的并行處理能力。
  分區在邏輯上對應著一個Log，當生產者將消息寫入分區時，實際上是寫入到了分區對應的Log中。Log是一個邏輯概念，可以對應到磁盤上的一個文件夾。Log由多個Segment組成，每個Segment對應一個日志文件和索引文件。在面對海量數據時，為避免出現超大文件，每個日志文件的大小是有限制的，當超出限制后則會創建新的Segment，繼續對外提供服務。這里要注意，因為Kafka采用順序I/O，所以只向最新的Segment追加數據。
• 保留策略與日志壓縮
  無論消費者是否已經消費了消息，Kafka都會一直保存這些消息，但并不會像數據庫那樣長期保存。為了避免磁盤被占滿，Kafka會配置相應的“保留策略”（retention policy），以實現周期性地刪除陳舊的消息。

一種是根據消息保留的時間，當消息在Kafka中保存的時間超過了指定時間，就可以被刪除
根據Topic存儲的數據大小，當Topic所占的日志文件大小大于一個閾值，則可以開始刪除最舊的消息。

Kafka會啟動一個后臺線程，定期檢查是否存在可以刪除的消息。“保留策略”的配置是非常靈活的，可以有全局的配置，也可以針對Topic進行配置覆蓋全局配置。

• 消息壓縮
  在很多場景中，消息的key與value的值之間的對應關系是不斷變化的，就像數據庫中的數據會不斷被修改一樣，消費者只關心key對應的最新value值。此時，可以開啟Kafka的日志壓縮功能，Kafka會在后臺啟動一個線程，定期將相同key的消息進行合并，只保留最新的value值。
  壓縮過程：
  
• Broker
  一個單獨的Kafka服務器就是一個Broker。Broker的主要工作就是接收生產者發過來的消息，分配offset，之后保存到磁盤中；同時，接收消費者、其他Broker的請求，根據請求類型進行相應處理并返回響應。在一般的生產環境中，一個Broker獨占一臺物理服務器。
• 分區副本
  每個分區的副本集合中，都會選舉出一個副本作為Leader副本，Kafka在不同的場景下會采用不同的選舉策略所有的讀寫請求都由選舉出的Leader副本處理，其他都作為Follower副本，Follower副本僅僅是從Leader 副本處把數據拉取到本地之后，同步更新到自己的Log中。一般情況下，同一分區的多個副本會被分配到不同的Broker上，這樣，當Leader所在的Broker宕機之后，可以重新選舉新的Leader，繼續對外提供服務。
• ISR（In-Sync Replica）集合
  ISR（In-Sync Replica）集合表示的是目前“可用”（alive）且消息量與Leader相差不多的副本集合,其中每個副本必須滿足副本所在節點必須維持著與ZooKeeper的連接與副本最后一條消息的offset與Leader副本的最后一條消息的offset之間的差值不能超出指定的閾值*
  每個分區中的Leader副本都會維護此分區的ISR集合。寫請求首先由Leader副本處理，之后Follower副本會從Leader上拉取寫入的消息，這個過程會有一定的延遲，導致Follower副本中保存的消息略少于Leader副本，只要未超出閾值都是可以容忍的。如果一個Follower副本出現異常，比如：宕機，發生長時間GC而導致Kafka僵死或是網絡斷開連接導致長時間沒有拉取消息進行同步，就會違反上面的兩個條件，從而被Leader副本踢出ISR集合。當Follower副本從異常中恢復之后，會繼續與Leader副本進行同步，當Follower副本“追上”（即最后一條消息的offset的差值小于指定閾值）Leader副本的時候，此Follower副本會被Leader副本重新加入到ISR中。
• HW
  HW（HighWatermark）和LEO與上面的ISR集合緊密相關。HW標記了一個特殊的offset，當消費者處理消息的時候，只能拉取到HW之前的消息，HW之后的消息對消費者來說是不可見的。與ISR集合類似，HW也是由Leader副本管理的。當ISR集合中全部的Follower副本都拉取HW指定消息進行同步后，Leader副本會遞增HW的值。
• LEO（Log End Offset）
  LEO是所有的副本都會有的一個offset標記，它指向追加到當前副本的最后一個消息的offset。當生產者向Leader副本追加消息的時候，Leader副本的LEO標記會遞增；當Follower副本成功從Leader副本拉取消息并更新到本地的時候，Follower副本的LEO就會增加。

HW與LEO的關系如下圖

①Producer向此Partition推送消息。
②Leader副本將消息追加到Log中，并遞增其LEO。
③Follower副本從Leader副本拉取消息進行同步。
④Follower副本將拉取到的消息更新到本地Log中，并遞增其LEO。
⑤當ISR集合中所有副本都完成了對offset=11的消息的同步，Leader副本會遞增HW。
在①~⑤步完成之后，offset=11的消息就對生產者可見了。

• 為什么kafka數據冗余要設計成這樣？
  常見的數據同步包括同步復制和異步復制。
  同步復制要求所有能工作的Follower副本都復制完，這條消息才會被認為提交成功。一旦有一個
  Follower副本出現故障，就會導致HW無法完成遞增，消息就無法提交，生產者獲取不到消息。這
  種情況下，故障的Follower副本會拖慢整個系統的性能，甚至導致整個系統不可用。
  異步復制中，Leader副本收到生產者推送的消息后，就認為此消息提交成功。Follower副本則異步
  地從Leader副本同步消息。這種設計雖然避免了同步復制的問題，但同樣也存在一定的風險。現在
  假設所有Follower副本的同步速度都比較慢，它們保存的消息量都遠遠落后于Leader副本。
  
  此時Leader副本所在的Broker突然宕機，則會重新選舉新的Leader副本，而新Leader副本中沒有原來
  Leader副本的消息，這就出現了消息的丟失，而有些消費者則可能消費了這些丟失的消息，狀態變得不可控。
  Kafka權衡了同步復制和異步復制兩種策略，通過引入了ISR集合，巧妙地解決了上面兩種方案存在的
  缺陷：當Follower副本的延遲過高時，Leader副本被踢出ISR集合，消息依然可以快速提交，生產者可以快速得到響應，避免高延時的Follower副本影響整個Kafka集群的性能。當Leader副本所在的Broker突然宕機的時候，會優先將ISR集合中Follower副本選舉為Leader副本，新Leader副本中包含了HW之前的全部消息，這就避免了消息的丟失。值得注意是，Follower副本可以批量地從Leader副本復制消息，這就加快了網絡I/O，Follower 副本在更新消息時是批量寫磁盤，加速了磁盤的I/O，極大減少了Follower與Leader的差距。
• Cluster(集群)與Controller(指揮中心)
  多個Broker可以做成一個Cluster（集群）對外提供服務，每個Cluster當中會選舉出一個Broker來擔任
  Controller，Controller是Kafka集群的指揮中心，而其他Broker則聽從Controller指揮實現相應的功能。
  Controller負責管理分區的狀狀態、管理每個分區的副本狀態、監聽Zookeeper中數據的變化等工作。
  Controller也是一主多從的實現，所有Broker都會監聽Controller Leader的狀態，當Leader Controller出現故障時則重新選舉新的Controller Leader。
• 生產者
  生產者（Producer）的主要工作是生產消息，并將消息按照一定的規則推送到Topic的分區中。這里選
  擇分區的“規則”可以有很多種，例如：根據消息的key的Hash值選擇分區，或按序輪詢全部分區的方式。
• 消費者
  消費者（Consumer）的主要工作是從Topic中拉取消息，并對消息進行消費。某個消費者消費到
  Partition的哪個位置（offset）的相關信息，是Consumer自己維護的。 如下圖
  
• Consumer Group 消費者組
  在Kafka中，多個Consumer可以組成一個Consumer Group，一個Consumer只能屬于一個Consumer
  Group。Consumer Group保證其訂閱的Topic的每個分區只被分配給此Consumer Group中的一個消費者處理。如果不同Consumer Group訂閱了同一Topic，Consumer Group彼此之間不會干擾。這樣，如果要實現一個消息可以被多個消費者同時消費（“廣播”）的效果，則將每個消費者放入單獨的一個Consumer Group；如果要實現一個消息只被一個消費者消費（“獨占”）的效果，則將所有Consumer放入一個Consumer Group中。
  消費者組消費消息如圖：
  
  Consumer Group除了實現“獨占”和“廣播”模式的消息處理，Kafka還通過Consumer Group實現了消費者的水平擴展和故障轉移。在上圖中，當Consumer3的處理能力不足以處理兩個Partition中的數據時，可以通過向Consumer Group中添加消費者的方式，觸發Rebalance操作重新分配分區與消費者的對應關系，從而實現水平擴展。如下圖所示，添加Consumer4之后，Consumer3只消費Partition2中的消息，Partition3中的消息則由Consumer4來消費。
  
  下面來看消費者出現故障的場景，當Consumer4宕機時，Consumer Group會自動重新分配分區，如下圖所示，由Consumer3接管Consumer4對應的分區繼續處理。
  
  注，Consumer Group中消費者的數量并不是越多越好，當其中消費者數量超過分區的數量時，會導
  致有消費者分配不到分區，從而造成消費者的浪費

三、Kafak消息處理流程圖

• 單個Kafka Server單體模式
  
  Kafka的每個Topic（主題）都可以分為多個Partition（分區），每個分區都有多個Replica（副本），實現消息冗余備份。每個分區中的消息是不同的，這類似于數據庫中水平切分的思想，提高了并發讀寫的能力。而同一分區的不同副本中保存的是相同的消息，副本之間是一主多從的關系，其中Leader副本負責處理讀寫請求，Follower副本則只與Leader副本進行消息同步，當Leader副本出現故障時，則從Follower副本中重新選舉Leader副本對外提供服務。
• 集群Kafak Server 集群模式
  
  如上所示，生產者會根據業務邏輯產生消息，之后根據路由規則將消息發送到指定分區的Leader副本所在的Broker上。在Kafka服務端接收到消息后，會將消息追加到Log中保存，之后Follower副本會與
  Leader副本進行同步，當ISR集合中所有副本都完成了此消息的同步后，則Leader副本的HW會增加，并向生產者返回響應。
  當消費者加入到Consumer Group時，會觸發Rebalance操作將分區分配給不同的消費者消費。隨后，消費者會恢復其消費位置，并向Kafka服務端發送拉取消息的請求，Leader副本會驗證請求的offset以及其他相關信息，最后返回消息。