Reactor响应式编程(二)

简介

上篇文章我们简单的介绍了Reactor的发展史和基本的Flux和Mono的使用,本文将会进一步挖掘Reactor的高级用法,一起来看看吧。

自定义Subscriber

之前的文章我们提到了4个Flux的subscribe的方法:


这四个方法,需要我们使用lambda表达式来自定义consumer,errorConsumer,completeSonsumer和subscriptionConsumer这四个Consumer。

写起来比较复杂,看起来也不太方便,我们考虑一下,这四个Consumer是不是和Subscriber接口中定义的4个方法是一一对应的呢?


对的,所以我们有一个更加简单点的subscribe方法:


这个subscribe方法直接接收一个Subscriber类。从而实现了所有的功能。

自己写Subscriber太麻烦了,Reactor为我们提供了一个BaseSubscriber的类,它实现了Subscriber中的所有功能,还附带了一些其他的方法。

我们看下BaseSubscriber的定义:

注意,BaseSubscriber是单次使用的,这就意味着,如果它首先subscription到Publisher1,然后subscription到Publisher2,那么将会取消对第一个Publisher的订阅。

因为BaseSubscriber是一个抽象类,所以我们需要继承它,并且重写我们需要自己实现的方法。

下面看一个自定义的Subscriber:


BaseSubscriber中有很多以hook开头的方法,这些方法都是我们可以重写的,而Subscriber原生定义的on开头的方法,在BaseSubscriber中都是final的,都是不能重写的。

我们看一个定义:


可以看到,它内部实际上调用了hook的方法。

上面的CustSubscriber中,我们重写了两个方法,一个是hookOnSubscribe,在建立订阅的时候调用,一个是hookOnNext,在收到onNext信号的时候调用。

在这些方法中,给了我们足够的自定义空间,上面的例子中我们调用了request(1),表示再请求一个元素。

其他的hook方法还有: hookOnComplete, hookOnError, hookOnCancel 和 hookFinally。

Backpressure处理

我们之前讲过了,reactive stream的最大特征就是可以处理Backpressure。

什么是Backpressure呢?就是当consumer处理过不来的时候,可以通知producer来减少生产速度。

我们看下BaseSubscriber中默认的hookOnSubscribe实现:


可以看到默认是request无限数目的值。 也就是说默认情况下没有Backpressure。

通过重写hookOnSubscribe方法,我们可以自定义处理速度。

除了request之外,我们还可以在publisher中限制subscriber的速度。


在Flux中,我们有一个limitRate方法,可以设定publisher的速度。

比如subscriber request(100),然后我们设置limitRate(10),那么最多producer一次只会产生10个元素。

创建Flux

接下来,我们要讲解一下怎么创建Flux,通常来讲有4种方法来创建Flux。

使用generate

第一种方法就是最简单的同步创建的generate.

先看一个例子:


输出结果:


上面的例子中,我们使用generate方法来同步的生成元素。

generate接收两个参数:


第一个参数是stateSupplier,用来指定初始化的状态。

第二个参数是一个generator,用来消费SynchronousSink,并生成新的状态。

上面的例子中,我们每次将state+1,一直加到10。

然后使用subscribe来将所有的生成元素输出。

使用create

Flux也提供了一个create方法来创建Flux,create可以是同步也可以是异步的,并且支持多线程操作。

因为create没有初始的state状态,所以可以用在多线程中。

create的一个非常有用的地方就是可以将第三方的异步API和Flux关联起来,举个例子,我们有一个自定义的EventProcessor,当处理相应的事件的时候,会去调用注册到Processor中的listener的一些方法。


我们怎么把这个Listener的响应行为和Flux关联起来呢?


使用create就够了,create接收一个consumer参数:


这个consumer的本质是去消费FluxSink对象。

上面的例子在MyEventListener的事件中对FluxSink对象进行消费。

使用push

push和create一样,也支持异步操作,但是同时只能有一个线程来调用next, complete 或者 error方法,所以它是单线程的。

使用Handle

Handle和上面的三个方法不同,它是一个实例方法。

它和generate很类似,也是消费SynchronousSink对象。


不同的是它的参数是一个BiConsumer,是没有返回值的。

看一个使用的例子:

 

0

Reactor响应式编程(一)

简介

Reactor是reactivex家族的一个非常重要的成员,Reactor是第四代的reactive library,它是基于Reactive Streams标准基础上开发的,主要用来构建JVM环境下的非阻塞应用程序。

今天给大家介绍一下Reactor。

Reactor简介

Reactor是基于JVM的非阻塞API,他直接跟JDK8中的API相结合,比如:CompletableFuture,Stream和Duration等。

它提供了两个非常有用的异步序列API:Flux和Mono,并且实现了Reactive Streams的标准。

并且还可以和reactor-netty相结合,作为一些异步框架的底层服务,比如我们非常熟悉的Spring MVC 5中引入的WebFlux。

我们知道WebFlux的底层使用的是reactor-netty,而reactor-netty又引用了Reactor。所以,如果你在POM中引入了webFlux依赖:


那么项目将会自动引入Reactor。

如果你用的不是Spring webflux,没关系,你可以直接添加下面的依赖来使用Reactor:

reactive programming的发展史

最最开始的时候微软为.NET平台创建了Reactive Extensions (Rx) library。接着RxJava实现了JVM平台的Reactive。

然后Reactive Streams标准出现了,它定义了Java平台必须满足的的一些规范。并且已经集成到JDK9中的java.util.concurrent类中。

在Flow中定义了实现Reactive Streams的四个非常重要的组件,分别是Publisher,Subscriber,Subscription和Processor。

Iterable-Iterator 和Publisher-Subscriber的区别

一般来说reactive在面向对象的编程语言中是以观察者模式的扩展来使用的。

我们来具体看一下这个观察者模式的实现,以Publisher和Subscriber为例:


上面定义了两个接口,Publisher和Subscriber,Publisher的作用就是subscribe到subscriber。

而subscriber定义了4个on方法,用来触发特定的事件。

那么Publisher中的subscribe是怎么触发Subscriber的onSubscribe事件呢?

很简单,我们看一个具体的实现:


上面的例子是PullPublisher的subscribe实现。我们可以看到,在这个subscribe中触发了subscriber.onSubscribe方法。而这就是观察者模式的秘密。

或者说,当Publisher调用subscribe的时候,是主动push subscriber的onSubscribe方法。

熟悉Iterable-Iterator模式的朋友应该都知道,Iterator模式,其实是一个主动的pull模式,因为需要不断的去调用next()方法。所以它的控制权是在调用方。

为什么要使用异步reactive

在现代应用程序中,随着用户量的增多,程序员需要考虑怎么才能提升系统的处理能力。

传统的block IO的方式,因为需要占用大量的资源,所以是不适合这样的场景的。我们需要的是NO-block IO。

JDK中提供了两种异步编程的模型:

第一种是Callbacks,异步方法可以通过传入一个Callback参数的形式来在Callback中执行异步任务。比较典型的像是java Swing中的EventListener。

第二中就是使用Future了。我们使用Callable来提交一个任务,然后通过Future来拿到它的运行结果。

这两种异步编程会有什么问题呢?

callback的问题就在于回调地狱。熟悉JS的朋友应该很理解这个回调地狱的概念。

简单点讲,回调地狱就是在callback中又使用了callback,从而造成了这种callback的层级调用关系。

而Future主要是对一个异步执行的结果进行获取,它的 get()实际上是一个block操作。并且不支持异常处理,也不支持延迟计算。

当有多个Future的组合应该怎么处理呢?JDK8 实际上引入了一个CompletableFuture类,这个类是Future也是一个CompletionStage,CompletableFuture支持then的级联操作。不过CompletableFuture提供的方法不是那么的丰富,可能满足不了我的需求。

于是我们的Reactor来了。

Flux

Reactor提供了两个非常有用的操作,他们是 Flux 和 Mono。 其中Flux 代表的是 0 to N 个响应式序列,而Mono代表的是0或者1个响应式序列。

我们看一个Flux是怎么transfer items的:

Reactor响应式编程(一)

先看下Flux的定义:


可以看到Flux其实就是一个Publisher,用来产生异步序列。

Flux提供了非常多的有用的方法,来处理这些序列,并且提供了completion和error的信号通知。

相应的会去调用Subscriber的onNext, onComplete, 和 onError 方法。

Mono

我们看下Mono是怎么transfer items的:

Reactor响应式编程(一)

看下Mono的定义:


Mono和Flux一样,也是一个Publisher,用来产生异步序列。

Mono因为只有0或者1个序列,所以只会触发Subscriber的onComplete和onError方法,没有onNext。

另一方面,Mono其实可以看做Flux的子集,只包含Flux的部分功能。

Mono和Flux是可以互相转换的,比如Mono#concatWith(Publisher)返回一个Flux,而 Mono#then(Mono)返回一个Mono.

Flux和Mono的基本操作

我们看下Flux创建的例子:


可以看到Flux提供了很多种创建的方式,我们可以自由选择。

再看看Flux的subscribe方法:


subscribe可以一个参数都没有,也可以多达4个参数。

看下没有参数的情况:

注意,没有参数并不表示Flux的对象不被消费,只是不可见而已。

看下带参数的情况:consumer用来处理on each事件,errorConsumer用来处理on error事件,completeConsumer用来处理on complete事件,subscriptionConsumer用来处理on subscribe事件。

前面的3个参数很好理解,我们来举个例子:


我们构建了从1到4的四个整数的Flux,on each就是打印出来,如果中间有错误的话,就输出Error,全部完成就输出Done。

那么最后一个subscriptionConsumer是做什么用的呢?

subscriptionConsumer accept的是一个Subscription对象,我们看下Subscription的定义:


Subscription 定义了两个方法,用来做初始化用的,我们可以调用request(n)来决定这次subscribe获取元素的最大数目。

比如上面我们的例子中,虽然构建了4个整数,但是最终输出的只有2个。

上面所有的subscribe方法,都会返回一个Disposable对象,我们可以通过Disposable对象的dispose()方法,来取消这个subscribe。

Disposable只定义了两个方法:


dispose的原理是向Flux 或者 Mono发出一个停止产生新对象的信号,但是并不能保证对象产生马上停止。

有了Disposable,当然要介绍它的工具类Disposables。

Disposables.swap() 可以创建一个Disposable,用来替换或者取消一个现有的Disposable。

Disposables.composite(…​)可以将多个Disposable合并起来,在后面统一做处理。

总结

本文介绍了Reactor的基本原理和两非常重要的组件Flux和Mono,下一篇文章我们会继续介绍Reactor core的一些更加高级的用法。敬请期待。

0