一、背景

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

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

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

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

二、实现过程

Flink对流式解决做的很好的封装，应用Flink的时候简直不必关怀线程池、积压、数据失落等问题，然而应用Reactor实现相似的性能就必须对Reactor运行原理比拟理解，并且通过不同场景下测试，否则很容易出问题。

上面列举出实现过程中的外围点：

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

入门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...

2、背压反对

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

数据提交有两个办法：

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

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

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

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

3、窗口函数

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

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

在此场景中，应用buffer即可满足需要，bufferTimeout(int maxSize, Duration maxTime)反对最大个数，最大等待时间操作，Flink中的keys操作能够用groupBy、collectMap来实现。

4、消费者解决

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

三、总结

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

  1. 提交工作：提交数据反对同步异步两种形式，反对多线程提交，失常状况下响应很快，同步的办法如果队列满则阻塞。
  2. 丰盛的操作符解决流式数据。
  3. buffer操作符产生的数据多线程解决：同步提交到独自的消费者线程池，线程池工作满则阻塞。
  4. 消费者线程池：反对阻塞提交，保障不丢音讯，同时队列长度设置成0，因为后面曾经有队列了。
  5. 背压：消费者线程池阻塞后，会背压到buffer操作符，并背压到缓冲队列，缓存队列满背压到数据提交者。

2、和Flink的比照

实现的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...