创建 Flux 或 Mono 并订阅它的简单方式

Doubao 中英对照 Simple Ways to Create a Flux or Mono and Subscribe to It

目录

• 1. subscribe 方法示例
• 2. 用 Disposable 取消 subscribe()
• 3. Lambda 的替代方案：BaseSubscriber
• 4. 关于背压以及重塑请求的方式
  • 4.1. 改变下游需求量的操作符

开始使用 Flux 和 Mono 最简单的方式，就是使用它们各自类中众多工厂方法之一。

例如，要创建一个 String 序列，您既可以逐一列举这些字符串，也可以把它们放进一个集合中并据此创建 Flux，如下所示：

Flux<String> seq1 = Flux.just("foo", "bar", "foobar");

List<String> iterable = Arrays.asList("foo", "bar", "foobar");
Flux<String> seq2 = Flux.fromIterable(iterable);

其他工厂方法示例如下：

Mono<String> noData = Mono.empty(); 1

Mono<String> data = Mono.just("foo");

Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3); 2

• 请注意，即使没有值，该工厂方法也会保留泛型类型。

• 第一个参数是范围的起始值，第二个参数是要产生的元素个数。

在订阅方面，Flux 和 Mono 利用了 Java 8 的 lambda。您有多种 .subscribe() 变体可供选择，它们针对不同的回调组合接受 lambda，如下列方法签名所示：

subscribe(); 1

subscribe(Consumer<? super T> consumer); 2

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer); 3

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer); 4

subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer,
          Consumer<? super Subscription> subscriptionConsumer); 5

• 订阅并触发序列。

• 对每个产生的值做些处理。

• 处理值，同时对错误做出反应。

• 处理值和错误，并在序列成功完成时运行一些代码。

• 处理值、错误和成功完成，并对本次 subscribe 调用产生的 Subscription 做些处理。

提示

这些变体都会返回一个对订阅的引用，当不再需要数据时，您可以用它来取消订阅。取消时，源应停止产生值并清理它创建的所有资源。这种取消并清理的行为在 Reactor 中由通用的 Disposable 接口来表示。

1. subscribe 方法示例

本节给出 subscribe 方法五种签名各自的最小示例。以下代码展示了不带参数的基本方法示例：

Flux<Integer> ints = Flux.range(1, 3); 1
ints.subscribe(); 2

• 设置一个 Flux，当订阅者接入时产生三个值。

• 以最简单的方式订阅。

前述代码不产生任何可见输出，但它确实有效。该 Flux 产生了三个值。如果我们提供一个 lambda，就可以让这些值显现出来。下一个 subscribe 方法示例展示了让值显现的一种方式：

Flux<Integer> ints = Flux.range(1, 3); 1
ints.subscribe(i -> System.out.println(i)); 2

• 设置一个 Flux，当订阅者接入时产生三个值。

• 用一个会打印这些值的订阅者来订阅。

前述代码产生如下输出：

1
2
3

为演示下一个签名，我们有意引入一个错误，如下例所示：

Flux<Integer> ints = Flux.range(1, 4) 1
      .map(i -> { 2
        if (i <= 3) return i; 3
        throw new RuntimeException("Got to 4"); 4
      });
ints.subscribe(i -> System.out.println(i), 5
      error -> System.err.println("Error: " + error));

• 设置一个 Flux，当订阅者接入时产生四个值。

• 我们需要一个 map，以便对某些值做不同处理。

• 对于大多数值，直接返回该值。

• 对某个值，强制抛出错误。

• 用一个包含错误处理器的订阅者来订阅。

现在我们有两个 lambda 表达式：一个用于处理预期的内容，另一个用于处理错误。前述代码产生如下输出：

1
2
3
Error: java.lang.RuntimeException: Got to 4

subscribe 方法的下一个签名同时包含错误处理器和完成事件处理器，如下例所示：

Flux<Integer> ints = Flux.range(1, 4); 1
ints.subscribe(i -> System.out.println(i),
    error -> System.err.println("Error " + error),
    () -> System.out.println("Done")); 2

• 设置一个 Flux，当订阅者接入时产生四个值。

• 用一个包含完成事件处理器的 Subscriber 来订阅。

错误信号和完成信号都是终止事件，并且二者互斥（您永远不会同时收到两者）。要让完成处理器生效，我们必须注意不要触发错误。

完成回调没有输入，用一对空括号表示：它对应 Runnable 接口中的 run 方法。前述代码产生如下输出：

1
2
3
4
Done

2. 用 Disposable 取消 subscribe()

所有这些基于 lambda 的 subscribe() 变体都返回 Disposable 类型。在这种情况下，Disposable 接口表示这样一个事实：可以通过调用其 dispose() 方法来取消订阅。

对于 Flux 或 Mono，取消是一个信号，表示源应停止产生元素。然而，并不保证它会立即生效：有些源产生元素的速度极快，以至于它们甚至可能在收到取消指令之前就已经完成。

Disposables 类提供了一些围绕 Disposable 的工具方法。其中，Disposables.swap() 创建一个 Disposable 包装器，让您可以原子地取消并替换某个具体的 Disposable。例如，在 UI 场景中，当用户每次点击按钮时，您都想取消一个请求并用新的请求替换它，这时它会很有用。处置包装器本身就会将其关闭。这样做会处置当前的具体值以及今后所有尝试的替换。

另一个有趣的工具是 Disposables.composite(…)。这个组合让您可以收集多个 Disposable（例如与一次服务调用相关的多个在途请求），并在之后一次性处置它们所有。一旦组合的 dispose() 方法被调用，任何再添加 Disposable 的尝试都会立即处置它。

3. Lambda 的替代方案：BaseSubscriber

还有一个更通用的 subscribe 方法，它接受一个完整的 Subscriber，而不是用 lambda 组合出一个。为了帮助编写这样的 Subscriber，我们提供了一个可扩展的类，名为 BaseSubscriber。

注意

BaseSubscriber（或其子类）的实例是单次使用的，这意味着如果一个 BaseSubscriber 被订阅到第二个 Publisher，它会取消对第一个 Publisher 的订阅。这是因为重复使用同一个实例会违反 Reactive Streams 规范中“Subscriber 的 onNext 方法不得被并行调用”的规则。因此，匿名实现只有在直接声明于 Publisher#subscribe(Subscriber) 调用之内时才是可行的。

现在我们可以实现这样一个订阅者，称之为 SampleSubscriber。以下示例展示了它将如何接入一个 Flux：

SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
Flux<Integer> ints = Flux.range(1, 4);
ints.subscribe(ss);

以下示例展示了 SampleSubscriber 作为 BaseSubscriber 的一个极简实现可能的样子：

import org.reactivestreams.Subscription;

import reactor.core.publisher.BaseSubscriber;

public class SampleSubscriber<T> extends BaseSubscriber<T> {

	@Override
	public void hookOnSubscribe(Subscription subscription) {
		System.out.println("Subscribed");
		request(1);
	}

	@Override
	public void hookOnNext(T value) {
		System.out.println(value);
		request(1);
	}
}

SampleSubscriber 类继承了 BaseSubscriber，后者是 Reactor 中推荐用于用户自定义 Subscriber 的抽象类。该类提供了若干钩子，您可以重写它们来调整订阅者的行为。默认情况下，它会触发一个无界请求，行为与 subscribe() 完全一致。然而，当您想要自定义请求量时，扩展 BaseSubscriber 才更有用得多。

要自定义请求量，最起码要像我们这样实现 hookOnSubscribe(Subscription subscription) 和 hookOnNext(T value)。在我们的例子中，hookOnSubscribe 方法向标准输出打印一条语句并发起第一个请求。然后 hookOnNext 方法打印一条语句并发起后续请求，每次一个请求。

SampleSubscriber 类产生如下输出：

Subscribed
1
2
3
4

BaseSubscriber 还提供了 requestUnbounded() 方法以切换到无界模式（等价于 request(Long.MAX_VALUE)），以及一个 cancel() 方法。

它还有其他钩子：hookOnComplete、hookOnError、hookOnCancel 和 hookFinally（后者在序列终止时总会被调用，并以 SignalType 参数传入终止的类型）

备注

您几乎肯定需要实现 hookOnError、hookOnCancel 和 hookOnComplete 方法。您可能还想实现 hookFinally 方法。SampleSubscriber 是一个执行有界请求的 Subscriber 的最精简实现。

4. 关于背压以及重塑请求的方式

在 Reactor 中实现背压时，消费者压力传回源的方式是向上游操作符发送一个 request。当前请求的总和有时被称为当前“需求量”或“待处理请求”。需求量上限为 Long.MAX_VALUE，表示一个无界请求（意思是“以您最快的速度生产”——基本上相当于禁用背压）。

第一个请求来自订阅时的最终订阅者，然而最直接的几种订阅方式都会立即触发一个 Long.MAX_VALUE 的无界请求：

• subscribe() 及其大多数基于 lambda 的变体（除了带 Consumer<Subscription> 的那个）

• block()、blockFirst() 和 blockLast()

• 遍历 toIterable() 或 toStream() 的结果

自定义初始请求最简单的方式，是用一个重写了 hookOnSubscribe 方法的 BaseSubscriber 来 subscribe，如下例所示：

Flux.range(1, 10)
    .doOnRequest(r -> System.out.println("request of " + r))
    .subscribe(new BaseSubscriber<Integer>() {

      @Override
      public void hookOnSubscribe(Subscription subscription) {
        request(1);
      }

      @Override
      public void hookOnNext(Integer integer) {
        System.out.println("Cancelling after having received " + integer);
        cancel();
      }
    });

前述代码片段打印出如下内容：

request of 1
Cancelling after having received 1

注意

在操控请求时，您必须注意产生足够的需求量以推进序列，否则您的 Flux 可能会“卡住”。这就是为什么 BaseSubscriber 在 hookOnSubscribe 中默认采用无界请求。重写该钩子时，您通常应至少调用一次 request。

4.1. 改变下游需求量的操作符

需要记住的一点是，在订阅层级表达的需求量可以被上游链中的每个操作符重塑。一个典型的例子是 buffer(N) 操作符：如果它收到一个 request(2)，会被解释为对两个完整缓冲的需求。因此，由于缓冲需要 N 个元素才算满，buffer 操作符会把请求重塑为 2 x N。

您可能还注意到，有些操作符带有接受一个名为 prefetch 的 int 输入参数的变体。这是另一类会修改下游请求的操作符。它们通常是处理内部序列的操作符，会从每个传入元素派生出一个 Publisher（如 flatMap）。

**预取（Prefetch）**是一种调整对这些内部序列所发初始请求的方式。如果未指定，这些操作符大多数以 32 的需求量开始。

这些操作符通常还实现了一种补充优化（replenishing）：一旦操作符看到预取请求已完成 75%，它就会向上游重新请求 75%。这是一种启发式优化，目的是让这些操作符主动预判即将到来的请求。

最后，有几个操作符让您可以直接调整请求：limitRate 和 limitRequest。

limitRate(N) 会拆分下游请求，使它们以更小的批次向上游传播。例如，向 limitRate(10) 发出一个 100 的请求，最终会导致至多 10 个 10 的请求被传播到上游。请注意，在这种形式下，limitRate 实际上实现了前面讨论过的补充优化。

该操作符还有一个变体，让您也能调整补充量（在该变体中称为 lowTide）：limitRate(highTide, lowTide)。选择 lowTide 为 0 会得到严格的 highTide 请求批次，而不是由补充策略进一步调整的批次。

另一方面，limitRequest(N) 会将下游请求限制为一个最大总需求量。它把请求累加，最多到 N。如果某个 request 不会使总需求量超过 N，该请求就会被完整地传播到上游。当源已发出该数量之后，limitRequest 认为序列已完成，向下游发送一个 onComplete 信号，并取消源。