流式處理：使用 Apache Kafka的Streams API 實現 Rabobank 的實時財務告警

原文：https://www.confluent.io/blog/real-time-financial-alerts-rabobank-apache-kafkas-streams-api/

原作者：Jeroen van Disseldorp

本文討論使用 Apache Kafka 的 Streams API 向 Rabobank 的客戶發送告警。Rabobank（荷蘭合作銀行）總部位於荷蘭，在全球擁有 900 多個分支機構，48,000 名員工和 681 億歐元的資產。Rabobank 是一家由客戶和銀行組成的合作銀行，一家對社會負責的銀行。其目標是成為荷蘭金融市場的領導者。Rabobank 還致力於成為全球食品和農業領域的領先銀行。Rabobank 向全球數百萬客戶提供金融產品和服務。

在過去的幾年中，Rabobank 一直在為成為一家實時的，事件驅動的銀行而進行積極的投資。如果你熟悉銀行的業務流程，應該會明白這並不能一蹴而就。許多銀行業務流程都是在非商用硬體上以批處理作業的形式進行的，因此遷移工作非常艱巨。但如前所述，Rabobank 接受了這一挑戰

，並定義了一個業務事件匯流排（BEB，Business Event Bus），用於應用程序之間共享整個組織架構的業務事件。

Rabobank 選擇 Apache Kafka 作為底層的主要引擎，並編寫了自己的的 BEB 客戶端庫，以方便應用程序開發人員使用簡單的消息生產/消費以及災難恢復等功能。

Rabobank 採用 的 Kafka 設置，Kafka 集群在多個數據中心進行對稱鏡像。當數據中心出現故障或由操作人員干預後，BEB 客戶端（包括本文討論的基於 Kafka Streams 的應用程序）會切換到另一個 Kafka 集群，而無需重新啟動。實現在災難情景和計劃維護時段內的 24×7 不間斷運行。BEB 客戶端庫為生產者、消費者以及流式應用提供了這種切換機制。

Rabo Alerts是一個由一系列生產、消費、流式消息等微服務組成的系統，基於 BEB 實現。下面討論的所有數據類型和代碼都可以在 GitHub 中找到。本文將在一定程度上簡化源碼清單（如刪除未使用的欄位），但這些清單仍反映了生產中實際運行的代碼。

案例：Rabo Alerts

Rabo Alerts 服務可以讓 Rabobank 的客戶接收其關注的財務事件告警。例如某筆款項從賬戶中扣除或者記入賬戶，以及其它更複雜的事件。客戶可以根據自己的偏好配置告警，並通過第三方渠道發送：如電子郵件、簡訊和移動推送通知。值得一提的是，Rabo Alerts 並不是一項新的或試用服務，它已經投產十多年，可供數百萬賬戶持有者試用。

面臨的問題

舊的 Rabo Alerts 實現主要是在大型機系統上。所有的處理步驟都是面向批處理的，大型機會根據告警類型派生告警，並每隔幾分鐘發送，但每天只發送幾次。這種實現非常穩定可靠，但 Rabobank 希望解決兩個問題：（1）靈活性不足；（2）告警發送速度慢。

由於對現有告警進行更改或添加新（更智能）的告警需要很大的工作量，因此舊的 Rabo Alerts 對適應新業務需求的靈活性很低。在過去幾年中，Rabobank 在其在線環境中引入新功能的步伐大幅增加，舊有僵化的告警解決方案變得越來越成問題。

告警的傳遞速度也是一個問題，舊的 Rabo Alerts 可能需要 5 分鐘到 4-5 小時才能向用戶發送告警（取決於告警類型和批處理窗口）。如果在十年前，這個速度可能足夠快了，但如今客戶的期望值要高得多。現在 Rabobank 向客戶提供「相關信息」的時間窗口要比過去十年小得多。

因此，如何重新設計現有的機制，使其具有更強的擴展性及更快的速度，便是擺在眼前的問題。當然，重新設計的 Rabo Aerts 也需要穩定可靠，以便能夠正確地為現有數百萬的用戶群提供服務。

從小處著手

在過去的一年裡，我們一直使用 Kafka 及其 Streams API 重新設計和實現告警機制。由於整個 Rabo Alerts 服務相當龐大，我們決定從四個簡單但使用率高的告警開始：

餘額高於閾值

餘額低於閾值

超過閾值的貸記（Credit）

超過閾值的借記（Debit）

這些告警的每一個都可以從當前賬戶系統的支付信息流中派生出來。例如：「當我的餘額低於 100 歐元時向我發送簡訊」或「當有人給我超過 1000 歐元時向我推送消息」（通常用於存款通知）。

以下截圖說明如何通過手機銀行 app 配置 Rabo Alerts ——

告警拓撲

我們的第一步是重新設計告警過程，基本流程如下：

挖掘來自支付工廠的交易流，產生一連串的 AccountEntry（賬戶會計條目）。注意，每一個支付交易總是由兩個 AccountEntry 組成，即借記（Debit）和貸記（Credit）。

對每個 AccountEntry 執行以下步驟：

i. 查看客戶是否為給定賬號配置了 Rabo Alert。

ii. 如果是，檢查此 AccountEntry 是否符合客戶設置的告警條件。

iii. 如何符合，通過客戶配置的渠道（電子郵件、簡訊、消息推送）發送告警。

a. 將具有賬戶讀取許可權的賬號轉換為一個客戶列表。

b. 對每個客戶執行以下步驟：

步驟 1 需要與執行交易的核心銀行系統建立鏈接。

步驟 2a 需要建立一個查詢表，其中包含所有賬戶的所有客戶許可權。

步驟 2b 需要建立一個查詢表，其中包含所有客戶的 Rabo Alert 設置。

該流程的使用及其需求見下圖：

圖中所有白色框都是 Kafka 主題（Topic），其中列出了它們的 Avro 鍵/值數據類型。大部分數據類型都是不言自明的，但以下數據類型值得一提：

CustomerAlertSettings：特定客戶的告警設置，這些設置包括：

CustomerAlertAddresses：客戶用於接收告警消息的渠道及地址列表。移動推送地址此處以 表示，因為註冊移動設備的實際列表是在消息發送過程中確定的。

CustomerAccountAlertSettings：客戶為特定賬戶設定的告警配置列表。這個列表指定了客戶希望接收特定賬戶的哪些告警及其閾值。

ChannelType：可用的渠道類型枚舉，當前為 EMAIL、PUSH 和 SMS。

AccountEntry：一條支付賬戶的會計記賬。一個記賬條目是一個支付交易的一半，可以是一個借記條目（Debit），也可以是一個貸記條目（Credit）。

OutboundMessage：發送給客戶的消息內容。包含消息類型和參數，但不包含其定址。這些信息由 topic 的 Key 承載。

藍色框表示獨立的應用程序（或稱微服務），是使用 Spring Boot 實現的可執行 jar，並部署在託管平台上。它們一起組成了實現 Rabo Alerts 的所有必要功能：

Alert Settings Manager：告警配置管理器。向一個 （開啟了 Log Compaction 的 Kafka 主題）發布每個客戶的所有自定義告警設置。

Account Authorization Manager：賬戶授權管理器。賬戶並不是和客戶一對一綁定，而是可以由不同的用戶查看。例如，配偶之間共享賬戶；或企業賬戶針對不同員工的不同授權。這些情況下可能會產生任意的賬戶/用戶間的授權關係。該應用程序向一個 發布賬號和授權客戶ID的關係。它是實時的，以便授權的變化在發送告警時能立即生效。

Account Entry Bridge：通過 IBM MQ 從 Rabobank 基於大型機的支付工廠中檢索所有支付流，並轉發到 Kafka 的 topic。

Alerting：核心告警服務，參見下文。

Device Resolver：設備解析器，輔助應用。從外部系統查找所有客戶的移動設備，並將相同的告警消息寫入各個設備對應的 topic 中（PushId）。客戶移動設備的查找可以通過一個 完成，但由於各種不同的原因，此處是通過遠程服務調用的方式實現的。

Senders：每一個 Sender 消費其綁定的渠道 topic 的消息，並發送給定址客戶。每種渠道都被分配了各自的 Kafka topic，以使各種渠道的故障能彼此分離。例如，當電子郵件伺服器關閉時，告警消息依然可以通過消息推送的方式發送出去。

廢話少說，放碼過來

使用 Kafka Streams 編碼實現告警只需要 2 個類。

第一個類是 。這個類使用給定的 定義主要的 Kafka Streams 拓撲（Topology）。它實現了 BEB 的 ，是一個 BEB 客戶端庫使用的自定義介面，用於在應用程序啟動後或 Kafka 集群切換（如數據中心切換/故障轉移）時生成新的 Kafka Streams Topology。

該 Topology 定義了以下幾個步驟：

1-13 行，從消費 AccountEntry 流開始，當檢索到一個 AccountEntry 時，會查找哪些客戶有權訪問該賬戶，並將結果存儲在一個中間 topic 中，以 為 Key， 為 Value。該 topic 的意思是「這個客戶（Key）的這個 AccountEntry（Value）需要處理」。

14-20 行，針對每個客戶執行。檢查客戶的告警設置，如果 AccountEntry 符合客戶的告警設置，會要求輔助類生成 。

22-39 行，遍歷所有的 ，並將它們分配到各自的渠道 topic。

告警消息是在 17 行調用輔助類 生成的。其主要方法是 ，該方法獲取一個 AccountEntry，並從具有該賬戶查看許可權的客戶中獲取告警配置。以下是它的代碼：

該方法執行以下步驟：

13 行，流化所有賬戶相關的告警設置（一個賬戶一個對象）。

14 行，將告警設置中的賬號和 AccountEntry 中的賬號進行匹配。

15 行，流化告警設置中的各項設置。

16-21 行，構造要發送的一系列消息，以及用於發送消息的渠道列表（這裡對每種告警類型都使用了單獨的方法），結果是一個 流。

22 行，過濾空結果。

23-24 行，為指定渠道查找客戶地址，並返回一個 流。

25 行，收集所有結果，並作為 返回。

這個類的其它輔助方法：

：通過比較賬號和幣種，來匹配 AccountEntry 和賬戶告警設置。

：生成 告警消息（餘額高於/低於閾值）。

：生成 告警消息（超出閾值的借記/貸記）。

：查找指定渠道列表的客戶告警接收地址。

：構建一個 。

再加上一些其它額外的類來將這個功能包裝在一個獨立的應用程序中，這就是它的全部功能！

第一次測試運行

在第一次的初步實現後，我們進行了測試運行。事實令我們驚訝，並且期望值高漲。從支付訂單確認到移動設備收到告警的整個過程只需要一到兩秒鐘，且一秒的情況居多。這個過程還包括了支付工廠所花費的時間（驗證支付訂單，交易處理），因此響應時間可能會依當時的支付工廠工作量而有所不同。整個告警鏈——從 AccountEntry 在 Kafka 上發布，到將消息推送給客戶——通常在 120 毫秒內完成。在發送階段，推送（PUSH）告警是最快的，僅需 100-200 毫秒即可到達客戶的移動設備。電子郵件（EMAIL）和簡訊（SMS）稍慢，通常在發出消息後的 2-4 秒到達。相比之下，舊有的體系通常需要幾分鐘的時間才能提供告警。

下面的視頻演示了使用我的個人測試賬戶進行告警傳輸的速度。請注意，雖然是測試用的，但這也是一個正常運行的 Rabobank 支付賬戶！

【只是一段演示視頻，markdown 插入視頻比較麻煩，原文看吧，或看下面的文字解說】

首先我在我的設備上啟用了告警，並配置了閾值為 0 的 DebitedAboveThreshold 告警（「More withdrawn than」）。這意味著超過 0 歐元的任何支付都會向我發送告警。我設置了 PUSH 和 SMS 兩種渠道告警（視頻中未演示），因此告警會通過兩個渠道發給我。保存設置並返回主屏幕後，我開始向我的同事 Joris Meijer 轉賬 1 歐元，並通過指紋驗證。之後付款訂單被發送到支付工廠進行處理。在訂單確認關閉之前，推送（PUSH）告警已經在屏幕頂部彈出，如通知窗口所示。幾秒鐘後，相同的告警消息也以 SMS 的方式到達。

回顧

新機制簡潔而優雅，只需要少數 Java 類組成。這個邏輯大約四個星期寫完，但要使整個拓撲工作需要大約六個月的時間。這主要是因為 、 以及 需要和銀行的其它業務模塊達成一致。

在團隊內部的告警測試之後，需要更徹底更大規模的測試。畢竟我們希望確保客戶不會錯過告警或接收到不該接收告警。我們選用了 25,000 名 Rabobank 的員工作為試點小組，對這個新機制進行了為期兩個月的試用。這樣可以更好的觀察系統在生產數據及高負載下的運行表現。另外，Rabobank 的員工比付費客戶更能容忍告警失敗（有時確實會失敗）的情況。在試用期間，我們優化了告警生成並消除了一些外圍應用的邊界錯誤。

經批准，新體系於 6 月 8 日上線為數百萬 Rabobank 客戶提供服務。這對我們來說是非常激動人心的時刻——不僅因為它有效，而且因為我們永遠不可能回頭。我們通過延遲幾秒而不是幾分鐘或幾小時的告警，有效提升了客戶的期望值。如果由於某種原因導致某個組件服務失敗，客戶會立即注意到，因為告警會延遲。因此我們密切關注這套體系，但到目前為止，它一直運行良好且可預測。

下一步

新體系提供了實時告警，且易於擴展，滿足了 Rabobank 對於速度和靈活性的要求。但這裡提到的四種告警類型並不是全部。客戶還可以配置其它約 10 種告警，例如「當我收到來自指定賬戶的付款時提醒我」和「當付款單無法執行時提醒我」。下一步是將這些告警從大型機遷移到新體系，但這需要連接更多的支付系統，例如支付訂單執行引擎。我們將在未來的幾個月為此努力，且不會止於此，新的實現也激發了大量新的想法，我們將很快公開討論（甚至展示）。

關於作者

關於 Apache Kafka Streams API

如果你喜歡本文，可能會希望繼續使用以下資源了解有關 Apache Kafka Streams API 的更多信息：

關於譯文

本譯文經原作者授權後，首發於 K棧（kweny.io）。轉載請註明原作者以及原文和譯文出處。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！