这篇文章给大家介绍使用Reactor怎么实现一个Flink操作功能,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。
Flink对流式处理做的很好的封装,使用Flink的时候几乎不用关心线程池、积压、数据丢失等问题,但是使用Reactor实现类似的功能就必须对Reactor运行原理比较了解,并且经过不同场景下测试,否则很容易出问题。
下面列举出实现过程中的核心点:
入门Reactor的时候给的示例都是创建Flux的时候同时就把数据赋值了,比如:Flux.just、Flux.range等,从3.4.0版本后先创建Flux,再发送数据可使用Sinks完成。有两个比较容易混淆的方法:
Sinks.many().multicast() 如果没有订阅者,那么接收的消息直接丢弃
Sinks.many().unicast() 如果没有订阅者,那么保存接收的消息直到第一个订阅者订阅
Sinks.many().replay() 不管有多少订阅者,都保存所有消息
在此示例场景中,选择的是Sinks.many().unicast()
官方文档:/tupian/20230522/>
emitNext 指定提交失败策略同步提交
tryEmitNext 异步提交,返回提交成功、失败状态
在此场景我们不希望丢数据,可自定义失败策略,提交失败无限重试,当然也可以调用异步方法自己重试。
Sinks.EmitFailureHandler ALWAYS_RETRY_HANDLER = (signalType, emitResult) -> emitResult.isFailure();
在此之后就就可以调用Sinks.asFlux开心的使用各种操作符了。
在此之后就就可以调用Sinks.asFlux开心的使用各种操作符了。
Reactor支持两类窗口聚合函数:
window类:返回Mono(Flux)
buffer类:返回List
在此场景中,使用buffer即可满足需求,bufferTimeout(int maxSize, Duration maxTime)支持较大个数,较大等待时间操作,Flink中的keys操作可以用groupBy、collectMap来实现。
Reactor经过buffer后是一个一个的发送数据,如果使用publishOn或subscribeOn处理的话,只等待下游的subscribe处理完成才会重新request新的数据,buffer操作符才会重新发送数据。如果此时subscribe消费者耗时较长,数据流会在buffer流程阻塞,显然并不是我们想要的。
理想的操作是消费者在一个线程池里操作,可多线程并行处理,如果线程池满,再阻塞buffer操作符。解决方案是自定义一个线程池,并且当然线程池如果任务满submit支持阻塞,可以用自定义RejectedExecutionHandler来实现:
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);
提交任务:提交数据支持同步异步两种方式,支持多线程提交,正常情况下响应很快,同步的方法如果队列满则阻塞。
丰富的操作符处理流式数据。
buffer操作符产生的数据多线程处理:同步提交到单独的消费者线程池,线程池任务满则阻塞。
消费者线程池:支持阻塞提交,保证不丢消息,同时队列长度设置成0,因为前面已经有队列了。
背压:消费者线程池阻塞后,会背压到buffer操作符,并背压到缓冲队列,缓存队列满背压到数据提交者。
实现的Flink的功能:
不输Flink的丰富操作符
支持背压,不丢数据
优势:
轻量级,可直接在业务代码中使用
劣势:
内部执行流程复杂,容易踩坑,不如Flink傻瓜化
没有watermark功能,也就意味着只支持无序数据处理
没有savepoint功能,虽然我们用背压解决了部分问题,但是宕机后开始会丢失缓存队列和消费者线程池里的数据,补救措施是添加Java Hook功能
只支持单机,意味着你的缓存队列不能设置无限大,要考虑线程池的大小,且没有flink globalWindow等功能
需考虑对上游数据源的影响,Flink的上游一般是mq,数据量大时可自动堆积,如果本文的方案上游是http、rpc调用,产生的阻塞影响就不能忽略。补偿方案是每次提交数据都使用异步方法,如果失败则提交到mq中缓冲并消费该mq无限重试。
关于使用Reactor怎么实现一个Flink操作功能就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。
分享文章:使用Reactor怎么实现一个Flink操作功能-创新互联
URL分享:https://www.cdcxhl.com/article4/djddoe.html
成都网站建设公司_创新互联,为您提供企业网站制作、关键词优化、做网站、响应式网站、营销型网站建设、商城网站
声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联