使用反应堆怎么实现一个Flink操作功能 - 行业资讯 - 肥雀云

　　介绍

这篇文章给大家介绍使用反应堆怎么实现一个Flink操作功能,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。

Flink对流式处理做的很好的封装,使用Flink的时候几乎不用关心线程池,积压,数据丢失等问题,但是使反应堆用实现类似的功能就必须对反应堆运行原理比较了解,并且经过不同场景下测试,否则很容易出问题。

下面列举出实现过程中的核心点:

1,创建通量和发送数据分离

入门反应堆的时候给的示例都是创建通量的时候同时就把数据赋值了,比如:Flux.just, Flux.range等,从3.4.0版本后先创建通量,再发送数据可使用汇完成。有两个比较容易混淆的方法:

<李>

Sinks.many .multicast()()如果没有订阅者,那么接收的消息直接丢弃

<李>

Sinks.many .unicast()()如果没有订阅者,那么保存接收的消息直到第一个订阅者订阅

<李>

Sinks.many .replay()()不管有多少订阅者,都保存所有消息

在此示例场景中,选择的是Sinks.many () .unicast ()

官方文档:https://projectreactor.io/docs/core/release/reference/处理器

2,背压支持

上面方法的对象背压策略支持两种:BackpressureBuffer, BackpressureError,在此场景肯定是选择BackpressureBuffer,需要指定缓存队列,初始化方法如下:Queues.get (queueSize) . get ()

数据提交有两个方法:

<李>

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

<李>

tryEmitNext异步提交,返回提交成功,失败状态

在此场景我们不希望丢数据,可自定义失败策略,提交失败无限重试,当然也可以调用异步方法自己重试。

<代码>,下沉。EmitFailureHandler ALWAYS_RETRY_HANDLER=(signalType emitResult)→emitResult.isFailure();

在此之后就就可以调用Sinks.asFlux开心的使用各种操作符了。

在此之后就就可以调用水槽。asFlux开心的使用各种操作符了。

3,窗口函数

反应堆支持两类窗口聚合函数:

<李>

窗口类:返回Mono(流量)

<李>

缓冲区类:返回列表

在此场景中,使用缓冲即可满足需求,bufferTimeout (int最大容量、持续时间maxTime)支持最大个数,最大等待时间操作,Flink中的键操作可以用groupBy, collectMap来实现。

4,消费者处理

反应堆经过缓冲后是一个一个的发送数据,如果使用publishOn或subscribeOn处理的话,只等待下游的订阅处理完成才会重新要求新的数据,缓冲操作符才会重新发送数据。如果此时订阅消费者耗时较长,数据流会在缓冲流程阻塞,显然并不是我们想要的。

理想的操作是消费者在一个线程池里操作,可多线程并行处理,如果线程池满,再阻塞缓冲操作符。解决方案是自定义一个线程池,并且当然线程池如果任务满提交支持阻塞,可以用自定义RejectedExecutionHandler来实现:

, RejectedExecutionHandler executionHandler =, (r,执行人),→,{ 　　,,try { 　　,,,,executor.getQueue () .put (r); 　　,,},catch (InterruptedException e), { 　　,,,,Thread.currentThread () .interrupt (); 　　,,,,throw new RejectedExecutionException (“Producer thread interrupted",, e); 　　,,} 　　,}; 　　, 　　,new ThreadPoolExecutor (poolSize, poolSize, 　　,,,,0 l, TimeUnit.MILLISECONDS, 　　,,,,new SynchronousQueue<在(), 　　,,,,executionHandler);

三,总结

1,总结一下整体的执行流程

提交任务:提交数据支持同步异步两种方式,支持多线程提交,正常情况下响应很快,同步的方法如果队列满则阻塞。
丰富的操作符处理流式数据。
缓冲操作符产生的数据多线程处理:同步提交到单独的消费者线程池,线程池任务满则阻塞。
消费者线程池:支持阻塞提交,保证不丢消息,同时队列长度设置成0,因为前面已经有队列了。
背压:消费者线程池阻塞后,会背压到缓冲区操作符,并背压到缓冲队列,缓存队列满背压到数据提交者。