入门指南

反应流

该库是响应式流规范的实现。 响应式流 API 由以下组件组成：

1. 发布者

2. 订阅者

3. 订阅

Publisher是数量可能不受限制的排序元素的提供者，根据从其Subscriber或Subscriber的多个实例收到的需求进行发布。

为响应对Publisher.subscribe(Subscriber)的调用， Subscriber类上的方法的可能调用序列由以下协议给出：

onSubscribe onNext* (onError | onComplete)?

这意味着onSubscribe始终会发出信号，然后按照Subscriber的请求，可能有无限数量的onNext信号。 只要Subscription未取消，如果出现故障，则后跟一个onError信号，或者当没有更多元素可用时，后跟一个onComplete信号。

订阅者

Java Reactive Streams 驱动程序 API 镜像Java Sync 驱动程序API 以及导致网络 I/O 返回Publisher<T>类型的任何方法，其中T是操作的响应类型。

快速入门中的订阅者

在快速入门中，我们实现了几种不同的Subscriber类型。 尽管这是响应式流的人为场景，但我们确实会在开始下一个示例之前阻止一个示例的结果，以确保数据库的状态。

• ObservableSubscriber

  基本订阅者类是 ObservableSubscriber<T> ，一个存储SubscriberPublisher<T>它还包含一个await()方法，因此我们可以阻止结果，以在继续下一个示例之前确保数据库的状态。

• OperationSubscriber

  订阅后立即调用Subscription.request()的ObservableSubscriber的实现。

• PrintSubscriber

  OperationSubscriber的实现，用于在调用Subscriber.onComplete()方法时打印一条消息。

• ConsumerSubscriber

  的实现，它接受 并在调用 时调用OperationSubscriber 。ConsumerConsumer.accept(result)Subscriber.onNext(T result)

• PrintToStringSubscriber

  ConsumerSubscriber 的实现，用于在调用 Subscriber.onNext() 方法时打印 result 的string版本。

• PrintDocumentSubscriber

  ConsumerSubscriber的实现，用于在调用Subscriber.onNext()方法时打印Document类型的 JSON 版本。

阻塞和非阻塞示例

由于我们的Subscriber类型包含一个锁存器，该锁存器仅在调用Subscriber的onComplete()方法时才会释放，因此我们可以通过调用await方法使用该锁存器来阻止进一步的动作。 以下两个示例使用我们的自动请求PrintDocumentSubscriber 。

第一个是非阻塞的，第二个是阻塞的，等待Publisher完成：

// Create a publisher
Publisher<Document> publisher = collection.find();
// Non-blocking
publisher.subscribe(new PrintDocumentSubscriber());
Subscriber<Document> subscriber = new PrintDocumentSubscriber();
publisher.subscribe(subscriber);
subscriber.await(); // Block for the publisher to complete

发布者、订阅者和订阅

一般来说， Publisher 、 Subscriber和Subscription类型由低级 API 组成，预计用户和库将基于它们构建更具表现力的 API，而不是仅仅使用这些接口。 作为一个单独实现这些接口的库，用户将受益于这个不断发展的生态系统，这是响应式流的核心设计原则。