如何構建以事件驅動型實時信息系統

作者：陳峻譯 2020-09-22 07:00:00

本文向您介紹如何WSO2 API Manager和Kafka，基于事件驅動的方法，來構建實時的信息系統，以便提高資源的使用率和企業的競爭力。

【51CTO.com快譯】在如今競爭激烈的商業競爭環境中，數據的處理往往需要具有實時性。如果競爭對手在數據管控上事先采取了行動，那么勢必會在業務上取得一定的優勢。這也正是我們需要構建能夠實時處理信息(數據)的企業級系統的重要原因。在實時處理數據時，我們除了可以將系統設計為以異步的方式，對事件進行操作之外，還可以使用同步的請求響應消息，來構建實時的系統。同時，為了保持此類系統能夠高效地使用資源，我們需要了解如何基于事件驅動的方法，來構建實時的信息系統。

圖：實時信息系統解決方案架構

如上圖所示，各類事件通常來源于包括移動和Web應用在內的不同渠道。其中：

接收事件組件，負責在任何給定的時間內處理大量的數據，其速率從每秒上千個事件、到每秒數上百萬個事件不等。在接收到事件之后，該組件會將其傳遞給對應的處理組件。
事件處理組件，負責對事件中帶有的數據進行操作，包括：過濾，清理，轉換和匯總等。根據組件的行為，事件處理既可以獨立于事件接收組件、被單獨地執行，也可以作為相關的操作來完成。事件處理組件可以將原始事件存儲在各種數據商店(data store)中，以便進行審核(如果未被接收組件處理的話)，并且可以根據用例，將計算結果存儲在那些單獨的數據商店之中。因此，大多數事件都能夠被實時處理，并發布到事件發布組件中，以傳遞給使用者(consumer)。
事件發布組件，負責將處理后的數據實時地推送給使用者。這些使用者既可以是移動或Web應用，又可以是對已處理事件起作用的其他系統。除了這些實時事件的發布之外，在某些用例中，我們還需要通過HTTP通道，以同步、或請求-響應的方式，將處理后的摘要信息發布給移動和Web應用。

上述參考架構可用于那些需要處置現場正在發生的事件，并將其發布到后臺應用等多種用例中。例如：在出現緊急情況時，現場人員可以實時地發送有關需求與狀況的細節信息，而后臺團隊則能夠毫不拖延地進行必要的物資調配與派送。此外，我們也可以使用此類架構來構建農業的供應鏈。例如：農民們通過運貨車輛將農作物運送到連鎖超市。農民可以在農作物準備就緒時，更新其詳細信息。而超市后臺團隊則會實時地從各個位置獲取更新，并安排車輛及時地收集農作物，以避免延遲。

使用WSO2和Kafka的參考架構

消息代理是將消息發送者與接收者相分離的組件。目前，市場上有很多消息代理類產品，它們各有優、缺點。其中最流行的當屬Kafka、NATS和RabbitMQ。當然，Kafka也可以作為NATS和RabbitMQ的最佳功能性代理。

在此，我們選擇Kafka作為事件消息的代理;選擇功能豐富、簡單且開源的WSO2Streaming Integrator作為事件處理器;使用既支持流媒體、又支持REST風格的WSO2 API Manager，作為事件發布者。當然，這些組件也可以被市場上的其他類似工具所替換。下圖展示了構成實時事件驅動型信息系統的各個組件，及其相互連接。

圖：具有代理和WSO2平臺的實時事件驅動型信息系統

在該架構中，事件代理會接收來自移動和Web應用等源頭的事件負載。WSO2 Streaming Integrator會處理這些事件，然后將各種結果事件通過WebSocket連接，發布到WSO2 API Manager上。WSO2 API Manager擁有一個公布給網關的WebSocket API，諸如移動和Web之類的consumer(消費者)應用會使用該API，實時地接收各種事件。同時，WSO2 Streaming Integrator可以將原始事件和匯總的結果，通過標準的REST API，從WSO2 API Manager處公布給相關的consumer。下圖對上述架構進行了細化。

圖：具有Kafka和WSO2平臺詳細信息的實時事件驅動型信息系統

如上圖所示，事件源通過Kafka客戶端，將事件發布到Kafka代理中那些可用的topic(主題)處。WSO2 Streaming Integrator不但可以訂閱這些topic，還能通過已配置的Kafka源，實時地使用來自Kafka的各種事件。由Siddhi語言編寫的各項操作將處理這些事件，并傳遞給諸如WebSocket之類的事件sink(接收器)。同時，WSO2 SI會按需通過各種數據商店(data stores)，將事件存儲到對應的數據庫表中。

WSO2 API Manager通過WebSocket API來將WebSocket sink的詳細信息配置到API的端點上。據此，那些使用WebSocket API的客戶端應用將能實時地接收到已經處理的各種事件。

同時，那些已處理的信息和原始事件會被存儲到一個通過WSO2 Enterprise Integrator公布了REST數據服務的數據庫中。此處的數據服務是通過將WSO2 API Manager作為受保護的REST API予以公布，并通過客戶端應用實現同步通信的服務。此外，作為一種能夠支持大多數企業系統需求的成熟架構，我們可以通過擴展，來支持諸如：混合集成需求、API管理平臺等多種企業用例。

只有WSO2平臺，沒有Kafka的參考架構

如果貴組織剛開始著手構建實時的事件驅動型信息系統，而且數據負載量并不大的話，那么就可以僅使用WSO2平臺，來構建前文提到的精簡版架構。下圖展示了一種沒有消息代理的實現方式。

圖：具有WSO2平臺的實時事件驅動型信息系統

該架構與前文提到的架構之間唯一的區別在于：雖然缺少事件代理，但是客戶端應用能夠通過HTTP的調用，將事件直接發送到WSO2 Streaming Integrator處。當然，由于該架構沒有消息代理，因此WSO2 SI需要將原始事件存儲在數據庫中，以供各項審核。而它的其余功能則與前文的架構相同。下圖展示了該架構的詳細組成結構。

圖：具有WSO2平臺詳細信息的實時事件驅動型信息系統

如上圖所示，WSO2 SI被配置為通過HTTP接口來接收事件。而Siddhi應用中的HTTP源則被配置為通過不同的操作，來處理各種事件，然后發布到WebSocket sink中。同時，各種原始事件通過數據商店被存儲在數據庫中，并將各種聚合的結果通過不同的數據商店存儲到另一個數據表里。除此之外，該系統的其余功能與前文提到的基于代理的實現方式基本一致。

從Kafka到Websocket Siddhi應用的示例代碼

下面我們將給出一個Siddhi的應用示例。它能夠從Kafka的topic中讀取事件，并通過WebSocket服務器，將各種事件發布(或輸出)到某個日志sink處。當然，在發布之前，它會對每個事件進行簡單地檢查(或篩選)，以確保其數量小于500。具體代碼請參見--https://gist.github.com/chanakaudaya/efe8dfed2558811f0316a7839dbfef57。其中，您可以找到有關如何使用Streaming Integrator，來設置Kafka的詳細示例。同時，您也可以通過文檔鏈接--https://ei.docs.wso2.com/zh_CN/latest/streaming-integrator/examples/working-with-kafka/，來試運行該Siddhi應用。

如何創建連接到WebSocket端點的WebSocket API

如下圖所示，您可以通過WSO2 API Manager的發布者(publisher)接口來創建WebSocket API，并使用WS服務器將這些事件發布到客戶端。