Reactor:深入理解reactor core

Posted 2023-04-08

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

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

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

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

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

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

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

我们看下BaseSubscriber的定义：

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

下面看一个自定义的Subscriber：

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

我们看一个定义：

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

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

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

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

我们之前讲过了，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，通常来讲有4种方法来创建Flux。

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

先看一个例子：

输出结果：

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

generate接收两个参数：

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

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

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

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

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

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

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

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

使用create就够了，create接收一个consumer参数：

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

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

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

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

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

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

看一个使用的例子：

本文的例子 learn-reactive

深入了解SOG Reactor多功能工具

便携性是多功能工具存在的主要原因之一。

而在近几年，市场需求似乎发生了些变化，各家公司都开始发布简约风格的工具。

尤其是SOG，发布了许多简洁的设计。

比如它的Baton系列和Reactor系列。

SOG称Reactor上面配有十个小工具。

但其实主要用到的还是一把钳子、一把刀子、一个螺丝刀、剪线钳、开瓶器、钩子、挂绳孔。

Reactor上的钳子采用的是SOG的专利“复合杠杆”设计。

当你在手柄上施加压力的时候，这种压力会转移到钳头上，在使用时能更省力气。

杠杆原理的使用，使SOG的钳子设计比其他的钳子设计要好。

在这里，SOG添加了剪线钳性能。

螺丝刀是SOG在2016年推出的新款式。

除了剪线钳，螺丝刀通常也是多功能工具必备的工具之一。

这种螺丝刀如果不用的时候，可以收起来。

螺丝刀有大号和六角螺丝刀。

Reactor上的刀片长6.35厘米，钢材采用8Cr13MoV。

8Cr13MoV钢与420不锈钢相似。

刀片形状有点像反向的几何刀形，而且刀刃很锋利。

配有SOG SAT辅助弹簧系统，刀片通过拇指螺柱打开。

这里的刀用了衬垫锁设计，黑色涂层刀片。

手柄的设计不对称，这也导致了在使用时不是很舒服。

它的口袋夹是深入式的，而且是镂空的设计。

像它在办公室里，或家里一般使用还是可以的，但并不能和真正的那些工具相提并论。

优点：

深入式口袋夹

弹簧辅助刀片

复合杠杆钳子

缺点：

由于手柄设计，钳子不居中

刀片开刀有些困难

小编在线答疑 等你来撩！

喜欢的话点个赞吧~