這篇文章給大家介紹使用Reactor怎么實(shí)現(xiàn)一個(gè)Flink操作功能，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

成都創(chuàng)新互聯(lián)公司提供做網(wǎng)站、成都做網(wǎng)站、網(wǎng)頁設(shè)計(jì)，高端網(wǎng)站設(shè)計(jì)，一元廣告等致力于企業(yè)網(wǎng)站建設(shè)與公司網(wǎng)站制作，十載的網(wǎng)站開發(fā)和建站經(jīng)驗(yàn),助力企業(yè)信息化建設(shè)，成功案例突破上千余家,是您實(shí)現(xiàn)網(wǎng)站建設(shè)的好選擇.

實(shí)現(xiàn)過程

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

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

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

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

• Sinks.many().multicast() 如果沒有訂閱者，那么接收的消息直接丟棄

• Sinks.many().unicast() 如果沒有訂閱者，那么保存接收的消息直到第一個(gè)訂閱者訂閱

• Sinks.many().replay() 不管有多少訂閱者，都保存所有消息

  
  

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

官方文檔：/tupian/20230522/>

emitNext 指定提交失敗策略同步提交

tryEmitNext 異步提交，返回提交成功、失敗狀態(tài)

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

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

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

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

3、窗口函數(shù)

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

• window類：返回Mono(Flux)

• buffer類：返回List

  
  

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

4、消費(fèi)者處理

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

理想的操作是消費(fèi)者在一個(gè)線程池里操作，可多線程并行處理，如果線程池滿，再阻塞buffer操作符。解決方案是自定義一個(gè)線程池，并且當(dāng)然線程池如果任務(wù)滿submit支持阻塞，可以用自定義RejectedExecutionHandler來實(shí)現(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)很快，同步的方法如果隊(duì)列滿則阻塞。
豐富的操作符處理流式數(shù)據(jù)。
buffer操作符產(chǎn)生的數(shù)據(jù)多線程處理：同步提交到單獨(dú)的消費(fèi)者線程池，線程池任務(wù)滿則阻塞。
消費(fèi)者線程池：支持阻塞提交，保證不丟消息，同時(shí)隊(duì)列長(zhǎng)度設(shè)置成0，因?yàn)榍懊嬉呀?jīng)有隊(duì)列了。
背壓：消費(fèi)者線程池阻塞后，會(huì)背壓到buffer操作符，并背壓到緩沖隊(duì)列，緩存隊(duì)列滿背壓到數(shù)據(jù)提交者。

2、和Flink的對(duì)比

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

• 不輸Flink的豐富操作符

• 支持背壓，不丟數(shù)據(jù)

  
  

優(yōu)勢(shì)：

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

劣勢(shì)：

• 內(nèi)部執(zhí)行流程復(fù)雜，容易踩坑，不如Flink傻瓜化

• 沒有watermark功能，也就意味著只支持無序數(shù)據(jù)處理

• 沒有savepoint功能，雖然我們用背壓解決了部分問題，但是宕機(jī)后開始會(huì)丟失緩存隊(duì)列和消費(fèi)者線程池里的數(shù)據(jù)，補(bǔ)救措施是添加Java Hook功能

• 只支持單機(jī)，意味著你的緩存隊(duì)列不能設(shè)置無限大，要考慮線程池的大小，且沒有flink globalWindow等功能

• 需考慮對(duì)上游數(shù)據(jù)源的影響，F(xiàn)link的上游一般是mq，數(shù)據(jù)量大時(shí)可自動(dòng)堆積，如果本文的方案上游是http、rpc調(diào)用，產(chǎn)生的阻塞影響就不能忽略。補(bǔ)償方案是每次提交數(shù)據(jù)都使用異步方法，如果失敗則提交到mq中緩沖并消費(fèi)該mq無限重試。

關(guān)于使用Reactor怎么實(shí)現(xiàn)一個(gè)Flink操作功能就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到。

分享文章：使用Reactor怎么實(shí)現(xiàn)一個(gè)Flink操作功能-創(chuàng)新互聯(lián)
URL分享：http://www.chinadenli.net/article4/djddoe.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網(wǎng)站制作、關(guān)鍵詞優(yōu)化、做網(wǎng)站、響應(yīng)式網(wǎng)站、營(yíng)銷型網(wǎng)站建設(shè)、商城網(wǎng)站

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)