响应式流规范

原文详见官网：Reactive Stream

Reactive Streams

​ Reactive Streams的目的是为异步流处理提供一个无阻塞背压的标准。最新版本可在Maven Central上获得：

<dependency>
  <groupId>org.reactivestreams</groupId>
  <artifactId>reactive-streams</artifactId>
  <version>1.0.3</version>
</dependency>
<dependency>
  <groupId>org.reactivestreams</groupId>
  <artifactId>reactive-streams-tck</artifactId>
  <version>1.0.3</version>
  <scope>test</scope>
</dependency>

Goals, Design and Scope

​ 在异步系统中，处理数据流（尤其是其容量未预先确定的“实时”数据）需要特别小心。最突出的问题是需要仔细控制资源消耗，以便快速数据源不会压倒流目的地。为了能够在协作网络主机或单个机器内的多个CPU核心上并行使用计算资源，需要异步。

​ Reactive Streams的主要目标是管理跨异步边界的流数据交换-考虑将元素传递到另一个线程或线程池-同时确保接收端不会被迫缓冲任意数量的数据。换句话说，背压是这个模型的一个组成部分，以便允许在线程之间进行调解的队列被限制。如果背压信号是同步的，异步处理的好处将被否定（另请参阅Reactive Manifesto），因此已经注意要求Reactive Streams实现的所有方面都具有完全非阻塞和异步行为。

​ 本说明书的目的是允许创建许多符合的实现，这些实现凭借遵守规则将能够平滑地互操作，从而在流应用的整个处理图上保留上述益处和特性。

​ 应该注意的是，流操纵（变换、分裂、合并等）的精确性质是不可预测的。本规范未涵盖。反应流只关心在不同的API组件之间调解数据流。在它们的开发过程中，已经注意确保可以表达组合流的所有基本方式。总之，Reactive Streams是JVM的面向流的库的标准和规范：

• 处理可能无限数量的元素
• 按顺序
• 在组件之间异步传递元素
• 强制性非阻塞背压

Reactive Streams规范由以下部分组成：

1. API指定实现反应流的类型，并实现不同实现之间的互操作性。
2. 技术兼容性工具包（TCK）是用于实现一致性测试的标准测试套件。
3. 实现可以自由地实现规范中未涵盖的其他功能，只要它们符合API要求并通过TCK中的测试

API 组件

​ API由以下组件组成，这些组件需要由Reactive Stream实现提供：

1. Publisher
2. Subscriber
3. Subscription
4. Processor

​ 发布服务器是可能无限数量的序列元素的提供者，根据从其订阅服务器接收到的需求发布这些元素。响应于对 Publisher.subscribe(Subscriber) 的调用，对 Subscriber 上的方法的可能调用序列由以下协议给出：

onSubscribe onNext* (onError | onComplete)?

​ 这意味着onSubscribe总是被发送信号， 随后是可能无限数量的onNext信号（如Subscriber所请求的），如果存在故障，则随后是onError信号，或者当没有更多的元件可用时，则随后是onComplete信号-只要Subscription不被取消，所有这些都是如此。

注意事项

• 以下规范使用来自https：//www.ietf.org/rfc/rfc2119.txt的大写字母绑定词

术语

规范

1. Publisher (Code)

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

2. Subscriber (Code)

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

3. Subscription (Code)

public interface Subscription {
    public void request(long n);
    public void cancel();
}

4.Processor (Code)

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

异步与同步处理

​ Reactive Streams API规定所有元素（onNext）或终止信号（onError，onComplete）的处理不得阻塞Publisher。但是，每个on*处理程序都可以同步或异步处理事件。示例：

nioSelectorThreadOrigin map(f) filter(p) consumeTo(toNioSelectorOutput)

​ 它有一个双循环的起点和一个双循环的终点。让我们假设起点和终点都是选择器事件循环。Subscription.request(n)必须从目的地链接到原点。现在，每个实现都可以选择如何做到这一点。下面的代码使用管道字符|来表示线程边界（队列和调度），使用R#来表示资源（可能是线程）。

nioSelectorThreadOrigin | map(f) | filter(p) | consumeTo(toNioSelectorOutput)
-------------- R1 ----  | - R2 - | -- R3 --- | ---------- R4 ----------------

​ 在这个例子中，3个消费者中的每一个，map、filter和consumeTo异步地调度工作。它可以在同一个事件循环（蹦床），单独的线程，无论什么。

nioSelectorThreadOrigin map(f) filter(p) | consumeTo(toNioSelectorOutput)
------------------- R1 ----------------- | ---------- R2 ----------------

​ 这里，通过向NioSelectorOutput事件循环添加工作，异步调度的只是最后一步。map和filter步骤在源线程上同步执行。或者另一种实现可以将操作融合到最终消费者：

nioSelectorThreadOrigin | map(f) filter(p) consumeTo(toNioSelectorOutput)
--------- R1 ---------- | ------------------ R2 -------------------------

​ 所有这些变体都是“异步流”。它们都有自己的位置，每个都有不同的权衡，包括性能和实现复杂性。

​ Reactive Streams契约允许实现灵活地管理资源和调度，并在非阻塞，异步，动态推拉流的范围内混合异步和同步处理。为了允许所有参与API元素的完全异步实现-Publisher/Subscription/Subscriber/Processor-这些接口定义的所有方法都返回void。

队列边界

​ 一个基本的设计原则是，所有的缓冲区大小是有界的，这些界限必须是已知的，并由用户控制。这些界限用元素计数来表示（这反过来又转换为onNext的调用计数）。任何旨在支持无限流（特别是高输出速率流）的实现都需要始终沿着边界，以避免内存不足错误并限制资源使用。

​ 由于背压是强制性的，因此可以避免使用无界缓冲器。一般来说，队列可能无限制增长的唯一情况是发布方在很长一段时间内保持比订阅方更高的速率，但这种情况是由反压处理的。队列边界可以由订阅者用信号通知对适当数量的元素的需求来控制。在任何时间点，订阅者都知道：

• 请求的元素总数：P
• 已处理的元素数：N

​ 那么，可以到达的元素的最大数量是P - N，直到向发布者发出更多的需求信号。在订户也知道其输入缓冲器中的元素B的数量的情况下，则该界限可以被细化为P - B - N。

​ 发布者必须尊重这些界限，而不管它所代表的源是否可以被反压。如果源的生产率不受影响（例如时钟滴答声或鼠标移动），发布者必须选择缓冲或删除元素以遵守强加的边界。

​ 在接收到一个元素之后，用信号通知对一个元素的需求的订户有效地实现了停止和等待协议，其中需求信号等同于确认。通过提供多个要素的需求，确认成本被摊销。值得注意的是，允许订阅者在任何时间点发出需求信号，从而避免发布者和订阅者之间不必要的延迟（即保持其输入缓冲区充满，而不必等待完整的往返）。