RocketMQ-Streams 首個版本發布，輕量級計算的新選擇

RocketMQ-Streams 聚焦「大資料量->高過濾->輕視窗計算」場景，核心打造輕資源，高性能優勢，在資源敏感場景有很大優勢，最低 1Core，1G 可部署，通過大量過濾優化，性能比其他大資料提升 2-5 倍性能，廣泛應用于安全，風控，邊緣計算，訊息佇列流計算，
RocketMQ-Streams 兼容 Flink 的 SQL，udf/udtf/udaf，將來我們會和 Flink 生態做深度融合，即可以獨立運行，也可發布成 Flink 任務，跑在 Flink 集群，對于有 Flink 集群的場景，即能享有輕資源優勢，可以做到統一部署和運維，

RocketMQ-Streams 特點及應用場景

RocketMQ-Streams 應用場景

? 計算場景： 適合大資料量->高過濾->輕視窗計算的場景，不同于主流計算引擎，需要先部署集群，寫任務，發布，調優，運行這么復雜的程序，RocketMQ-Streams 本身就是一個 lib 包，基于 SDK 寫完流任務，可以直接運行，支持大資料開發需要的計算特性：Exactly-ONCE，靈活視窗（滾動、滑動、會話），雙流Join，高吞吐、低延遲、高性能，最低 1Core，1G 可以運行，

? SQL引擎 ： RocketMQ-Streams 可視作一個 SQL 引擎，兼容 Flink SQL 語法，支持 Flink udf/udtf/udaf 的擴展，支持 SQL 熱升級，寫完 SQL，通過 SDK 提交 SQL，就可以完成 SQL 的熱發布，

? ETL引擎： RocketMQ-Streams 還可視作 ETL 引擎，在很多大資料場景，需要完成資料從一個源經過 ETl，匯聚到統一存盤，里面內置了 grok，正則決議等函式，可以結合 SQL 一塊完成資料 ETL ，

? 開發 SDK，它也是一個資料開發 SDK 包，里面的大多陣列件都可以單獨使用，如 Source/sink，它屏蔽了資料源，資料存盤細節，提供統一編程介面，一套代碼，切換輸入輸出，不需要改變代碼，

RocketMQ-Streams 設計思路

設計目標
? 依賴少，部署簡單，1Core，1G 單實體可部署，可隨意擴展規模，
? 實作需要的大資料特性：Exactly-ONCE，靈活視窗（滾動、滑動、會話），雙流 Join，高吞吐、低延遲、高性能，
? 實作成本可控，實作低資源，高性能，
? 兼容 Flink SQL，UDF/UDTF，讓非技術人員更易上手，

設計思路
? 采用 shared-nothing 的分布式架構設計，依賴訊息佇列做負載均衡和容錯機制，單實體可啟動，增加實體實作能力擴展，并發能力取決于分片數，
? 利用訊息佇列的分片做 shuffle，利用訊息佇列負載均衡實作容錯，
? 利用存盤實作狀態備份，實作 Exactly-ONCE 的語意，用結構化遠程存盤實作快速啟動，不必等本地存盤恢復，

RocketMQ-Streams 特點和創新

RocketMQ-Streams SDK 詳解

Hello World
按照慣例，我們先從一個例子來了解 RocketMQ-Streams

? namespace：相同 namespace 的任務可以跑在一個行程里，可以共享配置
? pipelineName：job name
? DataStreamSource：創建 source 節點
? map：用戶函式，可以通過實作 MapFunction 擴展功能
? toPrint：結果列印出來
? start：啟動任務
? 運行上面代碼就會啟動一個實體，如果想多實體并發，可以啟動多個實體，每個實體消費部分 RocketMQ 的資料，
? 運行結果：把原始訊息拼接上“—”，并列印出來

RocketMQ-Streams SDK

? StreamBuilder 做為起點，通過設定 namespace，jobName 創建一個 DataStreamSource ，
? DataStreamSource 通過 from 方法，設定 source，創建 DataStream 物件，
? DataStream 提供多種操作，會產生不同的流：
? to 操作產生 DataStreamAction
? window 操作產生 WindowStream 配置 window 引數
? join 操作產生 JoinStream 配置 join 條件
? Split 操作產生 SplitStream 配置 split 條件
? 其他操作產生 DataStream
? DataStreamAction 啟動整個任務，也可以配置任務的各種策略引數，支持異步啟動和同步啟動，

RocketMQ-Streams 算子

RocketMQ-Streams 算子

SQL 有兩種部署模式，1 是直接運行 client 啟動 SQL，見第一個紅框；2 是搭建 server 集群，通過 client 提交 SQL 實作熱部署，見第二個紅框，

RocketMQ-Streams SQL 擴展，支持多種擴展方式：
? 通過 FlinkUDF,UDTF,UDAF 擴展 SQL 能力，在 SQL 中通過 create function 引入，有個限制條件，即 UDF 在 open 時未用到 Flink FunctionContext 的內容，
? 通過內置函式擴展 SQL 的函式，語法同 Flink 語法，函式名是內置函式的名稱，類名是固定的，如下圖，引入了一個 now 的函式，輸出當前時間，系統內置了 200 多個函式，可按需引入，

? 通過擴展函式實作，實作一個函式很簡單，只需要在 class 上標注 Function，在需要發布成函式的方法上標注 FunctionMethod，并設定需要發布的函式名即可，如果需要系統資訊，前面兩個函式可以是 IMessage 和 Abstract，如果不需要，直接寫引數即可，引數無格式要求，如下圖，創建了一個 now 的函式，兩種寫法都可以，可以通過 currentTime=now()來呼叫，會在 Message 中增加一個 key=currentTime，value=https://www.cnblogs.com/alisystemsoftware/archive/2022/02/24/當前時間的變數，

? 把現有 java 代碼發布成函式，通過策略配置，把 java 代碼的類名，方法名，期望用到的函式名，配置進去，把 java 的 jar 包 copy 到 jar 包目錄即可，下圖是幾種擴展的應用實體，

RocketMQ-Streams 架構及原理實作

整體架構

Source 實作

? Source 要求實作最少消費一次的語意，系統通過 checkpoint 系統訊息實作，在提交 offset 前發送 checkpoint 訊息，通知所有算子重繪記憶體，
? Source 支持分片的自動負載均衡和容錯，
? 資料源在分片移除時，發送移除系統訊息，讓算子完成分片清理作業，
? 當有新分片時， 發送新增分片訊息，讓算子完成分片的初始化，
? 資料源通過 start 方法，啟動 consuemr 獲取訊息，
? 原始訊息經過編碼，附加頭部資訊包裝成 Message 投遞給后續算子，

Sink 實作

? Sink 是實時性和吞吐的一個結合，
? 實作一個 Sink 只要繼承 AbstractSink 類實作 batchInsert 方法即可，batchInsert 的含義是一批資料寫入存盤，需要子類呼叫存盤介面實作，盡量應用存盤的批處理介面，提高吞吐，
? 常規的使用方式是寫 Message->cache->flush->存盤的方式，系統會嚴格保證，每次批次寫入存盤的量不超過 batchsize 的量，如果超了，會拆分成多批寫入，

? Sink 有一個 cache，資料默認寫 cache，批次寫入存盤，提高吞吐量，（一個分片一個 cache），
? 可以開啟自動重繪，每個分片會有一個執行緒，定時重繪 cache 資料到存盤，提高實時性，實作類：DataSourceAutoFlushTask ，
? 也可以通過呼叫 flush 方法重繪 cache 到存盤，
? Sink 的 cache 會有記憶體保護，當 cache 的訊息條數>batchSize，會強制重繪，釋放記憶體，

RocketMQ-Streams Exactly-ONCE

? Source 確保在 commit offset 時，會發送 checkpoint 系統訊息，收到訊息的組件會完成存盤操作，訊息至少消費一次，
? 每條訊息會有訊息頭部，里面封裝了 QueueId 和 offset ，
? 組件在存盤資料時，會把 QueueId 和處理的最大 offset 存盤下來，當有訊息重復時，根據 maxoffset 去重，
? 記憶體保護，一個 checkpoint 周期可能有多次 flush（條數觸發），保障記憶體占用可控，

RocketMQ-Streams Window

? 支持滾動，滑動和會話視窗，支持事件時間和自然時間（訊息進入算子的時間），
? 支持高性能模式和高可靠模式，高性能模式不依賴遠程存盤，但在分片切換時的視窗資料會有丟失，
? 快速啟動，無需等本地存盤恢復，在發生錯誤或分片切換時，異步從遠程存盤恢復資料，同時直接訪問遠程存盤計算，
? 利用訊息佇列負載均衡，實作擴容縮容，每個 Queue 是一個分組，一個分組同一刻只被一臺機器消費，
? 正常計算依賴本地存盤，具備 Flink 相似的計算性能，

支持三種觸發模式，可以均衡 watermark 延遲和實時性要求

RocketMQ-Streams 在云安全的應用

在安全應用的背景

? 公共云轉戰專有云，在入侵檢測計算方面遇到了資源問題，大資料集群默認不輸出，輸出最低 6 臺高配機器，用戶很難接受因為買云盾增配一套大資料集群，
? 專有云用戶升級，運維困難，無法快速升級能力和修復 bug，

流計算在安全的應用

? 基于安全特點（大資料->高過濾->輕視窗計算）打造輕量級計算引擎：經過分析所有的規則都會做前置過濾，然后才會做較重的統計，視窗，join 操作，且過濾率比較高，基于此特點，可以用更輕的方案實作統計，join 操作，

? 通過 RocketMQ-Streams，覆寫 100%專有云規則（正則，join，統計），
? 輕資源，記憶體是公共云引擎的 1/70，CPU 是 1/6，通過指紋過濾優化，性能提升 5 倍以上，且資源不隨規則線性增加，新增規則無資源壓力，復用以前的正則引擎資源，可支持 95%以上局點，不需要增加額外物理資源，
? 通過高壓縮維表，支持千萬情報，1000 W 資料只需要 330 M 記憶體，
? 通過 C/S 部署模式，SQL 和引擎可熱發布，尤其護網場景，可快速上線規則，

RocketMQ-Streams 未來規劃

新版本下載地址：https://github.com/apache/rocketmq-streams/releases/tag/rocketmq-streams-1.0.0-preview
————————————————
著作權宣告：本文為CSDN博主「阿里巴巴云原生」的原創文章，遵循CC 4.0 BY-SA著作權協議，轉載請附上原文出處鏈接及本宣告，
原文鏈接：https://blog.csdn.net/alisystemsoftware/article/details/123096937
發布云原生技術最新資訊、匯集云原生技術最全內容，定期舉辦云原生活動、直播，阿里產品及用戶最佳實踐發布，與你并肩探索云原生技術點滴，分享你需要的云原生內容，

關注【阿里巴巴云原生】公眾號，獲取更多云原生實時資訊！

標籤：其他

上一篇：「螣龍安科｜潮汐.安全感知平臺」：為大型企業提供平臺式基礎設施，助力企業標準化輸出安全能力

下一篇：Unity產生黑夜與迷霧的效果