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