使用Kafka實現事件溯源

EventSourcing事件溯源是儲存這些實體相關的事件流（實則是明細表），而不是直接儲存實體的“當前”狀態。每個事件都是一個事實，它描述了實體發生的狀態變化（過去時態！）。眾所周知，事實是無可爭議的，不可改變的。

擁有這樣的事件流可以通過摺疊folding與該實體相關的所有事件來找出實體的當前狀態; 但請注意，反過來不可能 - 當僅儲存“當前”狀態時，我們會丟棄許多有價值的歷史資訊。

事件日誌是事實的主要來源：當前狀態始終可以從特定實體的事件流派生。獲取事件流和當前狀態並返回應用這些事件流修改實體後的狀態：Event => State => State。這是一個函式式管道，對於這樣的狀態修改函式，如果有一個初始狀態值，那麼當前狀態是事件流的摺疊folding。（狀態修改函式需要是純粹的，以便可以多次自由地應用於相同的事件)。

在卡夫卡儲存事件

要解決的第一個問題是如何在卡夫卡儲存事件？有三種可能的策略：

1. 將所有實體型別的所有事件儲存在一個主題topic中（具有多個分割槽）
2. 每個實體型別對應一個主題，例如所有使用者相關事件的單獨主題，所有與產品相關的事件對於一個主題等。
3. 每個實體例項對應一個主題，例如每個單個使用者和每個單個產品的單獨主題

第三種策略是不可行的。如果系統中的每個新使用者都需要建立新主題，我們會得到無限數量的主題。這樣以後進行任何型別的聚合也將非常困難，例如搜尋引擎中中需要索引所有使用者，需要使用大量的主題，此外這種方式並不是所有人都預先知道的。

因此，我們可以在1.和2之間進行選擇。兩者都有其優點和缺點：只需一個主題，就可以更容易地獲得所有事件的全域性檢視。另一方面，對於每個實體型別一個主題，可以分別對每個實體型別流進行分割槽和擴充套件。兩者之間的選擇取決於用例。

還可以以額外儲存為代價來獲取兩者：從實體型別主題all-events主題派生所有事件全域性檢視。

本文我們假設我們正在使用每個實體型別對應單個主題這樣方案，但是很容易推廣到多個主題或實體型別。

（編輯：正如ofollow,noindex" target="_blank">Chris Hunt 在Twitter上所說，Martin Kleppmann撰寫了一篇非常好的文章 ，深入探討了如何為主題和分割槽分配事件）。

基本的事件溯源儲存操作

從支援事件源的儲存中獲得的最基本操作是讀取特定實體的“當前”（摺疊folded）狀態。通常，每個實體都有某種形式id。因此，根據此id，我們的儲存系統應可以返回其當前狀態。

事件日誌是事實的主要來源：當前狀態始終可以從特定實體的事件流派生。為了做到這一點，儲存引擎需要一個純粹的（無副作用）函式，獲取事件和當前狀態並返回修改後的狀態：Event => State => State。給定這樣的函式和初始狀態值，當前狀態是事件流的摺疊。（狀態修改函式需要是純粹的，以便可以多次自由地應用於相同的事件。）

Kafka中“讀取當前狀態”操作的簡單實現將從主題中流式傳輸所有事件，過濾它們以僅包括指定事件id並使用指定摺疊函式摺疊它們。如果存在大量事件（並且隨著時間的推移，事件的數量僅增加），這可能是一個緩慢且耗費資源的操作。即使結果將快取在服務節點的記憶體中，仍然需要定期重新建立，例如由於節點故障或快取逐出。

因此，我們需要一種更好的方法。這就是卡夫卡流 （Kafka-streams）和狀態儲存發揮作用的地方。Kafka-streams在一組節點上執行，這些節點共同消費一些主題。與常規Kafka使用者consumer一樣，為每個節點分配了消費主題的多個分割槽。但是，kafka-streams為資料提供了更高級別的操作，從而可以更輕鬆地建立派生流。

kafka-streams一個高階操作是可以將流摺疊到本地儲存中。每個本地儲存僅包含指定節點使用的分割槽中的資料。開箱即用的本地儲存實現有兩種：記憶體中的實現和基於各種資料庫的實現。

回到事件源，我們可以將事件流摺疊到狀態儲存資料庫中，儲存每個實體的“當前狀態”到本地資料庫中。

以下是使用Java API（serde代表序列化器/反序列化器）將摺疊事件轉換為本地儲存的方式：

KStreamBuilder builder = <b>new</b> KStreamBuilder();
builder.stream(keySerde, valueSerde, <font>"my_entity_events"</font><font>)
.groupByKey(keySerde, valueSerde)
</font><font><i>// the folding function: should return the new state</i></font><font>
.reduce((currentState, event) -> ..., </font><font>"my_entity_store"</font><font>);
.toStream(); </font><font><i>// yields a stream of intermediate states</i></font><font>
<b>return</b> builder;
</font>

有關完整示例，請檢視Confluent 的訂單微服務示例 。

（編輯：正如Sergei Egorov 和Nikita Salnikov 在Twitter上注意到的，對於事件採購設定，您可能希望更改預設的Kafka保留設定，以便基於時間或基於大小的更新限制生效，並可選擇啟用壓實compaction。）

查詢當前的狀態

我們現在建立了一個狀態儲存，其中包含來自分割槽的所有實體的當前狀態，但是如何查詢它？如果查詢是本地的（同一節點），那麼它非常簡單：

streams
.store(<font>"my_entity_store"</font><font>, QueryableStoreTypes.keyValueStore());
.get(entityId);
</font>

但是，如果我們想要查詢另一個節點上存在的資料呢？我們如何找出它是哪個節點？在這裡，最近為Kafka引入的另一個功能是：互動式查詢。使用它們，可以查詢Kafka的元資料並找出哪個節點處理給定id的主題分割槽（這使用幕後的主題分割槽器）：

metadataService
.streamsMetadataForStoreAndKey(<font>"my_entity_store"</font><font>, entityId, keySerde)
</font>

然後是將請求轉發到適當的節點。請注意，如何處理和實現節點間通訊 - REST，akka-remote或任何其他方式 - 這些已經超出了kafka-streams的範圍。Kafka只允許訪問本地狀態儲存，並提供有關給定狀態儲存的主機的資訊id。

故障轉移

狀態儲存看起來不錯，但如果節點出現故障會怎麼樣？為某個分割槽重新建立本地狀態儲存也可能是一項昂貴的操作。由於kafka-stream重新平衡（在新增或刪除節點之後），它可能導致長時間延遲或請求失敗。

這就是為什麼預設情況下持久狀態儲存需要logged日誌的原因：也就是說，對儲存的所有更改都會另外寫入changelog-topic。這個主題是緊湊的（我們只需要每個的最新條目id，沒有變化的歷史，因為歷史被儲存在事件中），因此儘可能小。多虧了這一點，在另一個節點上重新建立儲存可以更快。

但這仍然可能導致重新平衡的延遲。為了進一步減少它們，kafka-streams可以選擇為每個儲存保留許多備用副本（num.standby.replicas）。這些副本在應用時會使用更改日誌主題中的所有更新，並且只要當前分割槽失敗，就可以該分割槽下其他複製副本節點充當主要狀態儲存。

一致性

使用預設設定，Kafka提供至少一次交付。也就是說，在節點故障的情況下，可能會多次傳遞某些訊息。例如，如果系統在寫入狀態儲存更改日誌之後但在提交該特定事件的偏移量之前失敗，則會將事件實現狀態儲存兩次。這可能不是問題：我們的狀態更新函式（Event => State => State）可以很好地應對這種情況。但這也不必; 在這種情況下，我們可以利用Kafka的一次性保證 。這些確切的保證只適用於讀寫Kafka主題，但這就是我們在這裡所做的一切：更新狀態儲存的更改日誌和提交偏移都是Kafka主題在幕後寫的 ，這些可以在事務中完成。

因此，如果我們的狀態更新函式需要，我們可以使用一個配置選項processing.guarantee開啟精確一次的流處理，這會導致效能下降，但是 - 沒有任何東西是免費的。

監聽事件

現在我們已經涵蓋了查詢和更新每個實體的“當前狀態” - 如果需要其功能（副作用）怎麼樣？在某些時候，這將是必要的，例如：

• 傳送通知電子郵件
• 在搜尋引擎中索引實體
• 通過REST（或SOAP，CORBA等）呼叫外部服務）

所有這些任務都以某種方式阻塞並涉及I / O（這是副作用的本質），因此將它們作為狀態更新邏輯的一部分執行可能不是一個好主意：這可能會導致速率增加“主”事件迴圈中的失敗併產生效能瓶頸。

此外，狀態更新邏輯函式（Event => State => State）可以多次執行（在故障或重啟的情況下），並且通常我們希望最小化多次執行指定某個事件的副作用的數量。

幸運的是，在我們處理Kafka主題時，我們有很大的靈活性。更新狀態儲存的流階段可以傳送未更改的事件（或者，如果需要，可以修改），並且可以以任意方式消費該結果流/主題（在Kafka中，主題和流是等效的）。而且，可以在狀態更新階段之前或之後消費它。最後，如果我們想要至少一次或最多一次執行副作用，我們也可以控制。只有在副作用成功完成後才能通過提交消耗的事件主題的偏移量來實現至少一次。相反，在最多一次，通過提交偏移前執行副作用。

至於如何執行副作用，根據使用情況，有許多選項。首先，我們可以定義一個Kafka-streams階段，它為每個事件執行副作用，作為流處理函式的一部分。這很容易設定，但是當涉及重試，偏移管理和同時執行許多事件的副作用時，這不是一個非常靈活的解決方案。在更高階的情況下，使用例如reactive-kafka 或其他“直接”Kafka主題消費者來定義處理可能更合適。

一個事件也可能觸發其他事件  - 例如“訂單”事件可能觸發“準備發貨”和“通知客戶”事件。這也可以使用kafka-streams階段來實現。

最後，如果我們想將事件或從事件中提取的一些資料儲存在資料庫或搜尋引擎中，例如ElasticSearch或PostgreSQL，我們可能會使用Kafka Connect連線 器來處理我們所有的主題消耗細節。 

建立檢視和預測

通常，系統的要求不僅僅是查詢和處理單個實體流。還需要支援組合多個事件流的聚合。這種聚合流通常稱為投影projections，摺疊後可用於建立資料檢視。是否可以使用Kafka實現這一點？

再次請記住，在基本級別，我們只是處理儲存我們事件的Kafka主題：因此，我們擁有“原始”Kafka消費者/生產者，kafka-streams組合者甚至KSQL的所有權力 來定義預測。例如，使用kafka-streams，我們可以使用程式碼或類似SQL的KSQL 過濾流，對映，按鍵分組，聚合時間或會話視窗等。

這些流可以持久儲存並可用於使用狀態儲存和互動式查詢進行查詢，就像我們對單個實體流所做的那樣。

走得更遠

隨著系統的發展，為了防止事件流無限增長，某種形式的壓縮或儲存“當前狀態” 快照可能會派上用場。這樣，我們只能儲存一些最近的快照以及它們之後發生的事件。

雖然Kafka 沒有對快照的直接支援，就像其他一些事件溯源系統的情況一樣，使用一些已經提到的機制，例如流，消費者，狀態儲存，絕對有可能新增這種功能。

加起來

雖然Kafka最初的設計並未考慮到事件源，但它是作為資料流引擎設計，主題可複製，可分割槽，狀態可儲存和流API等都非常靈活。因此，可以在Kafka之上實施一個事件溯源系統而不需要太多努力。此外，由於幕後總是有Kafka主題，我們可以獲得額外的靈活性，可以使用高階流API或低級別消費者