关于java:使用Reactor完成类似的Flink的操作

Flink 在解决流式工作的时候有很大的劣势，其中 windows 等操作符能够很不便的实现聚合工作，然而 Flink 是一套独立的服务，业务流程中如果想应用须要将数据发到 kafka，用 Flink 解决完再发到 kafka，而后再做业务解决，流程很繁琐。

比方在业务代码中想要实现相似 Flink 的 window 按工夫批量聚合性能，如果纯手动写代码比拟繁琐，应用 Flink 又太重，这种场景下应用响应式编程 RxJava、Reactor 等的 window、buffer 操作符能够很不便的实现。

响应式编程框架也早已有了背压以及丰盛的操作符反对，能不能用响应式编程框架解决相似 Flink 的操作呢，答案是必定的。

本文应用 Reactor 来实现 Flink 的 window 性能来举例，其余操作符实践上雷同。文中波及的代码：github

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()

官网文档：https://projectreactor.io/doc…

下面办法的对象背压策略反对两种：BackpressureBuffer、BackpressureError，在此场景必定是抉择 BackpressureBuffer，须要指定缓存队列，初始化办法如下：Queues.<String>get(queueSize).get()

数据提交有两个办法：

emitNext 指定提交失败策略同步提交
tryEmitNext 异步提交，返回提交胜利、失败状态

在此场景咱们不心愿丢数据，可自定义失败策略，提交失败有限重试，当然也能够调用异步办法本人重试。

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

在此之后就就能够调用 Sinks.asFlux 开心的应用各种操作符了。

Reactor 反对两类窗口聚合函数：

window 类：返回 Mono(Flux<T>)
buffer 类：返回 List<T>

在此场景中，应用 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 有限重试。

本文源码地址：https://github.com/sofn/react…

Reactor 官网文档：https://projectreactor.io/doc…

Flink 文档：https://ci.apache.org/project…

Reactive 操作符：http://reactivex.io/documenta…

本文作者： 木小丰 ，美团 Java 高级工程师，关注架构、软件工程、全栈等，不定期分享软件研发过程中的实际、思考。欢送关注公共号：Java 研发

本文链接：https://lesofn.com/archives/s…

关于java:使用Reactor完成类似的Flink的操作

一、背景

二、实现过程

1、创立 Flux 和发送数据拆散

2、背压反对

3、窗口函数

4、消费者解决

三、总结

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

2、和 Flink 的比照

四、附录