簡介

Kafka是基于發(fā)布訂閱的消息系統(tǒng)。最初起源于LinkedIn，于2011年成為開源Apache項目，然后于2012年成為Apache頂級項目。Kafka用Scala和Java編寫，因其分布式可擴展架構及可持久化、高吞吐率特征而被廣泛使用。

[[282225]]

消息隊列

通常在項目中，我們會因為如下需求而引入消息隊列模塊：

1.解耦：消息系統(tǒng)相當于在處理過程中間插入了一個隱含的、基于數(shù)據(jù)的接口層。無需預先定義不同的接口地址和請求應答規(guī)范，這允許數(shù)據(jù)上下游獨立決定雙方的處理過程，只需要約定數(shù)據(jù)格式即可任意擴展服務類型和業(yè)務需求。

2.緩沖：消息系統(tǒng)作為一個緩沖池，應對常見的訪問量不均衡情形。比如特殊節(jié)假日的流量劇增和每日不同時段的訪問量差異。以及處理不同數(shù)據(jù)類型所需的不同實時性。使整個業(yè)務處理架構以較低成本獲得一定靈活性。

3. 異步：很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。

Kafka的特點

作為一種分布式的，基于發(fā)布/訂閱的消息系統(tǒng)。Kafka的主要設計目標如下：

1.以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間復雜度的訪問性能。

2.高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條以上消息的傳輸。

3.支持Kafka Server間的消息分區(qū)，及分布式消費，同時保證每個Partition內(nèi)的消息順序傳輸。

4.同時支持離線數(shù)據(jù)處理和實時數(shù)據(jù)處理。

5.支持在線水平擴展。

Kafka體系架構

如上圖所示，一個典型的Kafka體系架構包括若干Producer(可以是服務器日志，業(yè) 務數(shù)據(jù)，頁面前端產(chǎn)生的page view等等)，若干Broker(Kafka支持水平擴展，一般Broker數(shù)量越多，集群吞吐率越高)，若干Consumer (Group)，以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置，選舉leader，以及在Consumer Group發(fā)生變化時進行rebalance。Producer使用push模式將消息發(fā)布到Broker，Consumer使用pull模式從Broker訂閱并消費消息。

名詞解釋：

Topic & Partition

一個topic可以認為一個一類消息，每個topic將被分成多個partition，每個partition在存儲層面是append log文件。任何發(fā)布到此partition的消息都會被追加到log文件的尾部，每條消息在文件中的位置稱為offset(偏移量)，offset為一個long型的數(shù)字，它唯一標記一條消息。每條消息都被append到partition中，順序?qū)懘疟P因此效率非常高。這是Kafka高吞吐率的重要基礎。

Producer發(fā)送消息到Broker時，會根據(jù)Paritition機制選擇將其存儲到哪一個Partition。如果Partition機制設置合理，所有消息可以均勻分布到不同的Partition里，這樣就實現(xiàn)了負載均衡。如果一個Topic對應一個文件，那這個文件所在的機器I/O將會成為這個Topic的性能瓶頸，而有了Partition后，不同的消息可以并行寫入不同Broker的不同Partition里，極大的提高了吞吐率。可以通過配置項num.partitions來指定新建Topic的默認Partition數(shù)量，也可在創(chuàng)建Topic時通過參數(shù)指定，同時也可以在Topic創(chuàng)建之后通過Kafka提供的工具修改。

Kafka的復制機制

Kafka 中的每個主題分區(qū)都被復制了 n 次，其中的 n 是主題的復制因子(replication factor)。這允許 Kafka 在集群服務器發(fā)生故障時自動切換到這些副本，以便在出現(xiàn)故障時消息仍然可用。Kafka 的復制是以分區(qū)為粒度的，分區(qū)的預寫日志被復制到 n 個服務器。 在 n 個副本中，一個副本作為 leader，其他副本成為 followers。顧名思義，producer 只能往 leader 分區(qū)上寫數(shù)據(jù)(讀也只能從 leader 分區(qū)上進行)，followers 只按順序從 leader 上復制日志。

日志復制算法(log replication algorithm)必須提供的基本保證是，如果它告訴客戶端消息已被提交，而當前 leader 出現(xiàn)故障，新選出的 leader 也必須具有該消息。在出現(xiàn)故障時，Kafka 會從失去 leader 的 ISR 里面選擇一個 follower 作為這個分區(qū)新的 leader ;換句話說，是因為這個 follower 是跟上 leader 寫進度的。

每個分區(qū)的 leader 會維護一個 ISR。當 producer 往 Broker 發(fā)送消息，消息先寫入到對應 leader 分區(qū)上，然后復制到這個分區(qū)的所有副本中。只有將消息成功復制到所有同步副本(ISR)后，這條消息才算被提交。由于消息復制延遲受到最慢同步副本的限制，因此快速檢測慢副本并將其從 ISR 中刪除非常重要。 Kafka 復制協(xié)議的細節(jié)會有些細微差別。

Kafka的同步機制

Kafka不是完全同步，也不是完全異步，而是一種ISR(In-Sync Replicas)機制：

1. leader會維護一個與其基本保持同步的Replica列表，該列表稱為ISR，每個Partition都會有一個ISR，而且是由leader動態(tài)維護 。

2. 如果一個follower比一個leader落后太多，或者超過一定時間未發(fā)起數(shù)據(jù)復制請求，則leader將其從ISR中移除

3. 當ISR中所有Replica都向Leader發(fā)送ACK時，leader才commit，這時候producer才能認為一個請求中的消息都commit了。

Kafka提供了數(shù)據(jù)復制算法保證，如果leader發(fā)生故障或掛掉，一個新leader被選舉并被接受客戶端的消息成功寫入。Kafka確保從同步副本列表中選舉一個副本為leader，或者說follower追趕leader數(shù)據(jù)。leader負責維護和跟蹤ISR中所有follower滯后的狀態(tài)。當Producer發(fā)送一條消息到Broker后，leader寫入消息并復制到所有follower。消息提交之后才被成功復制到所有的同步副本。消息復制延遲受最慢的follower限制，重要的是快速檢測慢副本，如果follower“落后”太多或者失效，leader將會把它從ISR中刪除。

消息傳輸保障

前面已經(jīng)介紹了Kafka如何進行有效的存儲，以及了解了Producer和Consumer如何工作。接下來討論的是Kafka如何確保消息在Producer和Consumer之間傳輸。有以下三種可能的傳輸保障(delivery guarantee):

• At most once: 消息可能會丟，但絕不會重復傳輸
• At least once：消息絕不會丟，但可能會重復傳輸
• Exactly once：每條消息肯定會被傳輸一次且僅傳輸一次

Kafka的消息傳輸保障機制非常直觀。當Producer向Broker發(fā)送消息時，一旦這條消息被commit，由于副本機制(replication)的存在，它就不會丟失。但是如果Producer發(fā)送數(shù)據(jù)給Broker后，遇到的網(wǎng)絡問題而造成通信中斷，那producer就無法判斷該條消息是否已經(jīng)提交(commit)。雖然Kafka無法確定網(wǎng)絡故障期間發(fā)生了什么，但是Producer可以retry多次，確保消息已經(jīng)正確傳輸?shù)紹roker中，所以目前Kafka實現(xiàn)的是At least once。

Consumer從Broker中讀取消息后，可以選擇Commit，該操作會在Zookeeper中存下該Consumer在該Partition下讀取的消息的offset。該Consumer下一次再讀該Partition時會從下一條開始讀取。如未commit，下一次讀取的開始位置會跟上一次commit之后的開始位置相同。當然也可以將consumer設置為autocommit，即Consumer一旦讀取到數(shù)據(jù)立即自動commit。如果只討論這一讀取消息的過程，那Kafka是確保了exactly once, 但是如果由于前面Producer與Broker之間的某種原因?qū)е孪⒌闹貜停敲催@里就是At least once。

考慮這樣一種情況，當Consumer讀完消息之后先commit再處理消息，在這種模式下，如果consumer在commit后還沒來得及處理消息就crash了，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息，這就對應于At most once了。讀完消息先處理再commit。這種模式下，如果處理完了消息在commit之前Consumer crash了，下次重新開始工作時還會處理剛剛未commit的消息，實際上該消息已經(jīng)被處理過了，這就對應于at least once。

要做到exactly once就需要引入消息去重機制。Kafka文檔中提及GUID(Globally Unique Identifier)的概念，通過客戶端生成算法得到每個消息的unique id，同時可映射至broker上存儲的地址，即通過GUID便可查詢提取消息內(nèi)容，也便于發(fā)送方的冪等性保證，需要在broker上提供此去重處理模塊，目前版本尚不支持。

針對GUID, 如果從客戶端的角度去重，那么需要引入集中式緩存，必然會增加依賴復雜度，另外緩存的大小難以界定。不只是Kafka, 類似RabbitMQ以及RocketMQ這類商業(yè)級中間件也只保障at least once, 且也無法從自身去進行消息去重。所以我們建議業(yè)務方根據(jù)自身的業(yè)務特點進行去重，比如業(yè)務消息本身具備冪等性，或者借助Redis等其他產(chǎn)品進行去重處理。

Kafka作為消息隊列：

傳統(tǒng)的消息有兩種模式：隊列和發(fā)布訂閱。 在隊列模式中，消費者池從服務器讀取消息(每個消息只被其中一個讀取); 發(fā)布訂閱模式：消息廣播給所有的消費者。這兩種模式都有優(yōu)缺點，隊列的優(yōu)點是允許多個消費者瓜分處理數(shù)據(jù)，這樣可以擴展處理。但是，隊列不像多個訂閱者，一旦消息者進程讀取后故障了，那么消息就丟了。而發(fā)布和訂閱允許你廣播數(shù)據(jù)到多個消費者，由于每個訂閱者都訂閱了消息，所以沒辦法縮放處理。

Kafka中的Consumer Group有兩種形式：

a、隊列：允許同名的消費者組成員共同處理。

b、發(fā)布訂閱：廣播消息給多個消費者組。

Kafka的每個topic都具有這兩種模式。

傳統(tǒng)的消息系統(tǒng)按順序保存數(shù)據(jù)，如果多個消費者從隊列消費，則服務器按存儲的順序發(fā)送消息，但是，盡管服務器按順序發(fā)送，多個并行請求將會是異步的，因此消息可能亂序到達。這意味著只要消息存在并行消費的情況，順序就無法保證。消息系統(tǒng)常常通過僅設1個消費者來解決這個問題，但是這意味著沒用到并行處理。

Kafka有比傳統(tǒng)的消息系統(tǒng)更強的順序保證。通過并行Topic的Parition，Kafka提供了順序保證和負載均衡。每個Partition僅由同一個消費者組中的一個消費者消費到。并確保消費者是該Partition的唯一消費者，并按順序消費數(shù)據(jù)。每個topic有多個分區(qū)，則需要對多個消費者做負載均衡，但請注意，相同的消費者組中不能有比分區(qū)更多的消費者，否則多出的消費者一直處于空等待，不會收到消息。

Kafka作為存儲系統(tǒng)

所有發(fā)布消息到消息隊列和消費分離的系統(tǒng)，實際上都充當了一個臨時存儲系統(tǒng)。Kafka還是一個非常高性能的存儲系統(tǒng)。寫入到Kafka的數(shù)據(jù)將寫到磁盤并復制到集群中保證容錯性。并允許生產(chǎn)者等待消息應答，直到消息完全寫入。Kafka的存儲結構保證無論服務器上有50KB或50TB數(shù)據(jù)，執(zhí)行效率是相似的，因此可達到水平擴展的目標。還可以認為Kafka是一種專用于高性能，低延遲，提交日志存儲，復制，和傳播特殊用途的分布式文件系統(tǒng)。

Kafka流處理

Kafka的更高目標是實時流處理。在Kafka中，流處理持續(xù)獲取輸入topic的數(shù)據(jù)，進行處理加工，然后寫入輸出topic。例如，一個零售APP，接收銷售和出貨的輸入流，統(tǒng)計數(shù)量或調(diào)整價格后輸出。

簡單的需求可以直接使用Producer和Consumer API進行處理。對于復雜的轉換，Kafka提供了更強大的Streams API，可構建聚合計算或連接流到一起的復雜應用程序。

綜上所述，Kafka 的設計可以幫助我們解決很多架構上的問題。但是想要用好 Kafka 的高性能、低耦合、高可靠性等特性，我們需要非常了解 Kafka，以及我們自身的業(yè)務需求，綜合考慮應用場景。

【本文是51CTO專欄機構“AiChinaTech”的原創(chuàng)文章，微信公眾號( id: tech-AI)”】

戳這里，看該作者更多好文