6 種事件驅(qū)動的架構(gòu)模式
點擊上方“服務端思維”，選擇“設(shè)為星標”
回復”669“獲取獨家整理的精選資料集
回復”加群“加入全國服務端高端社群「后端圈」
在過去一年里，我一直是數(shù)據(jù)流團隊的一員，負責Wix事件驅(qū)動的消息傳遞基礎(chǔ)設(shè)施（基于 Kafka）。有超過 1400 個微服務使用這個基礎(chǔ)設(shè)施。在此期間，我實現(xiàn)或目睹了事件驅(qū)動消息傳遞設(shè)計的幾個關(guān)鍵模式，這些模式有助于創(chuàng)建一個健壯的分布式系統(tǒng)，該系統(tǒng)可以輕松地處理不斷增長的流量和存儲需求。
1.消費與投影
針對那些使用非常廣泛、已經(jīng)成為瓶頸的服務
當有遺留服務存儲著大型域?qū)ο蟮臄?shù)據(jù)，這些數(shù)據(jù)使用又非常廣泛，使得該遺留服務成為瓶頸時，此模式可以提供幫助。
在 Wix，我們的 MetaSite 服務就面臨著這樣的情況，它為 Wix 用戶創(chuàng)建的每個站點保存了大量的元數(shù)據(jù)，比如站點版本、站點所有者以及站點上安裝了哪些應用程序——已安裝應用上下文（The Installed Apps Context.）。
這些信息對于 Wix 的許多其他微服務（團隊）很有價值，比如 Wix Stores、Wix booking、Wix Restaurants 等等。這個服務被超過 100 萬 RPM 的請求轟炸，它們需要獲取站點元數(shù)據(jù)的不同部分。
從服務的各種 API 可以明顯看出，它處理了客戶端服務的太多不同的關(guān)注點。
MetaSite 服務處理大約 1M RPM 的各類請求
我們想要回答的問題是，如何以最終一致的方式將讀請求從該服務轉(zhuǎn)移出來？
使用 Kafka 創(chuàng)建“物化視圖”
負責這項服務的團隊決定另外創(chuàng)建一個服務，只處理 MetaSite 的一個關(guān)注點——來自客戶端服務的“已安裝應用上下文”請求。
首先，他們將所有數(shù)據(jù)庫的站點元數(shù)據(jù)對象以流的方式傳輸?shù)?Kafka 主題中，包括新站點創(chuàng)建和站點更新。一致性可以通過在 Kafka Consumer 中進行 DB 插入來實現(xiàn)，或者通過使用CDC產(chǎn)品（如Debezium）來實現(xiàn)。
其次，他們創(chuàng)建了一個有自己數(shù)據(jù)庫的“只寫”服務（反向查找寫入器），該服務使用站點元數(shù)據(jù)對象，但只獲取已安裝應用上下文并寫入數(shù)據(jù)庫。即將站點元數(shù)據(jù)的某個“視圖”（已安裝的應用程序）投影到數(shù)據(jù)庫中。
已安裝應用上下文消費與投影
第三，他們創(chuàng)建了一個“只讀”服務，只接受與已安裝應用上下文相關(guān)的請求，通過查詢存儲著“已安裝應用程序”視圖的數(shù)據(jù)庫來滿足請求。
讀寫分離
效果
通過將數(shù)據(jù)以流的方式傳輸?shù)?Kafka，MetaSite 服務完全同數(shù)據(jù)消費者解耦，這大大降低了服務和 DB 的負載。
通過消費來自 Kafka 的數(shù)據(jù)，并為特定的上下文創(chuàng)建一個“物化視圖”，反向查找寫入器服務能夠創(chuàng)建一個最終一致的數(shù)據(jù)投影，大幅優(yōu)化了客戶端服務的查詢需求。
將讀服務與寫服務分開，可以方便地擴展只讀 DB 副本和服務實例的數(shù)量，這些實例可以處理來自全球多個數(shù)據(jù)中心的不斷增長的查詢負載。
2.端到端事件驅(qū)動
針對簡單業(yè)務流程的狀態(tài)更新
請求-應答模型在瀏覽器-服務器交互中特別常見。借助 Kafka 和WebSocket，我們就有了一個完整的事件流驅(qū)動，包括瀏覽器-服務器交互。
這使得交互過程容錯性更好，因為消息在 Kafka 中被持久化，并且可以在服務重啟時重新處理。該架構(gòu)還具有更高的可伸縮性和解耦性，因為狀態(tài)管理完全從服務中移除，并且不需要對查詢進行數(shù)據(jù)聚合和維護。
考慮一下這種情況，將所有 Wix 用戶的聯(lián)系方式導入 Wix 平臺。
這個過程涉及到兩個服務：Contacts Jobs 服務處理導入請求并創(chuàng)建導入批處理作業(yè)，Contacts Importer 執(zhí)行實際的格式化并存儲聯(lián)系人（有時借助第三方服務）。
傳統(tǒng)的請求-應答方法需要瀏覽器不斷輪詢導入狀態(tài)，前端服務需要將狀態(tài)更新情況保存到數(shù)據(jù)庫表中，并輪詢下游服務以獲得狀態(tài)更新。
而使用 Kafka 和 WebSocket 管理者服務，我們可以實現(xiàn)一個完全分布式的事件驅(qū)動過程，其中每個服務都是完全獨立工作的。
使用 Kafka 和 WebSocket 的 E2E 事件驅(qū)動
首先，瀏覽器會根據(jù)開始導入請求訂閱 WebSocket 服務。
它需要提供一個 channel-Id，以便 WebSocket 服務能夠?qū)⑼ㄖ酚苫卣_的瀏覽器：
打開 WebSocket 通知“通道”
第二，瀏覽器需要向 Jobs 服務發(fā)送一個 HTTP 請求，聯(lián)系人信息使用 CSV 格式，并附加 channel-Id，這樣 Jobs 服務（和下游服務）就能夠向 WebSocket 服務發(fā)送通知。注意，HTTP 響應將立即返回，沒有任何內(nèi)容。
第三，Jobs 服務在處理完請求后，會生成并向 Kafka 主題發(fā)送作業(yè)請求。
HTTP Import 請求和生成的 Import Job 消息
第四，Contacts Importer**服務消費來自 Kafka 的作業(yè)請求，并執(zhí)行實際的導入任務。當它完成時，它可以通知 WebSocket 服務作業(yè)已經(jīng)完成，而 WebSocket 服務又通知瀏覽器。
工作已消費、已處理和已完成狀態(tài)通知
效果
使用這種設(shè)計，在導入過程的各個階段通知瀏覽器變得很簡單，而且不需要保持任何狀態(tài)，也不需要任何輪詢。
Kafka 的使用使得導入過程更具彈性和可擴展性，因為多個服務可以處理來自同一個原始導入 http 請求的作業(yè)。
使用 Kafka 復制，很容易將每個階段放在最合適的數(shù)據(jù)中心和地理位置。也許導入器服務需要在谷歌 DC 上，以便可以更快地導入谷歌聯(lián)系人。
WebSocket 服務的傳入通知請求也可以生成到 Kafka，然后復制到 WebSocket 服務所在的數(shù)據(jù)中心。
3.內(nèi)存 KV 存儲
針對 0 延遲數(shù)據(jù)訪問
有時，我們需要動態(tài)對應用程序進行持久化配置，但我們不想為它創(chuàng)建一個全面的關(guān)系數(shù)據(jù)庫表。
一個選擇是用HBase/Cassandra/DynamoDB為所有應用創(chuàng)建一個大的寬列存儲表，其主鍵包含標識應用域的前綴（例如“store_taxes_”）。
這個解決方案效果很好，但是通過網(wǎng)絡(luò)取值存在無法避免的延遲。它更適合于更大的數(shù)據(jù)集，而不僅僅是配置數(shù)據(jù)。
另一種方法是有一個位于內(nèi)存但同樣具有持久性的鍵/值緩存——Redis AOF提供了這種能力。
Kafka 以壓縮主題的形式為鍵/值存儲提供了類似的解決方案（保留模型確保鍵的最新值不會被刪除）。
在 Wix，我們將這些壓縮主題用作內(nèi)存中的 kv-store，我們在應用程序啟動時加載（消費）來自主題的數(shù)據(jù)。這有一個 Redis 沒有提供的好處，這個主題還可以被其他想要獲得更新的用戶使用。
訂閱和查詢
考慮以下用例——兩個微服務使用壓縮主題來做數(shù)據(jù)維護：Wix Business Manager（幫助 Wix 網(wǎng)站所有者管理他們的業(yè)務）使用一個壓縮主題存放支持的國家列表，Wix Bookings（允許安排預約和課程）維護了一個“（Time Zones）”壓縮主題。從這些內(nèi)存 KV 存儲中檢索值的延遲為 0。
各內(nèi)存 KV 存儲以及相應的 Kafka 壓縮主題
Wix Bookings 監(jiān)聽“國家（Countries）”主題的更新：
Bookings 消費來自壓縮主題 Countries 的更新
當 Wix Business Manager 將另一個國家添加到“國家”主題時，Wix Bookings 會消費此更新，并自動為“時區(qū)”主題添加一個新的時區(qū)?，F(xiàn)在，內(nèi)存 KV 存儲中的“時區(qū)”也通過更新增加了新的時區(qū)：
South Sudan 的時區(qū)被加入壓縮主題
我們沒有在這里停下來。Wix Events（供 Wix Users 管理事件傳票和 RSVP）也可以使用 Bookings 的時區(qū)主題，并在一個國家因為夏令時更改時區(qū)時自動更新其內(nèi)存 kv-store。
兩個內(nèi)存 KV 存儲消費同一個壓縮主題
4.調(diào)度并遺忘
當存在需要確保計劃事件最終被處理的需求時
在許多情況下，需要 Wix 微服務根據(jù)某個計劃執(zhí)行作業(yè)。
Wix Payments Subscriptions 服務就是一個例子，它管理基于訂閱的支付（例如瑜伽課程的訂閱）。
對于每個月度或年度訂閱用戶，必須通過支付提供程序完成續(xù)訂過程。
為此，Wix 自定義的 Job Scheduler 服務調(diào)用由 Payments Subscription 服務預先配置好的 REST 端點。
訂閱續(xù)期過程在后臺進行，不需要（人類）用戶參與。這就是為什么最終可以成功續(xù)訂很重要，即使臨時有錯誤——例如第三支付提供程序不可用。
要確保這一過程是完全彈性的，一種方法是由作業(yè)調(diào)度器重復請求 Payment Subscriptions 服務（續(xù)訂的當前狀態(tài)保存在數(shù)據(jù)庫中），對每個到期但尚未續(xù)期的訂閱進行輪詢。這將需要數(shù)據(jù)庫上的悲觀/樂觀鎖定，因為同一用戶同一時間可能有多個訂閱續(xù)期請求（來自兩個單獨的正在進行的請求）。
更好的方法是首先生成 Kafka 請求。為什么？因為請求的處理將由 Kafka 的消費者順序完成（對于每個特定的用戶），所以不需要并行工作的同步機制。
此外，一旦消息生成并發(fā)送到 Kafka，我們就可以通過引入消費者重試來確保它最終會被成功處理。由于有這些重試，請求調(diào)度的頻率可能就會低很多。
在這種情況下，我們希望可以保持處理順序，這樣重試邏輯可以在兩次嘗試之間（以“指數(shù)退避”間隔進行）簡單地休眠。
Wix 開發(fā)人員使用我們自定義的Greyhound消費者，因此，他們只需指定一個 BlockingPolicy，并根據(jù)需要指定適當?shù)闹卦囬g隔。
在某些情況下，消費者和生產(chǎn)者之間可能會產(chǎn)生延遲，如長時間持續(xù)出錯。在這些情況下，有一個特殊的儀表板用于解除阻塞，并跳過開發(fā)人員可以使用的消息。
如果消息處理順序不是強制性的，那么 Greyhound 中還有一個使用“重試主題”的非阻塞重試策略。
當配置重試策略時，Greyhound 消費者將創(chuàng)建與用戶定義的重試間隔一樣多的重試主題。內(nèi)置的重試生成器將在出錯時生成一條下一個重試主題的消息，該消息帶有一個自定義頭，指定在下一次調(diào)用處理程序代碼之前應該延遲多少時間。
還有一個死信隊列，用于重試次數(shù)耗盡的情況。在這種情況下，消息被放在死信隊列中，由開發(fā)人員手動審查。
這種重試機制是受 Uber這篇文章的啟發(fā)。
Wix 最近開放了 Greyhound 的源代碼，不久將提供給測試用戶。要了解更多信息，可以閱讀 GitHub 上的自述文件。
總結(jié)：
Kafka 允許按順序處理每個鍵的請求（例如使用 userId 進行續(xù)訂），簡化工作進程邏輯；
由于 Kafka 重試策略的實現(xiàn)大大提高了容錯能力，續(xù)期請求的作業(yè)調(diào)度頻率大大降低。
5.事務中的事件
當冪等性很難實現(xiàn)時
考慮下面這個典型的電子商務流程。
Payments 服務生成一個 Order Purchase Completed 事件到 Kafka。現(xiàn)在，Checkout 服務將消費此消息，并生成自己的 Order Checkout Completed 消息，其中包含購物車中的所有商品。
然后，所有下游服務（Delivery、Inventory 和 Invoices）將消費該消息并繼續(xù)處理（分別準備發(fā)貨、更新庫存和創(chuàng)建發(fā)票）。
如果下游服務可以假設(shè) Order Checkout Completed 事件只由 Checkout 服務生成一次，則此事件驅(qū)動流的實現(xiàn)會簡單很多。
為什么？因為多次處理相同的 Checkout Completed 事件可能導致多次發(fā)貨或庫存錯誤。為了防止下游服務出現(xiàn)這種情況，它們將需要存儲去重后的狀態(tài)，例如，輪詢一些存儲以確保它們以前沒有處理過這個 Order Id。
通常，這是通過常見的數(shù)據(jù)庫一致性策略實現(xiàn)的，如悲觀鎖定和樂觀鎖定。
幸運的是，Kafka 為這種流水線事件流提供了一個解決方案，每個事件只處理一次，即使當一個服務有一個消費者-生產(chǎn)者對（例如 Checkout），它消費一條消息，并產(chǎn)生一條新消息。
簡而言之，當 Checkout 服務處理傳入的 Payment Completed 事件時，它需要將 Checkout Completed 事件的發(fā)送過程封裝在一個生產(chǎn)者事務中，它還需要發(fā)送消息偏移量（使 Kafka 代理能夠跟蹤重復的消息）。
事務期間生成的任何消息將僅在事務完成后才對下游消費者（Inventory Service）可見。
此外，位于 Kafka 流開始位置的 Payment Service Producer 必須轉(zhuǎn)變?yōu)閮绲龋↖dempotent）生產(chǎn)者——這意味著代理將丟棄它生成的任何重復消息。
要了解更多信息，請觀看我的視頻“Kafka中的恰好一次語義”。
6.事件聚合
當你想知道整個批次的事件已經(jīng)被消費時
在上半部分，我描述了在 Wix 將聯(lián)系人導入到 Wix CRM 平臺的業(yè)務流程。后端包括兩個服務。一個是作業(yè)服務，我們提供一個 CSV 文件，它會生成作業(yè)事件到 Kafka。還有一個聯(lián)系人導入服務，它會消費并執(zhí)行導入作業(yè)。
假設(shè) CSV 文件有時非常大，將工作負載分割成更小的作業(yè)，每個作業(yè)中需要導入的聯(lián)系人就會更少，這個過程就會更高效。通過這種方式，這項工作可以在 Contacts Importer 服務的多個實例中并行。但是，當導入工作被拆分為許多較小的作業(yè)時，該如何知道何時通知最終用戶所有的聯(lián)系人都已導入？
顯然，已完成作業(yè)的當前狀態(tài)需要持久化，否則，內(nèi)存中哪些作業(yè)已完成的記錄可能會因為隨機的 Kubernetes pod 重啟而丟失。
一種在 Kafka 中進行持久化的方法是使用 Kafka 壓縮主題。這類主題可以看成是一種流式 KV 存儲。
在我們的示例中，Contacts Importer 服務（在多個實例中）通過索引消費作業(yè)。每當它處理完一些作業(yè)，就需要用一個 Job Completed 事件更新 KV 存儲。這些更新可以同時發(fā)生，因此，可能會出現(xiàn)競態(tài)條件并導致作業(yè)完成計數(shù)器失效。
原子 KV 存儲
為了避免競態(tài)條件，Contacts Importer 服務將完成事件寫到原子 KV 存儲類型的 Jobs-Completed-Store 中。
原子存儲確保所有作業(yè)完成事件將按順序處理。它通過創(chuàng)建一個“Commands”主題和一個“Store”壓縮主題來實現(xiàn)。
順序處理
從下圖可以看出，原子存儲如何生成每一條新的 Import-job-completed“更新”消息，并以[Import Request Id]+[total job count]作為鍵。借助鍵，我們就可以總是依賴 Kafka 將特定 requestId 的“更新”放在特定的分區(qū)中。
接下來，作為原子存儲的一部分，消費者-生產(chǎn)者對將首先偵聽每個新的更新，然后執(zhí)行 atomicStore 用戶請求的“命令”——在本例中，將已完成作業(yè)數(shù)量的值加 1。
端到端更新流示例
讓我們回到 Contacts Importer 服務流。一旦這個服務實例完成了某些作業(yè)的處理，它將更新 Job-Completed KVAtomicStore（例如，請求 Id 為 YYY 的導入作業(yè) 3 已經(jīng)完成）：
Atomic Store 將生成一條新消息到 job-completed-commands 主題，鍵為 YYY-6，值為 Job 3 Completed。
接下來，Atomic Store 的消費者-生產(chǎn)者對將消費此消息，并增加 KV Store 主題中鍵 YYY-6 的已完成作業(yè)計數(shù)。
恰好一次處理
注意，“命令”請求處理必須只發(fā)生一次，否則完成計數(shù)器可能不正確（錯誤增量）。為消費者-生產(chǎn)者對創(chuàng)建一個 Kafka 事務（如上文的模式 4 所述）對于確保統(tǒng)計準確至關(guān)重要。
AtomicKVStore 值更新回調(diào)
最后，一旦 KV 最新生成的已完成作業(yè)計數(shù)的值與總數(shù)匹配（例如 YYY 導入請求有 6 個已完成作業(yè)），就可以通知用戶（通過 WebSocket，參見本系列文章第一部分的模式 3）導入完成。通知可以作為 KV-store 主題生成動作的副作用，即調(diào)用用戶提供給 KV 原子存儲的回調(diào)。
注意事項：
完成通知邏輯不一定要在 Contacts Importer 服務中，它可以在任何微服務中，因為這個邏輯完全獨立于這個過程的其他部分，只依賴于 Kafka 主題。
不需要進行定期輪詢。整個過程都是事件驅(qū)動的，即以管道方式處理事件。
通過使用基于鍵的排序和恰好一次的 Kafka 事務，避免作業(yè)完成通知或重復更新之間的競態(tài)條件。
Kafka Streams API 非常適合這樣的聚合需求，其特性包括 groupBy（按 Import Request Id 分組）， reduce 或 count（已完成作業(yè)計數(shù)）和 filter (count 等于總作業(yè)數(shù))，然后是副作用 Webhook 通知。對于 Wix 來說，使用現(xiàn)有的生產(chǎn)者/消費者基礎(chǔ)設(shè)施更有意義，這對我們的微服務拓撲影響更小。
總結(jié)
這里的一些模式比其他的模式更為常見，但它們都有相同的原則。通過使用事件驅(qū)動的模式，可以減少樣板代碼（以及輪詢和鎖定原語），增加彈性（減少級聯(lián)失敗，處理更多的錯誤和邊緣情況）。此外，微服務之間的耦合要小得多（生產(chǎn)者不需要知道誰消費了它的數(shù)據(jù)），擴展也更容易，向主題添加更多分區(qū)（和更多服務實例）即可。
— 本文結(jié)束 —
關(guān)注我，回復 「加群」 加入各種主題討論群。
對「服務端思維」有期待，請在文末點個在看
喜歡這篇文章，歡迎轉(zhuǎn)發(fā)、分享朋友圈
