（19）Reactor Processors——响应式Spring的道法术器

骑猪看日落 2022-05-28 09:58 225阅读 0赞

> 本系列其他文章见：[《响应式Spring的道法术器》][Spring]。  
> 前情提要：[响应式流][Link 1] | [Reactor3快速上手][Reactor3] | [深入理解响应式流规范][Link 2]

## 2.9 Processor ##

`Processor`既是一种特别的发布者（`Publisher`）又是一种订阅者（`Subscriber`）。 所以你能够订阅一个`Processor`，也可以调用它们提供的方法来手动插入数据到序列，或终止序列。

前面一直在聊响应式流的四个接口中的三个：`Publisher`、`Subscriber`、`Subscription`，唯独`Processor`迟迟没有提及。原因在于想用好它们不太容易，多数情况下，我们应该进行避免使用`Processor`，通常来说仅用于一些特殊场景。

### 2.9.1 使用 Sink 来线程安全地生成流 ###

比起直接使用Processor，更好的方式是通过调用`sink()`来得到它的Sink。这个Sink是线程安全的，可以用于在应用程序中多线程并发地生成数据。例如，通过`UnicastProcessor`得到一个线程安全的 sink：

UnicastProcessor<Integer> processor = UnicastProcessor.create();
        FluxSink<Integer> sink = processor.sink(overflowStrategy);

多个线程可以并发地通过下边的方法生成数据到sink。

sink.next(n);

看到这里是不是感觉跟`generate`生成数据流的方式很像？所以**Reactor官方建议，当你想要使用Processor的时候，首先看看能否用`generate`实现同样的功能，或者看看是否有相应的操作符可以达到你想要的效果**。

### 2.9.2 Reactor 内置的 Processor ###

Reactor Core 内置多种 Processor。这些 processor 具有不同的语法，大概分为三类。

*  **直接的（direct）**（DirectProcessor 和 UnicastProcessor）：这些 processors 只能通过直接 调用 Sink 的方法来推送数据。
 *  **同步的（synchronous）**（EmitterProcessor 和 ReplayProcessor）：这些 processors 既可以直接调用 Sink 方法来推送数据，也可以通过订阅到一个上游的发布者来同步地产生数据。
 *  **异步的（asynchronous）**（WorkQueueProcessor 和 TopicProcessor）：这些 processors 可以将从多个上游发布者得到的数据推送下去。由于使用了 RingBuffer 的数据结构来缓存多个来自上游的数据，因此更加有健壮性。

异步的 processor 在实例化的时候最复杂，因为有许多不同的选项。因此它们暴露出一个 Builder 接口。 而简单的 processors 有静态的工厂方法。

**1）DirectProcessor**

`DirectProcessor`可以将信号分发给零到多个订阅者（Subscriber）。它是最容易实例化的，使用静态方法 create() 即可。另一方面，它的不足是无法处理背压。所以，当`DirectProcessor`推送的是 N 个元素，而至少有一个订阅者的请求个数少于 N 的时候，就会发出一个`IllegalStateException`。

一旦 Processor 结束（通常通过调用它的 Sink 的 error(Throwable) 或 complete() 方法）， 虽然它允许更多的订阅者订阅它，但是会立即向它们重新发送终止信号。

**2）UnicastProcessor**

`UnicastProcessor`可以使用一个内置的缓存来处理背压。代价就是它最多只能有一个订阅者（上一节的例子通过`publish`转换成了`ConnectableFlux`，所以可以接入两个订阅者）。

`UnicastProcessor`有多种选项，因此提供多种不同的`create`静态方法。例如，它默认是 无限的（unbounded） ：如果你在在订阅者还没有请求数据的情况下让它推送数据，它会缓存所有数据。

可以通过提供一个自定义的 Queue 的具体实现传递给 create 工厂方法来改变默认行为。如果给出的队列是有限的（bounded）， 并且缓存已满，而且未收到下游的请求，processor 会拒绝推送数据。

在上边“有限的”例子中，还可以在构造 processor 的时候提供一个回调方法，这个回调方法可以在每一个 被拒绝推送的元素上调用，从而让开发者有机会清理这些元素。

**3）EmitterProcessor**

`EmitterProcessor`能够向多个订阅者发送数据，并且可以对每一个订阅者进行背压处理。它本身也可以订阅一个发布者并同步获得数据。

最初如果没有订阅者，它仍然允许推送一些数据到缓存，缓存大小由`bufferSize`定义。 之后如果仍然没有订阅者订阅它并消费数据，对`onNext`的调用会阻塞，直到有订阅者接入 （这时只能并发地订阅了）。

因此第一个订阅者会收到最多`bufferSize`个元素。然而之后，后续接入的订阅者只能获取到它们开始订阅**之后**推送的数据。这个内部的缓存会继续用于背压的目的。

默认情况下，如果所有的订阅者都取消了订阅，它会清空内部缓存，并且不再接受更多的订阅者。这一点可以通过 create 静态工厂方法的 autoCancel 参数来配置。

**4）ReplayProcessor**

`ReplayProcessor`会缓存直接通过自身的 Sink 推送的元素，以及来自上游发布者的元素， 并且后来的订阅者也会收到重发（replay）的这些元素。

可以通过多种配置方式创建它：

*  缓存一个元素（cacheLast）。
 *  缓存一定个数的历史元素（create(int)），所有的历史元素（create()）。
 *  缓存基于时间窗期间内的元素（createTimeout(Duration)）。
 *  缓存基于历史个数和时间窗的元素（createSizeOrTimeout(int, Duration)）。

**5）TopicProcessor**

`TopicProcessor`是一个异步的 processor，它能够重发来自多个上游发布者的元素， 这需要在创建它的时候配置`shared`（build() 的 share(boolean) 配置）。

> 如果你企图在并发环境下通过并发的上游发布者调用`TopicProcessor`的`onNext`、 `onComplete`，或`onError`方法，就必须配置`shared`。否则，并发调用就是非法的，从而 processor 是完全兼容响应式流规范的。

`TopicProcessor`能够对多个订阅者发送数据。它通过对每一个订阅者关联一个线程来实现这一点， 这个线程会一直执行直到 processor 发出`onError`或`onComplete`信号，或关联的订阅者被取消。 最多可以接受的订阅者个数由构造者方法`executor`指定，通过提供一个有限线程数的 `ExecutorService`来限制这一个数。

这个 processor 基于一个`RingBuffer`数据结构来存储已发送的数据。每一个订阅者线程 自行管理其相关的数据在`RingBuffer`中的索引。

这个 processor 也有一个`autoCancel`构造器方法：如果设置为`true`（默认的），那么当 所有的订阅者取消之后，上游发布者也就被取消了。

**6）WorkQueueProcessor**

`WorkQueueProcessor`也是一个异步的 processor，也能够重发来自多个上游发布者的元素， 同样在创建时需要配置`shared`（它多数构造器配置与`TopicProcessor`相同）。

它放松了对响应式流规范的兼容，但是好处就在于相对于`TopicProcessor`来说需要更少的资源。 它仍然基于`RingBuffer`，但是不再要求每一个订阅者都关联一个线程，因此相对于`TopicProcessor`来说更具扩展性。

代价在于分发模式有些区别：来自订阅者的请求会汇总在一起，并且这个 processor 每次只对一个 订阅者发送数据，因此需要循环（round-robin）对订阅者发送数据，而不是一次全部发出的模式（无法保证完全公平的循环分发）。

`WorkQueueProcessor`多数构造器方法与`TopicProcessor`相同，比如`autoCancel`、`share`， 以及`waitStrategy`。下游订阅者的最大数目同样由构造器`executor`配置的`ExecutorService` 决定。

> 注意：最好不要有太多订阅者订阅`WorkQueueProcessor`，因为这会锁住 processor。如果你需要限制订阅者数量，最好使用一个`ThreadPoolExecutor`或 `ForkJoinPool`。这个 processor 能够检测到（线程池）容量并在订阅者过多时抛出异常。

[Spring]: http://blog.csdn.net/get_set/article/details/79466657
[Link 1]: http://blog.csdn.net/get_set/article/details/79466402
[Reactor3]: http://blog.csdn.net/get_set/article/details/79480172
[Link 2]: http://blog.csdn.net/get_set/article/details/79514670