作者 |倪澤

RocketMQ-Streams 是一款輕量級流處理引擎，應(yīng)用以SDK 的形式嵌入并啟動，即可進行流處理計算，不依賴于其他組件，最低1核1G可部署，在資源敏感場景具有很大優(yōu)勢。同時它支持 UTF/UTAF/UTDF 多種計算類型。目前已經(jīng)廣泛運用于安全，風(fēng)控，邊緣計算等場景。

本期將帶領(lǐng)大家從源碼的角度，解析RocketMQ-Streams的構(gòu)建，數(shù)據(jù)流轉(zhuǎn)過程。也會討論RocketMQ-Streams是如何實現(xiàn)故障恢復(fù)和擴縮容的。

一、使用示例

代碼示例：

public class RocketMQWindowExample {
    public static void main(String[] args) {
        DataStreamSource source = StreamBuilder.dataStream("namespace", "pipeline");
        source.fromRocketmq(
                "topicName",
                "groupName",
                false,
                "namesrvAddr")
                .map(message -> JSONObject.parseObject((String) message))
                .window(TumblingWindow.of(Time.seconds(10)))
                .groupBy("groupByKey")
                .sum("字段名", "輸出別名")
                .count("total")
                .waterMark(5)
                .setLocalStorageOnly(true)
                .toDataSteam()
                .toPrint(1)
                .start();

    }

}

pom文件依賴：

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-streams-clients</artifactId>
  <version>1.0.1-preview</version>
</dependency>

上述代碼是一個簡單的使用例子，它主要的功能是從RocketMQ中指定topic讀取數(shù)據(jù)，經(jīng)過轉(zhuǎn)化成JSON格式，以groupByKey字段值分組、10秒一個窗口，對OutFlow字段值進行累加，結(jié)果輸出到total字段，并打印到控制臺上。上述計算中還允許輸入亂序5秒，即窗口時間到達后不會馬上觸發(fā)，而是會等待5s，如果這個段時間內(nèi)，有窗口數(shù)據(jù)到達依然有效。上述setLocalStorageOnly為true表示不對狀態(tài)進行遠程存儲，僅使用RocksDB做本地存儲。目前1.0.1的RocketMQ-Streams版本依然使用Mysql作為遠程狀態(tài)存儲，下一版本將使用RocketMQ作為遠程狀態(tài)存儲。

二、RocketMQ總體架構(gòu)圖

RocketMQ-Streams 作為輕量流處理引擎，本質(zhì)上是作為RocketMQ 的客戶端消費數(shù)據(jù)，一個流處理實例可以處理多個隊列，而一個隊列只能被一個實例消費。若干RocketMQ-Streams 實例組成消費者組共同消費數(shù)據(jù)，通過擴容實例達到增加處理能力的消費，減少實例則會發(fā)生rebalance，消費的隊列自動重平衡到其他消費實例上。從上述圖中，我們還可以看出計算實例間不需要直接交換任何數(shù)據(jù)，可各自獨立完成所有計算處理。這種架構(gòu)簡化了RocketMQ-Streams 本身的設(shè)計，同時也可非常方便的進行實例擴縮容。

處理拓?fù)?/h4>

處理器拓?fù)錇閼?yīng)用定義了流處理過程的計算邏輯，它由一系列的處理器節(jié)點和數(shù)據(jù)流向組成。例如，在開頭的代碼示例中，整個處理拓?fù)溆蓅ource、map、groupBy、sum、count、print等處理節(jié)點組成。有兩種特殊的處理節(jié)點：

• source節(jié)點

他沒有任何上游節(jié)點，從外部讀入數(shù)據(jù)到RocketMQ-Streams，并交由下游處理。

• sink節(jié)點

他沒有任何下游節(jié)點，他將處理后的數(shù)據(jù)寫出到外部。

處理拓?fù)鋬H僅是流處理代碼的邏輯抽象，在流計算啟動時將會被實例化。為了設(shè)計簡單，目前一個流處理實例中僅有一張計算拓?fù)洹?/p>

在所有流處理算子之中，有兩種特別的算子，一種是涉及數(shù)據(jù)分組的算子groupBy，另一種是有狀態(tài)計算例如count等。這兩種算子會影響整個計算拓?fù)涞臉?gòu)建，下面將具體分析RocketMQ-Streams是如何處理他們的。

groupBy

分組算子groupBy特殊是因為經(jīng)過groupBy操作，后續(xù)算子期望對相同key的數(shù)據(jù)進行操作，例如經(jīng)過groupBy("年級")之后再進行sum就是對按照年級分組求和，這就要求需要將具有相同“年級”的數(shù)據(jù)重新路由到一個流計算實例上處理，如果不這樣做，每個實例上得出的結(jié)果都將是不完整的，整體輸出結(jié)果也將是錯誤的。

RocketMQ-Streams 采用 shuffle topic 這種方式來處理。具體說來，計算實例將groupBy數(shù)據(jù)重新發(fā)回RocketMQ的一個topic，并且在發(fā)回過程中按照key的hash值來選擇目標(biāo)隊列，再從這個topic讀取數(shù)據(jù)進行后續(xù)流處理。按照key hash后相同的key一定在一個隊列里面，而一個隊列只會被一個流處理實例消費，這樣就達到相同key被路由到一個實例上處理的效果。

有狀態(tài)算子

有狀態(tài)算子與無狀態(tài)算子相對。如果計算結(jié)果只與當(dāng)前輸入有關(guān)，和上一次輸入無關(guān)就是無狀態(tài)算子，例如filter、map、foreach結(jié)果只與當(dāng)前輸入有關(guān)系。還有一種算子的輸出結(jié)果不僅與當(dāng)前算子有關(guān)系還與上一次輸入有關(guān)，例如sum，需要對一段時間內(nèi)輸入進行求和，他就是有狀態(tài)算子。

RocketMQ-Streams利用RocksDB作為本地存儲，Mysql作為遠程存儲來保存狀態(tài)數(shù)據(jù)。他具體做法是：

當(dāng)發(fā)現(xiàn)消息來自新的隊列時，檢查是否需要加載狀態(tài)，如果需要異步加載狀態(tài)到RocksDB。

數(shù)據(jù)到達有狀態(tài)算子時，如果加載完成使用RocksDB中狀態(tài)進行計算，如果沒有，使用Mysql中狀態(tài)計算。

計算完成后，將狀態(tài)數(shù)據(jù)保存到RocksDB和Mysql中。

窗口觸發(fā)后，從RocksDB中查詢出狀態(tài)數(shù)據(jù)，并將結(jié)果向下游算子傳遞。

整體數(shù)據(jù)流向圖如下：

三、擴縮容與故障恢復(fù)

擴縮容和故障恢復(fù)是一個硬幣的兩面，即同一個事物的兩種表達，計算集群如果能正確擴縮容就等于具備故障恢復(fù)的能力，反之亦然。通過前面介紹我們知道，RocketMQ-Streams具有非常良好的擴縮容性能，擴容時只需要新部署一個流計算實例即可，縮容時停止計算實例即可。對于無狀態(tài)的計算來說比較簡單，擴容后，數(shù)據(jù)計算不需要之前的狀態(tài)。有狀態(tài)計算的擴縮容涉及到狀態(tài)的遷移。有狀態(tài)的擴縮容可由下圖表示：

當(dāng)計算實例從3個縮容到2個，借助于RocketMQ的rebalance，MQ會在計算實例之間重新分配。

Instance1上消費的MQ2和MQ3被分配到Instance2和Instance3上，這兩個MQ的狀態(tài)數(shù)據(jù)也需要遷移到Instance2和Instance3上，這也暗示，狀態(tài)數(shù)據(jù)是根據(jù)源數(shù)據(jù)分片保存的;擴容則是剛好相反的過程。

具體實現(xiàn)上，RocketMQ-Streams采用系統(tǒng)消息來觸發(fā)狀態(tài)的加載和持久化。

系統(tǒng)消息類別：

//新增消費隊列
NewSplitMessage

//不在消費某個隊列
RemoveSplitMessage

//客戶端持久化消費位點到MQ
CheckPointMessage

當(dāng)發(fā)現(xiàn)消息來自一個新的RocketMQ隊列(MessageQueue)，RocketMQ-Streams之前沒有處理過來自該隊列的消息，會先于數(shù)據(jù)前發(fā)送NewSplitMessage消息，通過處理拓?fù)湎掠嗡阕觽鬟f，當(dāng)有狀態(tài)算子收到該消息時會將新增隊列對應(yīng)的狀態(tài)加載到本地內(nèi)存RocksDB中，當(dāng)數(shù)據(jù)真正到達時，就根據(jù)這個狀態(tài)繼續(xù)計算。

當(dāng)因為計算實例增加或者RocketMQ集群變動，rebalance后，計算實例不再消費某個隊列(MessageQueue)時，會發(fā)出RemoveSplitMessage消息，有狀態(tài)算子刪除本地RocksDB中的狀態(tài)。

CheckPointMessage是一種特別的系統(tǒng)消息，他的作用與實現(xiàn)exactly-once有關(guān)。我們在擴縮容過程中需要做到exactly-once，才能保證擴縮容或故障恢復(fù)對計算結(jié)果沒有影響。RocketMQ-streams向broker提交消費offset前會產(chǎn)生CheckPointMessage消息，向下游拓?fù)鋫鬟f，他將保證即將提交消費位點的所有消息都已經(jīng)被sink處理掉。

開源地址：

RocketMQ-Streams 倉庫地址：

https://github.com/apache/rocketmq-streams

RocketMQ 倉庫地址：

https://github.com/apache/rocketmq

作者：倪澤，RocketMQ 資深貢獻者， RocketMQ-Streams 維護者之一，阿里云技術(shù)專家。