wso2~event-flow的介紹

我們來詳細(xì)分析和介紹一下 WSO2 平臺(tái)（特別是 WSO2 Stream Processor / WSO2 SI / Siddhi Runner）中 Event Receivers、Event Streams、Execution Plans 和 Event Publishers 這些核心概念。它們是構(gòu)建實(shí)時(shí)流處理和復(fù)雜事件處理（CEP）應(yīng)用的基礎(chǔ)組件。

核心思想：事件驅(qū)動(dòng)架構(gòu) (EDA)

這些組件共同構(gòu)成了一個(gè)事件處理管道（Event Processing Pipeline），其核心流程是：

1. 接收事件： 從外部來源獲取原始事件數(shù)據(jù) (Event Receivers)。
2. 定義結(jié)構(gòu)： 為事件數(shù)據(jù)定義一個(gè)明確的格式和含義 (Event Streams)。
3. 處理邏輯： 對(duì)事件流進(jìn)行過濾、轉(zhuǎn)換、關(guān)聯(lián)、聚合、模式匹配等計(jì)算 (Execution Plans)。
4. 發(fā)布結(jié)果： 將處理結(jié)果（可能是新事件、告警、聚合值等）發(fā)送到下游系統(tǒng) (Event Publishers)。

組件之間的關(guān)系:

核心概念詳解：

1. 事件接收器 (Event Receivers)

   • 功能： 它們是事件處理管道的入口點(diǎn)。負(fù)責(zé)監(jiān)聽、接收、解析和轉(zhuǎn)換來自各種外部源的事件數(shù)據(jù)，并將其注入到 WSO2 平臺(tái)內(nèi)部的 Event Streams 中。
   • 作用：
     • 連接外部世界： 連接到消息隊(duì)列（Kafka, JMS, RabbitMQ）、數(shù)據(jù)庫（通過輪詢）、HTTP/S 端點(diǎn)（REST, WebSocket）、TCP/UDP 套接字、文件、郵件等。
     • 協(xié)議適配： 理解不同來源的協(xié)議和傳輸格式。
     • 數(shù)據(jù)轉(zhuǎn)換： 將接收到的原始數(shù)據(jù)（如 JSON, XML, CSV, Binary）解析并映射到內(nèi)部事件流 (Event Stream) 定義的格式（通常是鍵值對(duì)或?qū)ο螅＿@個(gè)過程通常涉及定義映射規(guī)則（如 @map 注解）。
     • 注入事件： 將轉(zhuǎn)換后的、格式統(tǒng)一的事件對(duì)象發(fā)布到指定的內(nèi)部事件流中，供后續(xù)處理。
   • 關(guān)鍵點(diǎn)：
     • 一個(gè) Receiver 通常對(duì)應(yīng)一個(gè)特定的外部源和一種協(xié)議。
     • 定義了輸入事件如何映射到目標(biāo) Event Stream 的 schema。
     • 是平臺(tái)與外部事件生產(chǎn)者（如傳感器、應(yīng)用程序、日志文件）的橋梁。

2. 事件流 (Event Streams)

   • 功能： 定義了在 WSO2 平臺(tái)內(nèi)部流動(dòng)的事件數(shù)據(jù)的結(jié)構(gòu)和元數(shù)據(jù)。它們是事件在管道中傳輸和處理的載體和契約。
   • 作用：
     • 數(shù)據(jù)模型： 明確規(guī)定一個(gè)事件包含哪些屬性（字段），每個(gè)屬性的數(shù)據(jù)類型是什么（e.g., string, int, long, float, double, bool）。
     • 唯一標(biāo)識(shí)： 每個(gè)流有一個(gè)唯一的名稱（通常由 Stream ID 和 Stream Version 組成，如 FooStream:1.0.0）。
     • 數(shù)據(jù)承載： 實(shí)際的事件數(shù)據(jù)按照這個(gè)定義的結(jié)構(gòu)在系統(tǒng)中流動(dòng)。
     • 組件間連接： Event Receivers 將事件注入特定的 Stream， Execution Plans 從 Streams 讀取事件進(jìn)行處理，并將結(jié)果寫入新的 Streams， Event Publishers 從 Streams 讀取事件進(jìn)行發(fā)布。
   • 關(guān)鍵點(diǎn)：
     • 類似于數(shù)據(jù)庫表定義或消息隊(duì)列中消息的 Schema。
     • 確保處理邏輯 (Execution Plans) 明確知道它正在處理的數(shù)據(jù)是什么樣子。
     • 允許多個(gè)組件（Receivers, Execution Plans, Publishers）通過共享的 Stream 進(jìn)行解耦的通信。
     • 流中的事件通常是不可變的。

3. 執(zhí)行計(jì)劃 (Execution Plans)

   • 功能： 這是事件處理管道的大腦和核心邏輯單元。它定義了如何處理一個(gè)或多個(gè)輸入事件流 (Event Streams) 中的數(shù)據(jù)，執(zhí)行計(jì)算、分析、轉(zhuǎn)換，并產(chǎn)生結(jié)果輸出到一個(gè)或多個(gè)輸出事件流 (Event Streams)。
   • 作用：
     • 處理邏輯容器： 包含一個(gè)或多個(gè)Siddhi 查詢 (SiddhiQL Queries)。SiddhiQL 是一種專門為流處理和 CEP 設(shè)計(jì)的 SQL-like 語言。
     • 復(fù)雜計(jì)算： 執(zhí)行各種操作，包括：
       • 過濾： select * from InputStream[price > 100]
       • 投影： select symbol, price from InputStream
       • 窗口操作： 基于時(shí)間（time window）或數(shù)量（length window）對(duì)事件進(jìn)行分組聚合（avg, sum, min, max, count）或排序。
       • 模式匹配： 檢測(cè)事件序列中的特定模式（every e1=InputStream1 -> e2=InputStream2[e2.value > e1.value]）。
       • 關(guān)聯(lián)/連接： 將不同流的事件基于條件關(guān)聯(lián)起來（join, union）。
       • 機(jī)器學(xué)習(xí)/預(yù)測(cè)： 使用內(nèi)置或擴(kuò)展的函數(shù)進(jìn)行簡(jiǎn)單預(yù)測(cè)或異常檢測(cè)。
       • 外部調(diào)用： 調(diào)用外部服務(wù)或函數(shù)（通過擴(kuò)展）。
     • 狀態(tài)管理： 維護(hù)處理過程中需要的狀態(tài)（如窗口內(nèi)的數(shù)據(jù)、計(jì)數(shù)器、會(huì)話信息）。
     • 輸出生成： 處理的結(jié)果會(huì)以新事件的形式寫入到定義的輸出事件流中。
   • 關(guān)鍵點(diǎn)：
     • 一個(gè) Execution Plan 是處理邏輯的部署單元。它被打包和部署到 WSO2 SP/SI/Siddhi Runner 運(yùn)行時(shí)中。
     • 使用強(qiáng)大的 SiddhiQL 語言定義業(yè)務(wù)邏輯。
     • 消費(fèi)輸入流，生產(chǎn)輸出流。
     • 可以非常復(fù)雜，實(shí)現(xiàn)實(shí)時(shí)分析、告警生成、實(shí)時(shí)儀表盤數(shù)據(jù)準(zhǔn)備等。
     • 替代術(shù)語注意： 在較新版本的 WSO2 SP/Siddhi Runner 中，“Execution Plan” 的概念有時(shí)直接被稱為 “Siddhi App”。一個(gè) Siddhi App 文件 (.siddhi) 包含了 Stream 定義、Receiver/Publisher 配置（通過注解）和 SiddhiQL 查詢，本質(zhì)上就是一個(gè)自包含的 Execution Plan。

4. 事件發(fā)布器 (Event Publishers)

   • 功能： 它們是事件處理管道的出口點(diǎn)。負(fù)責(zé)從內(nèi)部的 Event Streams 中讀取處理完成的事件結(jié)果，將其轉(zhuǎn)換成目標(biāo)系統(tǒng)所需的格式，并發(fā)布/傳輸到各種外部接收系統(tǒng)（Sinks）。
   • 作用：
     • 連接下游系統(tǒng)： 連接到消息隊(duì)列（Kafka, JMS）、數(shù)據(jù)庫（JDBC 插入/更新）、HTTP/S 端點(diǎn)（REST API 調(diào)用）、TCP/UDP 套接字、文件、郵件、日志等。
     • 協(xié)議適配： 使用下游系統(tǒng)理解的協(xié)議進(jìn)行通信。
     • 數(shù)據(jù)轉(zhuǎn)換： 將內(nèi)部事件流格式的事件對(duì)象映射并序列化成外部系統(tǒng)要求的格式（如 JSON, XML, CSV, Binary）。這個(gè)過程通常也涉及定義映射規(guī)則（如 @map 注解）。
     • 傳輸事件： 將轉(zhuǎn)換后的數(shù)據(jù)可靠地（或盡力而為）發(fā)送到目標(biāo)系統(tǒng)。
   • 關(guān)鍵點(diǎn)：
     • 一個(gè) Publisher 通常對(duì)應(yīng)一個(gè)特定的外部目的地和一種協(xié)議。
     • 定義了如何將輸出 Event Stream 的數(shù)據(jù)映射到目標(biāo)格式。
     • 是平臺(tái)與外部事件消費(fèi)者（如數(shù)據(jù)庫、儀表盤、告警系統(tǒng)、其他應(yīng)用程序）的橋梁。

組件間關(guān)系與數(shù)據(jù)流：

想象一個(gè)實(shí)時(shí)的股票價(jià)格監(jiān)控和告警系統(tǒng)：

1. Event Receiver (HTTP): 監(jiān)聽一個(gè) HTTP 端口，接收來自多個(gè)股票數(shù)據(jù)源的 JSON 格式的價(jià)格更新事件。它配置了 @map(type='json')，將 JSON 字段映射到名為 StockTickStream 的內(nèi)部事件流的屬性 (symbol, price, timestamp)。
2. Event Stream (StockTickStream): 定義了 symbol (string), price (float), timestamp (long)。
3. Execution Plan (Siddhi App):
   • 定義輸入流：StockTickStream
   • 編寫 SiddhiQL 查詢：
     • 計(jì)算每個(gè)股票 1 分鐘滾動(dòng)平均價(jià)：... select symbol, avg(price) as avgPrice ... group by symbol ...
     • 檢測(cè)價(jià)格瞬間暴漲（比如 5 秒內(nèi)漲幅超過 10%）：... every (e1=StockTickStream) -> e2=StockTickStream[e1.symbol==e2.symbol and (e2.price - e1.price)/e1.price > 0.1] within 5 sec ...
   • 定義輸出流：
     • AvgPriceStream (schema: symbol, avgPrice)
     • SpikeAlertStream (schema: symbol, startPrice, endPrice, increasePct)
4. Event Publishers:
   • Publisher 1 (Kafka): 訂閱 AvgPriceStream。配置 @map(type='json')，將事件轉(zhuǎn)換為 JSON 并發(fā)布到 Kafka 的 stock-avg-prices 主題，供實(shí)時(shí)儀表盤消費(fèi)。
   • Publisher 2 (Email/JMS): 訂閱 SpikeAlertStream。配置映射，將事件內(nèi)容格式化為告警文本，通過 Email 發(fā)送給交易員或發(fā)送到 JMS 隊(duì)列供告警系統(tǒng)處理。

總結(jié)：

• Event Receivers & Publishers： 處理與外部系統(tǒng)的連接、協(xié)議適配和數(shù)據(jù)格式轉(zhuǎn)換（輸入/輸出適配器）。它們是邊界組件。
• Event Streams： 定義了在系統(tǒng)內(nèi)部流動(dòng)的事件數(shù)據(jù)的標(biāo)準(zhǔn)結(jié)構(gòu)和契約。它們是數(shù)據(jù)流動(dòng)的管道和連接組件的紐帶。
• Execution Plans (Siddhi Apps)： 包含了使用 SiddhiQL 編寫的核心業(yè)務(wù)邏輯和處理規(guī)則。它們消費(fèi)輸入流，進(jìn)行處理（過濾、聚合、模式匹配等），并將結(jié)果寫入輸出流。這是實(shí)現(xiàn)實(shí)時(shí)分析和 CEP 功能的引擎。

理解這些概念的關(guān)鍵：

• 管道化： 數(shù)據(jù)從 Receiver 進(jìn)入 -> 流入 Stream -> 被 Execution Plan 處理 -> 結(jié)果流入另一個(gè) Stream -> 由 Publisher 發(fā)送出去。
• 解耦： Streams 使得 Receivers、Execution Plans 和 Publishers 可以獨(dú)立定義、修改和擴(kuò)展，只要它們遵守 Stream 的契約（Schema）。
• 聲明式處理： Execution Plans 使用 SiddhiQL 這種高級(jí)的聲明式語言來描述“做什么”（如找出平均值或模式），而不是詳細(xì)指定“怎么做”（如遍歷循環(huán)），由運(yùn)行時(shí)引擎優(yōu)化執(zhí)行。

掌握這四個(gè)核心概念是有效設(shè)計(jì)、開發(fā)和部署基于 WSO2 Stream Processor / Siddhi 的實(shí)時(shí)流處理應(yīng)用程序的基礎(chǔ)。