快捷導航

如何使用Reactor完成類似Flink的操作

更新時間：2021年03月01日 09:27:07 作者：木小豐

這篇文章主要介紹了如何使用Reactor完成類似Flink的操作，幫助大家更好的理解和學習使用Java，感興趣的朋友可以了解下

一、背景

Flink在處理流式任務(wù)的時候有很大的優(yōu)勢，其中windows等操作符可以很方便的完成聚合任務(wù)，但是Flink是一套獨立的服務(wù)，業(yè)務(wù)流程中如果想使用需要將數(shù)據(jù)發(fā)到kafka，用Flink處理完再發(fā)到kafka，然后再做業(yè)務(wù)處理，流程很繁瑣。

比如在業(yè)務(wù)代碼中想要實現(xiàn)類似Flink的window按時間批量聚合功能，如果純手動寫代碼比較繁瑣，使用Flink又太重，這種場景下使用響應(yīng)式編程RxJava、Reactor等的window、buffer操作符可以很方便的實現(xiàn)。

響應(yīng)式編程框架也早已有了背壓以及豐富的操作符支持，能不能用響應(yīng)式編程框架處理類似Flink的操作呢，答案是肯定的。

本文使用Reactor來實現(xiàn)Flink的window功能來舉例，其他操作符理論上相同。文中涉及的代碼：github

二、實現(xiàn)過程

Flink對流式處理做的很好的封裝，使用Flink的時候幾乎不用關(guān)心線程池、積壓、數(shù)據(jù)丟失等問題，但是使用Reactor實現(xiàn)類似的功能就必須對Reactor運行原理比較了解，并且經(jīng)過不同場景下測試，否則很容易出問題。

下面列舉出實現(xiàn)過程中的核心點：

1、創(chuàng)建Flux和發(fā)送數(shù)據(jù)分離

入門Reactor的時候給的示例都是創(chuàng)建Flux的時候同時就把數(shù)據(jù)賦值了，比如：Flux.just、Flux.range等，從3.4.0版本后先創(chuàng)建Flux，再發(fā)送數(shù)據(jù)可使用Sinks完成。有兩個比較容易混淆的方法：

Sinks.many().multicast() 如果沒有訂閱者，那么接收的消息直接丟棄
Sinks.many().unicast() 如果沒有訂閱者，那么保存接收的消息直到第一個訂閱者訂閱
Sinks.many().replay() 不管有多少訂閱者，都保存所有消息

在此示例場景中，選擇的是Sinks.many().unicast()

官方文檔：https://projectreactor.io/docs/core/release/reference/#processors

2、背壓支持

上面方法的對象背壓策略支持兩種：BackpressureBuffer、BackpressureError，在此場景肯定是選擇BackpressureBuffer，需要指定緩存隊列，初始化方法如下：Queues.get(queueSize).get()

數(shù)據(jù)提交有兩個方法：

emitNext 指定提交失敗策略同步提交
tryEmitNext 異步提交，返回提交成功、失敗狀態(tài)

在此場景我們不希望丟數(shù)據(jù)，可自定義失敗策略，提交失敗無限重試，當然也可以調(diào)用異步方法自己重試。

Sinks.EmitFailureHandler ALWAYS_RETRY_HANDLER = (signalType, emitResult) -> emitResult.isFailure();

在此之后就就可以調(diào)用Sinks.asFlux開心的使用各種操作符了。

3、窗口函數(shù)

Reactor支持兩類窗口聚合函數(shù)：

window類：返回Mono(Flux)
buffer類：返回List

在此場景中，使用buffer即可滿足需求，bufferTimeout(int maxSize, Duration maxTime)支持最大個數(shù)，最大等待時間操作，F(xiàn)link中的keys操作可以用groupBy、collectMap來實現(xiàn)。

4、消費者處理

Reactor經(jīng)過buffer后是一個一個的發(fā)送數(shù)據(jù)，如果使用publishOn或subscribeOn處理的話，只等待下游的subscribe處理完成才會重新request新的數(shù)據(jù)，buffer操作符才會重新發(fā)送數(shù)據(jù)。如果此時subscribe消費者耗時較長，數(shù)據(jù)流會在buffer流程阻塞，顯然并不是我們想要的。

理想的操作是消費者在一個線程池里操作，可多線程并行處理，如果線程池滿，再阻塞buffer操作符。解決方案是自定義一個線程池，并且當然線程池如果任務(wù)滿submit支持阻塞，可以用自定義RejectedExecutionHandler來實現(xiàn)：

 RejectedExecutionHandler executionHandler = (r, executor) -> {
   try {
     executor.getQueue().put(r);
   } catch (InterruptedException e) {
     Thread.currentThread().interrupt();
     throw new RejectedExecutionException("Producer thread interrupted", e);
   }
 };
 
 new ThreadPoolExecutor(poolSize, poolSize,
     0L, TimeUnit.MILLISECONDS,
     new SynchronousQueue<>(),
     executionHandler);

三、總結(jié)

1、總結(jié)一下整體的執(zhí)行流程

提交任務(wù)：提交數(shù)據(jù)支持同步異步兩種方式，支持多線程提交，正常情況下響應(yīng)很快，同步的方法如果隊列滿則阻塞。
豐富的操作符處理流式數(shù)據(jù)。
buffer操作符產(chǎn)生的數(shù)據(jù)多線程處理：同步提交到單獨的消費者線程池，線程池任務(wù)滿則阻塞。
消費者線程池：支持阻塞提交，保證不丟消息，同時隊列長度設(shè)置成0，因為前面已經(jīng)有隊列了。
背壓：消費者線程池阻塞后，會背壓到buffer操作符，并背壓到緩沖隊列，緩存隊列滿背壓到數(shù)據(jù)提交者。

2、和Flink的對比

實現(xiàn)的Flink的功能：

不輸Flink的豐富操作符
支持背壓，不丟數(shù)據(jù)

優(yōu)勢：

輕量級，可直接在業(yè)務(wù)代碼中使用

劣勢：

內(nèi)部執(zhí)行流程復雜，容易踩坑，不如Flink傻瓜化
沒有watermark功能，也就意味著只支持無序數(shù)據(jù)處理
沒有savepoint功能，雖然我們用背壓解決了部分問題，但是宕機后開始會丟失緩存隊列和消費者線程池里的數(shù)據(jù)，補救措施是添加Java Hook功能
只支持單機，意味著你的緩存隊列不能設(shè)置無限大，要考慮線程池的大小，且沒有flink globalWindow等功能
需考慮對上游數(shù)據(jù)源的影響，F(xiàn)link的上游一般是mq，數(shù)據(jù)量大時可自動堆積，如果本文的方案上游是http、rpc調(diào)用，產(chǎn)生的阻塞影響就不能忽略。補償方案是每次提交數(shù)據(jù)都使用異步方法，如果失敗則提交到mq中緩沖并消費該mq無限重試。