事件源，最終一致性，微服務(wù)，CQRS等等，這些越多越多的概念被現(xiàn)代開發(fā)者所熟悉。從細(xì)粒度的服務(wù)組裝到復(fù)雜的以業(yè)務(wù)為中心的應(yīng)用架構(gòu)，這其中最重要的一塊就是以中間件為基礎(chǔ)的業(yè)務(wù)脫藕。本文我們介紹中間件基礎(chǔ)構(gòu)建塊——事務(wù)流。其主導(dǎo)者是Apache Kafka，事實(shí)上的事務(wù)流平臺(tái)標(biāo)準(zhǔn)，還會(huì)介紹Kafka的一個(gè)Web界面工具Kafdrop。

概述

事務(wù)流平臺(tái)屬于更廣泛的面向消息的中間件(MoM)類，與傳統(tǒng)的消息隊(duì)列和主題類似，但是由于日志結(jié)構(gòu)的不變性，它提供了更強(qiáng)大的時(shí)間保證和大幅度性能提高。簡而言之，由于事務(wù)流的寫操作只限于順序追加，所以更加高效。

傳統(tǒng)消息隊(duì)列(MQ)中的消息往往是任意排序的，并且通常彼此獨(dú)立，而流中的事務(wù)(或記錄)往往是按時(shí)間順序或因果關(guān)系排序的。而且，事務(wù)流會(huì)保留其記錄，而MQ一旦讀取了一條消息，就會(huì)丟棄它。因此，事務(wù)流往往更適合事件驅(qū)動(dòng)的體系結(jié)構(gòu)，包括事件源，最終一致性和CQRS等(當(dāng)然，也包括FIFO消息隊(duì)列，但是FIFO隊(duì)列和成熟的事務(wù)流平臺(tái)之間的差異非常大，而不僅限于訂購)。

事務(wù)流平臺(tái)是MoM領(lǐng)域中相對(duì)較新的范例。與數(shù)百種MQ風(fēng)格的消息代理相比較，只有少數(shù)幾種主流可用。與已建立的標(biāo)準(zhǔn)(例如AMQP，MQTT，XMPP和JMS)相比，事務(wù)流空間中還沒有與之等效的標(biāo)準(zhǔn)。

事務(wù)流平臺(tái)是當(dāng)前持續(xù)研究和實(shí)驗(yàn)的活躍領(lǐng)域。但是，事務(wù)流平臺(tái)不僅僅是一個(gè)商用產(chǎn)品，或者復(fù)雜的學(xué)術(shù)問題。它可以廣泛應(yīng)用于消息傳遞和事務(wù)場(chǎng)景，可用于例行性替換消息隊(duì)列的傳統(tǒng)使用場(chǎng)景。

架構(gòu)概述

下圖簡要概述了Kafka組件體系結(jié)構(gòu)。此處限于篇幅，我們不詳細(xì)介紹Kafka內(nèi)部工作原理。

kafka組成

Kafka是一個(gè)分布式系統(tǒng)，包含如下幾個(gè)關(guān)鍵組件：

Broker(代理)節(jié)點(diǎn)：負(fù)責(zé)批量I/O操作和集群內(nèi)的持續(xù)持久化。代理附加日志文件，這些文件中包含由集群托管的主題分區(qū)。可以在多個(gè)代理之間復(fù)制分區(qū)，以實(shí)現(xiàn)水平可伸縮性和增加的持久性，這些復(fù)制的分區(qū)被稱為副本。有一個(gè)代理節(jié)點(diǎn)為控制節(jié)點(diǎn)(控制者)，其他副本受其管理(追隨者)。一個(gè)代理節(jié)點(diǎn)會(huì)被選舉為集群控制器，負(fù)責(zé)分區(qū)狀態(tài)的內(nèi)部管理，還負(fù)責(zé)仲裁給定分區(qū)的領(lǐng)導(dǎo)者跟隨者角色。

ZooKeeper節(jié)點(diǎn)：在后臺(tái)Kafka需要一種方法來管理集群中總體控制器狀態(tài)。如果控制器出于某種原因退出，則有一個(gè)協(xié)議可以從剩余的代理集中選出另一個(gè)控制器。ZooKeeper很大程度上實(shí)現(xiàn)了控制器選舉，心跳等的實(shí)際機(jī)制。ZooKeeper還充當(dāng)各種配置存儲(chǔ)庫，維護(hù)集群元數(shù)據(jù)，領(lǐng)導(dǎo)者和跟隨者狀態(tài)，配額，用戶信息，ACL和其他內(nèi)部管理項(xiàng)目。由于底層的選舉和共識(shí)協(xié)議，ZooKeeper節(jié)點(diǎn)的數(shù)量必須為奇數(shù)。

生產(chǎn)者：負(fù)責(zé)將消息發(fā)布到Kafka主題的客戶端應(yīng)用程序。由于Kafka具有日志結(jié)構(gòu)的性質(zhì)，并且能夠在多消費(fèi)者生態(tài)系統(tǒng)之間共享主題，因此只有生產(chǎn)者才能修改底層日志文件中的數(shù)據(jù)。實(shí)際I/O由代理節(jié)點(diǎn)代表生產(chǎn)者客戶端執(zhí)行。可以將任意數(shù)量生產(chǎn)者消息發(fā)布到同一Kafka主題，并選擇用于保存記錄的分區(qū)。

消費(fèi)者：從主題讀取消息的客戶端應(yīng)用程序。任意數(shù)量的消費(fèi)者都可以從同一主題中閱讀內(nèi)容;但是，根據(jù)消費(fèi)者的配置和分組，存在一些規(guī)則來管理消費(fèi)者之間的記錄分配。

分區(qū)，記錄、偏移量和主題

分區(qū)是記錄的完全有序序列，每一個(gè)分區(qū)對(duì)應(yīng)一個(gè)append日志，這是Kafka的基礎(chǔ)。每一條記錄具有一個(gè)ID：64位整數(shù)偏移量和毫秒級(jí)的時(shí)間簽。它可能會(huì)存在一個(gè)鍵和一個(gè)值。兩者都是字節(jié)數(shù)組，并且都是可選的。術(shù)語"完全排序"僅表示對(duì)于任何給定的生產(chǎn)者，記錄將按照應(yīng)用程序發(fā)出的順序進(jìn)行寫入。如果記錄P在Q之前發(fā)布，則P將在分區(qū)中的Q之前。(假設(shè)P和Q共享一個(gè)分區(qū)。)此外，所有消費(fèi)者將以相同的順序讀取它們。對(duì)于每個(gè)可能的消費(fèi)者，將始終在Q之前讀取P。在大多數(shù)用例中，這種訂購保證至關(guān)重要。通常，已發(fā)布的記錄將與某些現(xiàn)實(shí)事務(wù)相對(duì)應(yīng)，并且保留這​​些事務(wù)的時(shí)間表通常是必不可少的。

記錄的偏移量是分區(qū)中一條記錄的唯一標(biāo)識(shí)分。偏移量是稀疏地址空間中嚴(yán)格單調(diào)遞增的整數(shù)，每個(gè)記錄偏移量始終高于其上一個(gè)記錄偏移量，并且相鄰偏移量之間可能存在可變的間隙。如果啟用了壓縮或作為事務(wù)的結(jié)果，則必然會(huì)存在間隙，所以偏移量也有可能不是連續(xù)的。

應(yīng)用程序不應(yīng)嘗試從字面上解釋偏移量，也不應(yīng)該猜測(cè)下一個(gè)偏移量是多少。但是，可以根據(jù)偏移量推斷任何記錄的相對(duì)順序，按記錄的偏移量對(duì)記錄進(jìn)行排序。

下圖顯示了內(nèi)部分區(qū)的結(jié)構(gòu)：

第一個(gè)偏移量(也稱為low-water mark，低水位標(biāo)記)是要顯示給消費(fèi)者的第一個(gè)消息。由于Kafka的保留期限制，因此不一定是第一個(gè)發(fā)布的消息。可以根據(jù)時(shí)間和/或分區(qū)大小來修剪記錄。當(dāng)有這種情況發(fā)生時(shí)，低水位線似乎會(huì)后移，早于低水位線的記錄將被截?cái)唷?/p>

主題是分區(qū)的邏輯組成。一個(gè)主題可以具有一個(gè)或多個(gè)分區(qū)，而一個(gè)分區(qū)只能有一個(gè)主題或者一個(gè)主題的部分。主題是Kafka的基礎(chǔ)，允許并行和負(fù)載平衡。前面我們說過分區(qū)顯示總順序。由于主題內(nèi)的分區(qū)是相互獨(dú)立的，因此稱該主題具有部分順序。簡而言之，這意味著某些記錄可以互相排序，而相對(duì)于某些其他記錄則不可排序。總順序和部分順序的概念雖然聽起來有些學(xué)術(shù)化，但在構(gòu)建性能事務(wù)流管道中非常重要。它使我們能夠在可能的地方并行處理記錄，同時(shí)在必須的地方保持順序。稍后，我們將探討記錄順序，消費(fèi)者并行性和主題大小的概念。

實(shí)例：消息發(fā)布

實(shí)踐是檢驗(yàn)真理的唯一標(biāo)準(zhǔn)，我將理論付諸實(shí)踐，通過實(shí)例說明概念。我們將啟動(dòng)一對(duì)Docker容器，一個(gè)用Kafka容器，另一個(gè)為Kafdrop容器。我們使用Docker Compose方式啟用容器。

在選定目錄中創(chuàng)建一個(gè)docker-compose.yaml文件，內(nèi)容如下：

為了方便起見，我們用obsidiandynamics/kafka鏡像，它會(huì)將Kafka和ZooKeeper巧妙地打包在一個(gè)鏡像中。然后通過docker-compose up啟動(dòng)容器。啟動(dòng)成功后，可以通過瀏覽器中訪問localhost:9000，就能看到Kafdrop登陸界面。

實(shí)例中是一個(gè)單代理集群，還沒有任何主題。我們可以使用Kafka的命令行工具創(chuàng)建一個(gè)主題并發(fā)布一些消息。我們可以使用docker exec工具對(duì)kafka容器進(jìn)行操作方便地調(diào)用內(nèi)置的CLI工具：

docker exec -it kafka-kafdrop_kafka_1 bash

上面的命令將讓我么進(jìn)入容器的shell命令行界面。工具位于/opt/kafka/bin目錄中，cd進(jìn)入該目錄：

創(chuàng)建一個(gè)名為streams-intro的主題，其中包含3個(gè)分區(qū)：

切換回Kafdrop界面，現(xiàn)在我們就能在列表中看到新主創(chuàng)建的主題。

接著，我們可以使用kafka-console-producer工具發(fā)布消息：

注意：kafka-topics使用--bootstrap-server參數(shù)來配置Kafka代理列表，而kafka-console-producer則使用--broker-list。

記錄由換行符分隔。鍵和值部分由冒號(hào)分隔，如key.separator屬性所指示。本例下，我們可以輸入下內(nèi)容：

完成后，按CTRL + D鍵完成消息發(fā)布。然后切換回Kafdrop，然后單擊streams-intro主題。將看到該主題的概述以及基礎(chǔ)分區(qū)的詳細(xì)分類：

我們創(chuàng)建了一個(gè)包含三個(gè)分區(qū)的主題。然后，我們使用兩個(gè)唯一的鍵foo和bar發(fā)布了五條記錄。Kafka使用鍵將記錄映射到分區(qū)，這樣具有相同鍵的所有記錄將始終出現(xiàn)在同一分區(qū)上。很方便，也很重要，它可以使發(fā)布者指定準(zhǔn)確的記錄順序。稍后，我們將更詳細(xì)地討論鍵哈希和分區(qū)分配。

查看分區(qū)表，分區(qū)#0的第一個(gè)和最后一個(gè)偏移分別為0和2。分區(qū)#2的值為零和3個(gè)，而分區(qū)#1的顯示為空白。在Kafdrop網(wǎng)絡(luò)用戶界面中單擊#0，會(huì)將會(huì)轉(zhuǎn)到主題查看器：

可以看到在bar鍵下發(fā)布的兩條記錄。注意，它們與foo記錄完全無關(guān)。

消費(fèi)者和消費(fèi)組

上面我們實(shí)例講了，聽過生產(chǎn)者發(fā)布消息，將記錄發(fā)送到流中。這些記錄被組織成井井有條的分區(qū)。Kafka的發(fā)布-訂閱拓?fù)渥裱`活的多到多模型，所以，可以有任意數(shù)量的生產(chǎn)者和消費(fèi)者同時(shí)與流進(jìn)行交互。根據(jù)實(shí)際的解決方案，流拓?fù)湟部梢砸粚?duì)多，多對(duì)一。下面我們講，如何消費(fèi)這些記錄。

消費(fèi)者是通過客戶端庫連接到Kafka集群的進(jìn)程或線程。消費(fèi)者通常(但不一定)是一個(gè)整體消費(fèi)組的成員。該組由group.id屬性指定。消費(fèi)組實(shí)際上是Kafka中的負(fù)載平衡機(jī)制，負(fù)責(zé)在組內(nèi)的各個(gè)消費(fèi)者實(shí)例之間大致平均地進(jìn)行分區(qū)分配。當(dāng)組中的第一個(gè)消費(fèi)者訂閱該主題時(shí)，它將收到該主題中的所有分區(qū)。當(dāng)?shù)诙€(gè)消費(fèi)者隨后加入時(shí)，它將獲得大約一半的分區(qū)，從而減輕了第一個(gè)使用者的負(fù)擔(dān)。當(dāng)消費(fèi)者離開時(shí)(通過斷開連接或超時(shí))，該過程將反向進(jìn)行，其余的使用者將可用更多數(shù)量的分區(qū)。

因此，消費(fèi)者消費(fèi)某個(gè)主題中的記錄，從Kafka及其所屬的其他消費(fèi)者分配的分區(qū)中提取了份額。就負(fù)載平衡而言，這應(yīng)該非常簡單。但是，這里有一個(gè)關(guān)鍵點(diǎn)，使用記錄的行為并不能將其刪除。起初這似乎是矛盾的，特別是如果將消耗行為與消耗聯(lián)系起來。(如果有的話，應(yīng)該將消費(fèi)者稱為"閱讀者"。)一個(gè)簡單的事實(shí)是，消費(fèi)者對(duì)主題及其分區(qū)絕對(duì)沒有任何影響。主題是僅追加，只能由生產(chǎn)者或Kafka本身(作為壓縮或清除的一部分)進(jìn)行追加記錄。消費(fèi)者的只讀操作是"便宜的"，因此，可以讓許多人在不增加集群負(fù)擔(dān)的情況下tail日志。這是事務(wù)流和傳統(tǒng)消息隊(duì)列之間的又一區(qū)別，這是至關(guān)重要的。

消費(fèi)者在內(nèi)部維護(hù)一個(gè)偏移量，該偏移量指向分區(qū)中的下一個(gè)記錄，從而在每次連續(xù)讀取時(shí)都增加偏移量。消費(fèi)者首次訂閱主題時(shí)，可以選擇從主題的頭端或尾端開始。通過將auto.offset.reset屬性設(shè)置為latest, earliest 或者none，可以控制這個(gè)行為。在后一種情況下，如果消費(fèi)者組不存在先前的偏移量，則將觸發(fā)異常。

消費(fèi)者在本地保留其偏移狀態(tài)向量。由于不同消費(fèi)組中的消費(fèi)者不會(huì)互相干擾，因此可能有許多人同時(shí)閱讀同一主題。消費(fèi)者按照自己的偏移讀取消息;緩慢的或積壓的消費(fèi)者對(duì)其同組其他人也不會(huì)有影響。

為了說明這個(gè)概念，我們考慮一個(gè)包含兩個(gè)分區(qū)的主題為場(chǎng)景。兩個(gè)消費(fèi)者組-A和B-訂閱了該主題。每個(gè)組具有三個(gè)實(shí)例，使用者被命名為A1，A2，A3，B1，B2和B3。下圖說明了兩組如何共享主題，以及消費(fèi)者如何彼此獨(dú)立地瀏覽記錄。

仔細(xì)看上圖，會(huì)發(fā)現(xiàn)缺少某些東西。消費(fèi)者A3和B1不在上圖中。這是因?yàn)镵afka保證分區(qū)只能分配給其消費(fèi)組中的一個(gè)消費(fèi)者。由于每個(gè)組中有三個(gè)消費(fèi)者，但是只有兩個(gè)分區(qū)，因此一個(gè)消費(fèi)者將保持空閑狀態(tài)，等待其所在組中的另一個(gè)消費(fèi)者離開。以這種方式，消費(fèi)組不僅是負(fù)載平衡機(jī)制，而且還是用于建立高性能管道而又不犧牲安全性的類似柵欄的排他性控制，特別是在要求只能由一個(gè)線程處理記錄的情況下或在任何給定時(shí)間進(jìn)行處理。

消費(fèi)組也用于確保可用性。通過定期從主題中提取記錄，消費(fèi)者可以向集群隱式反饋集群為"健康"狀態(tài)，從而將租約擴(kuò)展到其分區(qū)分配上。但是，如果消費(fèi)者未能在允許的期限內(nèi)再次閱讀，則將其視為有缺陷，并且將重新分配其分區(qū)，分配給該組中其余的"健康"消費(fèi)者。該截止日期由max.poll.interval.ms在消費(fèi)者客戶端屬性控制，默認(rèn)情況下設(shè)置為五分鐘。

用交通系統(tǒng)來做個(gè)類比，主題就像是高速公路，分區(qū)就是車道。記錄就是等同于汽車，其乘客對(duì)應(yīng)于記錄值。只要保持行車路線，幾輛車就可以安全地在同一條高速公路上行駛。共享相同線路的汽車按順序行駛，形成隊(duì)列。現(xiàn)在，假設(shè)每條車道通向一個(gè)匝道，將其流量轉(zhuǎn)移到某個(gè)位置。如果一個(gè)匝道堆積了，其他匝道可能仍能順暢流動(dòng)。

Kafka正是利用這種機(jī)制確保端到端的吞吐量，輕松地實(shí)現(xiàn)每秒達(dá)到數(shù)百萬條記錄的QPS。創(chuàng)建主題時(shí)，可以選擇分區(qū)計(jì)數(shù)，通道數(shù)。分區(qū)在一個(gè)消費(fèi)組中的各個(gè)消費(fèi)者之間大致均勻地劃分，并確保不會(huì)將分區(qū)同時(shí)分配給兩個(gè)(或多個(gè))消費(fèi)者。

注意：創(chuàng)建后，可以通過增加分區(qū)數(shù)來調(diào)整主題的大小。但是，無法在不重新創(chuàng)建主題的情況下減少分區(qū)數(shù)。

記錄對(duì)應(yīng)于事件、消息、命令或任何其他可流式傳輸?shù)膬?nèi)容。記錄的精確劃分方式由生產(chǎn)者決定。生產(chǎn)者可以在發(fā)布記錄時(shí)顯式分配分區(qū)索引，盡管這種方法很少使用。正如我們?cè)谇懊娴氖纠兴龅哪菢樱环N更常見的方法是為記錄分配鍵。鍵對(duì)Kafka完全不透明，換句話說，Kafka不會(huì)去解釋key的內(nèi)容，而是將其視為字節(jié)數(shù)組。使用一致的哈希技術(shù)對(duì)這些字節(jié)進(jìn)行哈希處理以得出分區(qū)索引。

共享相同散列的記錄可以保證占據(jù)相同的分區(qū)。假設(shè)一個(gè)主題具有多個(gè)分區(qū)，則具有不同鍵的記錄可能最終會(huì)位于不同的分區(qū)中。但是，由于哈希鍵沖突，具有不同哈希值的記錄也可能最終會(huì)在同一分區(qū)中。

生產(chǎn)者無需關(guān)心記錄將映射到哪個(gè)特定分區(qū)，只要相關(guān)記錄最終在同一分區(qū)中并且保留其順序。同樣，消費(fèi)者對(duì)也無需關(guān)心分配到那個(gè)分區(qū)，只要它們以與發(fā)布相同的順序接收記錄，并且其分區(qū)分配不會(huì)與組中的其他消費(fèi)者重復(fù)。

案例：交易平臺(tái)

假設(shè)我們正在尋找上市股票的特定價(jià)格模式，并在確定特定模式后發(fā)出交易信號(hào)。有大量庫存，可以理解的是，希望將它們并行處理。但是，任何給定的股票代碼的時(shí)間序列必須在單個(gè)使用者上順序處理。

Kafka使這個(gè)用例以及其他類似用例幾乎不容易實(shí)現(xiàn)。我們將創(chuàng)建兩個(gè)主題：價(jià)格，用來存儲(chǔ)原始價(jià)格數(shù)據(jù)。訂單主題，用來保存任何由產(chǎn)生的訂單。我們可以多劃分一些分區(qū)，可以讓我們充分的并行操作。

我們可以在價(jià)格主題上發(fā)布每個(gè)價(jià)格的記錄，并用股票代碼作為鍵。Kafka的自動(dòng)分區(qū)分配將確保每個(gè)股票代號(hào)由其組中的一個(gè)消費(fèi)者處理。消費(fèi)者實(shí)例可以自由擴(kuò)展和擴(kuò)展以匹配處理負(fù)載。消費(fèi)者組應(yīng)該有意義地命名，理想地反映消費(fèi)應(yīng)用程序的目的。比如trading-strategy.abc，它是一種名為" ABC"的虛擬交易策略。

消費(fèi)者確定了價(jià)格模式后，就可以在訂單主題上發(fā)布另一條消息，訂單請(qǐng)求。我們將召集另一個(gè)消費(fèi)組，訂單執(zhí)行，負(fù)責(zé)讀取訂單并將其轉(zhuǎn)發(fā)給經(jīng)紀(jì)人。

在這個(gè)簡單的示例中，我們創(chuàng)建了一個(gè)完全由事件驅(qū)動(dòng)且高度可擴(kuò)展的端到端交易管道，假設(shè)沒有其他瓶頸。我們可以在各個(gè)階段動(dòng)態(tài)添加更多的處理節(jié)點(diǎn)，以應(yīng)對(duì)需要增加的負(fù)載的情況。

假設(shè)您需要在通用數(shù)據(jù)源的驅(qū)動(dòng)下同時(shí)運(yùn)行的幾種交易策略。此外，交易策略將由不同的團(tuán)隊(duì)制定;目的是盡可能地使這些實(shí)現(xiàn)脫鉤，從而使團(tuán)隊(duì)能夠自主運(yùn)作，甚至可以使用不同的編程語言和工具鏈以各自的節(jié)奏進(jìn)行開發(fā)和部署。

Kafka靈活的多到多pub-sub體系結(jié)構(gòu)將狀態(tài)消耗與廣播語義相結(jié)合。通過使用不同的消費(fèi)群體，Kafka允許不同的應(yīng)用程序共享輸入主題，并按自己的進(jìn)度處理事件。第二種交易策略將需要一個(gè)專門的消費(fèi)群體：trading-strategy.xyz，將其特定的業(yè)務(wù)邏輯應(yīng)用于通用定價(jià)流，并將生成的訂單發(fā)布到相同的訂單主題。通過這種方式，Kafka能夠從易于重用和組合的離散元素構(gòu)建模塊化事件處理管道。

總結(jié)

事務(wù)流平臺(tái)是構(gòu)建模塊化、松耦合、事件驅(qū)動(dòng)的應(yīng)用程序的高效構(gòu)建塊。在事務(wù)流的世界中，Kafka鞏固了其成為成功開源解決方案的地位，該解決方案兼顧了靈活性和高性能。并發(fā)和并行性是Kafka體系結(jié)構(gòu)的核心，形成部分排序的事務(wù)流，可以在可擴(kuò)展的消費(fèi)者生態(tài)系統(tǒng)中實(shí)現(xiàn)負(fù)載平衡。消費(fèi)者及其周圍組的簡單重新配置可以帶來非常不同的事件分配和處理語義。

當(dāng)然，Kafka并非沒有缺陷。可以說Kafka工具是低于標(biāo)準(zhǔn)的。大多數(shù)Kafka從業(yè)者長期以來都放棄了現(xiàn)成的CLI實(shí)用程序，而轉(zhuǎn)向其他開源工具(例如Kafdrop，Kafkacat和第三方商業(yè)產(chǎn)品(如Kafka Tool)。 Kafka的配置選項(xiàng)種類繁多，初學(xué)的默認(rèn)值有坑，新手可能會(huì)踩到這些坑。總而言之，Kafka代表了構(gòu)建和構(gòu)建復(fù)雜系統(tǒng)的方式的范式轉(zhuǎn)變，它的優(yōu)勢(shì)也是顯而易見的。