6 種事件驅動的架構模式

作者 | Natan Silnitsky

譯者 | 平川

策劃 | 萬佳

在過去一年裏，我一直是數據流團隊的一員，負責 Wix 事件驅動的消息傳遞基礎設施（基於 Kafka）。有超過 1400 個微服務使用這個基礎設施。在此期間，我實現或目睹了事件驅動消息傳遞設計的幾個關鍵模式，這些模式有助於創建一個健壯的分佈式系統，該系統可以輕鬆地處理不斷增長的流量和存儲需求。

1 消費與投影

MetaSite 服務處理大約 1M RPM 的各類請求

我們想要回答的問題是，如何以最終一致的方式將讀請求從該服務轉移出來？

 使用 Kafka 創建 “物化視圖”

負責這項服務的團隊決定另外創建一個服務，只處理 MetaSite 的一個關注點——來自客戶端服務的 “已安裝應用上下文” 請求。

• 首先，他們將所有數據庫的站點元數據對象以流的方式傳輸到 Kafka 主題中，包括新站點創建和站點更新。一致性可以通過在 Kafka Consumer 中進行 DB 插入來實現，或者通過使用 CDC 產品（如 Debezium）來實現。

• 其次，他們創建了一個有自己數據庫的 “只寫” 服務（反向查找寫入器），該服務使用站點元數據對象，但只獲取已安裝應用上下文並寫入數據庫。即將站點元數據的某個“視圖”（已安裝的應用程序）投影到數據庫中。

已安裝應用上下文消費與投影

• 第三，他們創建了一個 “只讀” 服務，只接受與已安裝應用上下文相關的請求，通過查詢存儲着 “已安裝應用程序” 視圖的數據庫來滿足請求。

讀寫分離

 效果

• 通過將數據以流的方式傳輸到 Kafka，MetaSite 服務完全同數據消費者解耦，這大大降低了服務和 DB 的負載。

• 通過消費來自 Kafka 的數據，併爲特定的上下文創建一個 “物化視圖”，反向查找寫入器服務能夠創建一個最終一致的數據投影，大幅優化了客戶端服務的查詢需求。

• 將讀服務與寫服務分開，可以方便地擴展只讀 DB 副本和服務實例的數量，這些實例可以處理來自全球多個數據中心的不斷增長的查詢負載。

2 端到端事件驅動

針對簡單業務流程的狀態更新

請求 - 應答模型在瀏覽器 - 服務器交互中特別常見。藉助 Kafka 和 WebSocket，我們就有了一個完整的事件流驅動，包括瀏覽器 - 服務器交互。

這使得交互過程容錯性更好，因爲消息在 Kafka 中被持久化，並且可以在服務重啓時重新處理。該架構還具有更高的可伸縮性和解耦性，因爲狀態管理完全從服務中移除，並且不需要對查詢進行數據聚合和維護。

考慮一下這種情況，將所有 Wix 用戶的聯繫方式導入 Wix 平臺。

這個過程涉及到兩個服務：Contacts Jobs 服務處理導入請求並創建導入批處理作業，Contacts Importer 執行實際的格式化並存儲聯繫人（有時藉助第三方服務）。

傳統的請求 - 應答方法需要瀏覽器不斷輪詢導入狀態，前端服務需要將狀態更新情況保存到數據庫表中，並輪詢下游服務以獲得狀態更新。

而使用 Kafka 和 WebSocket 管理者服務，我們可以實現一個完全分佈式的事件驅動過程，其中每個服務都是完全獨立工作的。

使用 Kafka 和 WebSocket 的 E2E 事件驅動

首先，瀏覽器會根據開始導入請求訂閱 WebSocket 服務。

它需要提供一個 channel-Id，以便 WebSocket 服務能夠將通知路由回正確的瀏覽器：

打開 WebSocket 通知 “通道”

第二，瀏覽器需要向 Jobs 服務發送一個 HTTP 請求，聯繫人信息使用 CSV 格式，並附加 channel-Id，這樣 Jobs 服務（和下游服務）就能夠向 WebSocket 服務發送通知。注意，HTTP 響應將立即返回，沒有任何內容。

第三，Jobs 服務在處理完請求後，會生成並向 Kafka 主題發送作業請求。

HTTP Import 請求和生成的 Import Job 消息

第四，Contacts Importer** 服務消費來自 Kafka 的作業請求，並執行實際的導入任務。當它完成時，它可以通知 WebSocket 服務作業已經完成，而 WebSocket 服務又通知瀏覽器。

工作已消費、已處理和已完成狀態通知

 效果

• 使用這種設計，在導入過程的各個階段通知瀏覽器變得很簡單，而且不需要保持任何狀態，也不需要任何輪詢。

• Kafka 的使用使得導入過程更具彈性和可擴展性，因爲多個服務可以處理來自同一個原始導入 http 請求的作業。

• 使用 Kafka 複製，很容易將每個階段放在最合適的數據中心和地理位置。也許導入器服務需要在谷歌 DC 上，以便可以更快地導入谷歌聯繫人。

• WebSocket 服務的傳入通知請求也可以生成到 Kafka，然後複製到 WebSocket 服務所在的數據中心。

3 內存 KV 存儲

針對 0 延遲數據訪問

有時，我們需要動態對應用程序進行持久化配置，但我們不想爲它創建一個全面的關係數據庫表。

一個選擇是用 HBase/Cassandra/DynamoDB 爲所有應用創建一個大的寬列存儲表，其主鍵包含標識應用域的前綴（例如 “store_taxes_”）。

這個解決方案效果很好，但是通過網絡取值存在無法避免的延遲。它更適合於更大的數據集，而不僅僅是配置數據。

另一種方法是有一個位於內存但同樣具有持久性的鍵 / 值緩存——Redis AOF 提供了這種能力。

Kafka 以壓縮主題的形式爲鍵 / 值存儲提供了類似的解決方案（保留模型確保鍵的最新值不會被刪除）。

在 Wix，我們將這些壓縮主題用作內存中的 kv-store，我們在應用程序啓動時加載（消費）來自主題的數據。這有一個 Redis 沒有提供的好處，這個主題還可以被其他想要獲得更新的用戶使用。

 訂閱和查詢

考慮以下用例——兩個微服務使用壓縮主題來做數據維護：Wix Business Manager（幫助 Wix 網站所有者管理他們的業務）使用一個壓縮主題存放支持的國家列表，Wix Bookings（允許安排預約和課程）維護了一個 “（Time Zones）” 壓縮主題。從這些內存 KV 存儲中檢索值的延遲爲 0。

各內存 KV 存儲以及相應的 Kafka 壓縮主題

Wix Bookings 監聽 “國家（Countries）” 主題的更新：

Bookings 消費來自壓縮主題 Countries 的更新

當 Wix Business Manager 將另一個國家添加到 “國家” 主題時，Wix Bookings 會消費此更新，並自動爲 “時區” 主題添加一個新的時區。現在，內存 KV 存儲中的 “時區” 也通過更新增加了新的時區：

South Sudan 的時區被加入壓縮主題

我們沒有在這裏停下來。Wix Events（供 Wix Users 管理事件傳票和 RSVP）也可以使用 Bookings 的時區主題，並在一個國家因爲夏令時更改時區時自動更新其內存 kv-store。

兩個內存 KV 存儲消費同一個壓縮主題

4 調度並遺忘

當存在需要確保計劃事件最終被處理的需求時

在許多情況下，需要 Wix 微服務根據某個計劃執行作業。

Wix Payments Subscriptions 服務就是一個例子，它管理基於訂閱的支付（例如瑜伽課程的訂閱）。

對於每個月度或年度訂閱用戶，必須通過支付提供程序完成續訂過程。

爲此，Wix 自定義的 Job Scheduler 服務調用由 Payments Subscription 服務預先配置好的 REST 端點。

訂閱續期過程在後臺進行，不需要（人類）用戶參與。這就是爲什麼最終可以成功續訂很重要，即使臨時有錯誤——例如第三支付提供程序不可用。

要確保這一過程是完全彈性的，一種方法是由作業調度器重複請求 Payment Subscriptions 服務（續訂的當前狀態保存在數據庫中），對每個到期但尚未續期的訂閱進行輪詢。這將需要數據庫上的悲觀 / 樂觀鎖定，因爲同一用戶同一時間可能有多個訂閱續期請求（來自兩個單獨的正在進行的請求）。

更好的方法是首先生成 Kafka 請求。爲什麼？因爲請求的處理將由 Kafka 的消費者順序完成（對於每個特定的用戶），所以不需要並行工作的同步機制。

此外，一旦消息生成併發送到 Kafka，我們就可以通過引入消費者重試來確保它最終會被成功處理。由於有這些重試，請求調度的頻率可能就會低很多。

在這種情況下，我們希望可以保持處理順序，這樣重試邏輯可以在兩次嘗試之間（以 “指數退避” 間隔進行）簡單地休眠。

Wix 開發人員使用我們自定義的 Greyhound 消費者，因此，他們只需指定一個 BlockingPolicy，並根據需要指定適當的重試間隔。

在某些情況下，消費者和生產者之間可能會產生延遲，如長時間持續出錯。在這些情況下，有一個特殊的儀表板用於解除阻塞，並跳過開發人員可以使用的消息。

如果消息處理順序不是強制性的，那麼 Greyhound 中還有一個使用 “重試主題” 的非阻塞重試策略。

當配置重試策略時，Greyhound 消費者將創建與用戶定義的重試間隔一樣多的重試主題。內置的重試生成器將在出錯時生成一條下一個重試主題的消息，該消息帶有一個自定義頭，指定在下一次調用處理程序代碼之前應該延遲多少時間。

還有一個死信隊列，用於重試次數耗盡的情況。在這種情況下，消息被放在死信隊列中，由開發人員手動審查。

這種重試機制是受 Uber 這篇文章的啓發。

https://eng.uber.com/reliable-reprocessing/

Wix 最近開放了 Greyhound 的源代碼，不久將提供給測試用戶。要了解更多信息，可以閱讀 GitHub 上的自述文件。

https://github.com/wix/greyhound#greyhound

總結：

• Kafka 允許按順序處理每個鍵的請求（例如使用 userId 進行續訂），簡化工作進程邏輯；

• 由於 Kafka 重試策略的實現大大提高了容錯能力，續期請求的作業調度頻率大大降低。

5 事務中的事件

當冪等性很難實現時

考慮下面這個典型的電子商務流程。

Payments 服務生成一個 Order Purchase Completed 事件到 Kafka。現在，Checkout 服務將消費此消息，並生成自己的 Order Checkout Completed 消息，其中包含購物車中的所有商品。

然後，所有下游服務（Delivery、Inventory 和 Invoices）將消費該消息並繼續處理（分別準備發貨、更新庫存和創建發票）。

如果下游服務可以假設 Order Checkout Completed 事件只由 Checkout 服務生成一次，則此事件驅動流的實現會簡單很多。

爲什麼？因爲多次處理相同的 Checkout Completed 事件可能導致多次發貨或庫存錯誤。爲了防止下游服務出現這種情況，它們將需要存儲去重後的狀態，例如，輪詢一些存儲以確保它們以前沒有處理過這個 Order Id。

通常，這是通過常見的數據庫一致性策略實現的，如悲觀鎖定和樂觀鎖定。

幸運的是，Kafka 爲這種流水線事件流提供了一個解決方案，每個事件只處理一次，即使當一個服務有一個消費者 - 生產者對（例如 Checkout），它消費一條消息，併產生一條新消息。

簡而言之，當 Checkout 服務處理傳入的 Payment Completed 事件時，它需要將 Checkout Completed 事件的發送過程封裝在一個生產者事務中，它還需要發送消息偏移量（使 Kafka 代理能夠跟蹤重複的消息）。

事務期間生成的任何消息將僅在事務完成後纔對下游消費者（Inventory Service）可見。

此外，位於 Kafka 流開始位置的 Payment Service Producer 必須轉變爲冪等（Idempotent）生產者——這意味着代理將丟棄它生成的任何重複消息。

要了解更多信息，請觀看我的視頻 “Kafka 中的恰好一次語義”。

https://www.youtube.com/watch?v=7O_UC_i1XY0

6 事件聚合

當你想知道整個批次的事件已經被消費時

在上半部分，我描述了在 Wix 將聯繫人導入到 Wix CRM 平臺的業務流程。後端包括兩個服務。一個是作業服務，我們提供一個 CSV 文件，它會生成作業事件到 Kafka。還有一個聯繫人導入服務，它會消費並執行導入作業。

假設 CSV 文件有時非常大，將工作負載分割成更小的作業，每個作業中需要導入的聯繫人就會更少，這個過程就會更高效。通過這種方式，這項工作可以在 Contacts Importer 服務的多個實例中並行。但是，當導入工作被拆分爲許多較小的作業時，該如何知道何時通知最終用戶所有的聯繫人都已導入？

顯然，已完成作業的當前狀態需要持久化，否則，內存中哪些作業已完成的記錄可能會因爲隨機的 Kubernetes pod 重啓而丟失。

一種在 Kafka 中進行持久化的方法是使用 Kafka 壓縮主題。這類主題可以看成是一種流式 KV 存儲。

在我們的示例中，Contacts Importer 服務（在多個實例中）通過索引消費作業。每當它處理完一些作業，就需要用一個 Job Completed 事件更新 KV 存儲。這些更新可以同時發生，因此，可能會出現競態條件並導致作業完成計數器失效。

 原子 KV 存儲

爲了避免競態條件，Contacts Importer 服務將完成事件寫到原子 KV 存儲類型的 Jobs-Completed-Store 中。

原子存儲確保所有作業完成事件將按順序處理。它通過創建一個 “Commands” 主題和一個 “Store” 壓縮主題來實現。

 順序處理

從下圖可以看出，原子存儲如何生成每一條新的 Import-job-completed“更新”消息，並以 [Import Request Id]+[total job count] 作爲鍵。藉助鍵，我們就可以總是依賴 Kafka 將特定 requestId 的 “更新” 放在特定的分區中。

接下來，作爲原子存儲的一部分，消費者 - 生產者對將首先偵聽每個新的更新，然後執行 atomicStore 用戶請求的 “命令”——在本例中，將已完成作業數量的值加 1。

 端到端更新流示例

讓我們回到 Contacts Importer 服務流。一旦這個服務實例完成了某些作業的處理，它將更新 Job-Completed KVAtomicStore（例如，請求 Id 爲 YYY 的導入作業 3 已經完成）：

Atomic Store 將生成一條新消息到 job-completed-commands 主題，鍵爲 YYY-6，值爲 Job 3 Completed。

接下來，Atomic Store 的消費者 - 生產者對將消費此消息，並增加 KV Store 主題中鍵 YYY-6 的已完成作業計數。

 恰好一次處理

注意，“命令” 請求處理必須只發生一次，否則完成計數器可能不正確（錯誤增量）。爲消費者 - 生產者對創建一個 Kafka 事務（如上文的模式 4 所述）對於確保統計準確至關重要。

 AtomicKVStore 值更新回調

最後，一旦 KV 最新生成的已完成作業計數的值與總數匹配（例如 YYY 導入請求有 6 個已完成作業），就可以通知用戶（通過 WebSocket，參見本系列文章第一部分的模式 3）導入完成。通知可以作爲 KV-store 主題生成動作的副作用，即調用用戶提供給 KV 原子存儲的回調。

 注意事項：

• 完成通知邏輯不一定要在 Contacts Importer 服務中，它可以在任何微服務中，因爲這個邏輯完全獨立於這個過程的其他部分，只依賴於 Kafka 主題。

• 不需要進行定期輪詢。整個過程都是事件驅動的，即以管道方式處理事件。

• 通過使用基於鍵的排序和恰好一次的 Kafka 事務，避免作業完成通知或重複更新之間的競態條件。

• Kafka Streams API 非常適合這樣的聚合需求，其特性包括 groupBy（按 Import Request Id 分組）， reduce 或 count（已完成作業計數）和 filter (count 等於總作業數)，然後是副作用 Webhook 通知。對於 Wix 來說，使用現有的生產者 / 消費者基礎設施更有意義，這對我們的微服務拓撲影響更小。

7 總結

這裏的一些模式比其他的模式更爲常見，但它們都有相同的原則。通過使用事件驅動的模式，可以減少樣板代碼（以及輪詢和鎖定原語），增加彈性（減少級聯失敗，處理更多的錯誤和邊緣情況）。此外，微服務之間的耦合要小得多（生產者不需要知道誰消費了它的數據），擴展也更容易，向主題添加更多分區（和更多服務實例）即可。

原文鏈接：

https://medium.com/wix-engineering/6-event-driven-architecture-patterns-part-1-93758b253f47

https://medium.com/wix-engineering/6-event-driven-architecture-patterns-part-2-455cc73b22e1

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/AiaSRKDu5Nn4zrneB_wpLA