常见问题、最佳实践与「我该如何……？」
目录
本节涵盖以下内容：
1. 如何包装一个同步阻塞调用？
信息源往往是同步且阻塞的。要在您的 Reactor 应用中处理这类源，请采用以下模式：
Mono blockingWrapper = Mono.fromCallable(() -> { 1
return /* make a remote synchronous call */ 2
});
blockingWrapper = blockingWrapper.subscribeOn(Schedulers.boundedElastic()); 3
使用
fromCallable创建一个新的Mono。返回该异步阻塞资源。
确保每次订阅都发生在来自
Schedulers.boundedElastic()的专用 worker 上。
您应当使用 Mono，因为源只返回一个值。您应当使用 Schedulers.boundedElastic，因为它会创建一个专用线程来等待阻塞资源，而不影响其他非阻塞处理；同时它还确保可创建的线程数量，以及在峰值期间可排队并延迟处理的阻塞任务数量，都有上限。
请注意，subscribeOn 并不会订阅 Mono。它只是指定在发生 subscribe 调用时使用哪种 Scheduler。
另外请注意，subscribeOn 操作符应紧跟在源之后，而其他操作符则定义在 subscribeOn 包装之后。
2. 我对 Flux 使用了一个操作符，但它似乎没生效。这是怎么回事？
请确保您所 .subscribe() 的那个变量，确实受到了您认为应当作用于它的那些操作符的影响。
Reactor 操作符是装饰器。它们返回一个不同的实例，包装源序列并添加行为。这就是为什么使用操作符的首选方式是把调用链式串联起来（chain）。
对比以下两个示例：
Flux<String> flux = Flux.just("something", "chain");
flux.map(secret -> secret.replaceAll(".", "*")); 1
flux.subscribe(next -> System.out.println("Received: " + next));
错误就在这里。结果并没有被赋回
flux变量。
Flux<String> flux = Flux.just("something", "chain");
flux = flux.map(secret -> secret.replaceAll(".", "*"));
flux.subscribe(next -> System.out.println("Received: " + next));
下面的示例更好（因为它更简单）：
Flux.just("something", "chain")
.map(secret -> secret.replaceAll(".", "*"))
.subscribe(next -> System.out.println("Received: " + next));
第一个版本输出如下：
Received: something
Received: chain
另外两个版本输出预期的值，如下：
Received: *********
Received: *****
3. 我的 Mono zipWith 或 zipWhen 从不被调用
考虑下面的示例：
myMethod.process("a") // this method returns Mono<Void>
.zipWith(myMethod.process("b"), combinator) //this is never called
.subscribe();
如果源 Mono 是 empty 的，或是一个 Mono<Void>（从任何角度看，Mono<Void> 都是空的），某些组合操作就永远不会被调用。
对于任何转换器来说都是典型情况，例如 zip 静态方法，或 zipWith、zipWhen 操作符——它们（按定义）需要来自每个源的一个元素才能产生输出。
因此，在 zip 的源上使用抑制数据的操作符会有问题。这类操作符的例子包括 then()、thenEmpty(Publisher<Void>)、ignoreElements() 和 ignoreElement()，以及 when(Publisher…)。
类似地，那些使用 Function<T,?> 来调节行为的操作符（如 flatMap），至少需要发出一个元素，Function 才有机会被应用。把这些操作应用到一个空（或 <Void>）序列上永远不会产生元素。
您可以使用 .defaultIfEmpty(T) 和 .switchIfEmpty(Publisher<T>)，分别用默认值或一个回退 Publisher<T> 来替换 T 类型的空序列，这有助于避免其中一些情况。请注意，这不适用于 Flux<Void>/Mono<Void> 源，因为您只能切换到另一个 Publisher<Void>，而它仍然保证是空的。下例使用 defaultIfEmpty：
myMethod.emptySequenceForKey("a") // this method returns empty Mono<String>
.defaultIfEmpty("") // this converts empty sequence to just the empty String
.zipWhen(aString -> myMethod.process("b")) //this is called with the empty String
.subscribe();
4. 将 zip 与空完成发布者一起使用
当把 zip 操作符与空完成发布者（即不发出任何元素就完成的发布者）一起使用时，务必注意以下行为。
考虑下面的测试用例：
@Test
public void testZipEmptyCompletionAllSubscribed() {
AtomicInteger cnt = new AtomicInteger();
Mono<Integer> mono1 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> mono2 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> zippedMono = Mono.zip(mono1, mono2, (v1, v2) -> v1);
zippedMono.subscribe();
assertEquals(2, cnt.get());
}
虽然在本例中，得到的 zippedMono 会同时订阅 mono1 和 mono2，但这种行为并非在所有情况下都有保证。例如，考虑下面的测试用例：
@Test
public void testZipEmptyCompletionOneSubscribed() {
AtomicInteger cnt = new AtomicInteger();
Mono<Integer> mono1 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> mono2 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> mono3 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> zippedMono = Mono.zip(mono1, Mono.zip(mono2, mono3, (v1, v2) -> v1), (v1, v2) -> v1);
zippedMono.subscribe();
assertEquals(1, cnt.get());
}
在本例中，当 mono1 空完成时，zippedMono 立即完成，而不会订阅 mono2 和 mono3。
因此，当使用 zip 操作符组合空完成发布者时，无法保证得到的发布者会订阅所有这些空完成发布者。
如果需要保持第二个测试用例所示的语义，并确保订阅所有待 zip 的发布者，请考虑使用 singleOptional 操作符，如下方测试用例所示：
@Test
public void testZipOptionalAllSubscribed() {
AtomicInteger cnt = new AtomicInteger();
Mono<Integer> mono1 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> mono2 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Integer> mono3 = Mono.create(sink -> {
cnt.incrementAndGet();
sink.success();
});
Mono<Optional<Integer>> zippedMono =
Mono.zip(
mono1.singleOptional(),
Mono.zip(mono2.singleOptional(), mono3.singleOptional(), (v1, v2) -> v1),
(v1, v2) -> v1);
zippedMono.subscribe();
assertEquals(3, cnt.get());
}
5. 如何用 retryWhen 模拟 retry(3)？
retryWhen 操作符可能相当复杂。希望下面这段代码能通过尝试模拟一个更简单的 retry(3)，帮助您理解它的工作原理：
AtomicInteger errorCount = new AtomicInteger();
Flux<String> flux =
Flux.<String>error(new IllegalArgumentException())
.doOnError(e -> errorCount.incrementAndGet())
.retryWhen(Retry.from(companion -> 1
companion.map(rs -> { 2
if (rs.totalRetries() < 3) return rs.totalRetries(); 3
else throw Exceptions.propagate(rs.failure()); 4
})
));
我们通过从一个
Functionlambda 适配而来定制Retry，而不是提供一个具体类。companion 发出
RetrySignal对象，其中携带了迄今已重试的次数和最近一次失败。为了允许三次重试，我们考虑索引 < 3 并返回一个要发出的值（这里我们直接返回索引）。
为了让序列以错误终止，我们在这三次重试之后抛出原始异常。
6. 如何用 retryWhen 实现指数退避？
指数退避（exponential backoff）会让每次重试之间的延迟逐渐增大，从而避免压垮源系统、引发全面崩溃。其道理在于：如果源产生了错误，它本已处于不稳定状态，不太可能立即恢复。因此盲目地立刻重试很可能再次产生错误，加剧不稳定。
自 3.3.4.RELEASE 起，Reactor 为此类重试提供了一个构建器，可与 Flux#retryWhen 配合使用：Retry.backoff。
下例展示了该构建器的简单用法，用钩子在重试尝试延迟的前后记录日志消息。它会延迟重试，并增大每次尝试之间的延迟（伪代码：delay = 100ms * 2^attempt_number_starting_at_zero）：
AtomicInteger errorCount = new AtomicInteger();
Flux<String> flux =
Flux.<String>error(new IllegalStateException("boom"))
.doOnError(e -> { 1
errorCount.incrementAndGet();
System.out.println(e + " at " + LocalTime.now());
})
.retryWhen(Retry
.backoff(3, Duration.ofMillis(100)).jitter(0d) 2
.doAfterRetry(rs -> System.out.println("retried at " + LocalTime.now() + ", attempt " + rs.totalRetries())) 3
.onRetryExhaustedThrow((spec, rs) -> rs.failure()) 4
);
我们将记录源发出错误的时间并计数。
我们配置一个最多 3 次尝试、无抖动的指数退避重试。
我们还会记录重试发生的时间以及重试尝试的序号（从 0 开始）。
默认情况下，会抛出一个
Exceptions.retryExhausted异常，并以最后一次failure()作为原因。这里我们将其定制为直接把原因作为onError发出。
订阅后，它会失败并在打印出以下内容后终止：
java.lang.IllegalStateException: boom at 00:00:00.0
retried at 00:00:00.101, attempt 0 1
java.lang.IllegalStateException: boom at 00:00:00.101
retried at 00:00:00.304, attempt 1 2
java.lang.IllegalStateException: boom at 00:00:00.304
retried at 00:00:00.702, attempt 2 3
java.lang.IllegalStateException: boom at 00:00:00.702
约 100ms 后第一次重试
约 200ms 后第二次重试
约 400ms 后第三次重试
7. 使用 publishOn() 时如何确保线程亲和性？
如 Schedulers 中所述，publishOn() 可用于切换执行上下文。publishOn 操作符会影响其下方链中其余操作符运行的线程上下文，直到再次出现 publishOn 为止。因此 publishOn 的位置很关键。
考虑下面的示例：
Sinks.Many<Integer> dataSinks = Sinks.many().unicast().onBackpressureBuffer();
Flux<Integer> source = dataSinks.asFlux();
source.publishOn(scheduler1)
.map(i -> transform(i))
.publishOn(scheduler2)
.doOnNext(i -> processNext(i))
.subscribe();
map() 中的 transform 函数运行在 scheduler1 的一个 worker 上，而 doOnNext() 中的 processNext 方法运行在 scheduler2 的一个 worker 上。每次订阅都会获得自己的 worker，因此推送给对应订阅者的所有元素都在同一个 Thread 上发布。
您可以使用单线程调度器，来为链中的不同阶段或不同订阅者确保线程亲和性。
8. 上下文日志（MDC）的好模式是什么？
大多数日志框架都支持上下文日志，让用户存储会反映到日志 pattern 中的变量，通常通过一个名为 MDC（「Mapped Diagnostic Context」，映射诊断上下文）的 Map 来实现。这是 Java 中 ThreadLocal 最常见的用途之一，因此这种模式假定被记录日志的代码与某个 Thread 是一一绑定的。
在 Java 8 之前，这或许是一个安全的假设，但随着 Java 语言引入函数式编程元素，情况已经有所变化……
让我们以一个原本是命令式、采用模板方法模式、后来转为更函数式风格的 API 为例。在模板方法模式下，靠的是继承。而在更函数式的做法中，高阶函数被传入以定义算法的「步骤」。事情现在更偏声明式而非命令式，这就让库得以自主决定每个步骤应在何处运行。例如，知道底层算法的哪些步骤可以并行后，库可以使用 ExecutorService 来并行执行其中一些步骤。
这类函数式 API 的一个具体例子，是 Java 8 引入的 Stream API 及其 parallel() 版本。在并行 Stream 中用 MDC 记录日志并非免费的午餐：需要确保 MDC 在每个步骤中都被捕获并重新应用。
函数式风格之所以能支持这类优化，是因为每个步骤都与线程无关、且引用透明；但它可能打破「单一 Thread」的 MDC 假设。确保任何上下文信息都能被所有阶段访问，最地道的方式是通过组合链把该上下文传递下去。在开发 Reactor 的过程中，我们遇到了同类问题，并希望避免这种非常直白、显式的做法。这就是引入 Context 的原因：只要以 Flux 和 Mono 作为返回值，它就会沿执行链传播，方式是让各阶段（操作符）窥视其下游阶段的 Context。因此，Reactor 没有使用 ThreadLocal，而是提供了这个类似 map 的对象，它绑定到一个 Subscription 而非某个 Thread。
既然我们已经确定，在声明式 API 中假定 MDC「开箱即用」并不是最佳假设，那么我们该如何针对响应式流中的事件（onNext、onError 和 onComplete）执行带上下文的日志记录呢？
本 FAQ 条目提供了一种可行的中间方案，适合想以直接、显式的方式围绕这些信号记录日志的场景。请务必先阅读向响应式序列添加上下文一节，尤其是要理解：写入必须发生在操作符链的底部，其上方的操作符才能看到它。
要把上下文信息从 Context 传到 MDC，最简单的方式是用一个 doOnEach 操作符包装日志语句，并配上少量样板代码。这些样板代码既取决于您所选的日志框架/抽象，也取决于您想放进 MDC 的信息，因此它必须放在您自己的代码库中。
下面是这样一个辅助函数的示例，围绕单个 MDC 变量、专注于记录 onNext 事件，并使用 Java 9 增强的 Optional API：
public static <T> Consumer<Signal<T>> logOnNext(Consumer<T> logStatement) {
return signal -> {
if (!signal.isOnNext()) return; 1
Optional<String> toPutInMdc = signal.getContextView().getOrEmpty("CONTEXT_KEY"); 2
toPutInMdc.ifPresentOrElse(tpim -> {
try (MDC.MDCCloseable cMdc = MDC.putCloseable("MDC_KEY", tpim)) { 3
logStatement.accept(signal.get()); 4
}
},
() -> logStatement.accept(signal.get())); 5
};
}
doOnEach的信号包含onComplete和onError。本例中我们只关心记录onNext。我们将从 Reactor
Context中提取一个感兴趣的值（参见 The Context API 一节）。本例使用 SLF4J 2 的
MDCCloseable，支持 try-with-resource 语法，在日志语句执行后自动清理 MDC。真正的日志语句由调用方以
Consumer<T>（onNext 值的消费者）形式提供。如果期望的键未在
Context中设置，我们走另一条路径，即不往 MDC 放入任何内容。
使用这段样板代码可确保我们是 MDC 的良好公民：在执行日志语句之前设置键，并在之后立即移除。不存在污染后续日志语句 MDC 的风险。
当然，这只是一个建议。您可能想从 Context 中提取多个值，或在 onError 时记录日志。您可能想为这些情况创建额外的辅助方法，或编写一个借助更多 lambda 来覆盖更多场景的统一方法。
无论如何，前面这个辅助方法的用法大致如下面这个响应式 web 控制器所示：
@GetMapping("/byPrice")
public Flux<Restaurant> byPrice(@RequestParam Double maxPrice, @RequestHeader(required = false, name = "X-UserId") String userId) {
String apiId = userId == null ? "" : userId; 1
return restaurantService.byPrice(maxPrice))
.doOnEach(logOnNext(r -> LOG.debug("found restaurant {} for ${}", 2
r.getName(), r.getPricePerPerson())))
.contextWrite(Context.of("CONTEXT_KEY", apiId)); 3
}
我们需要从请求头中取出上下文信息，放入
Context。这里我们用
doOnEach把辅助方法应用到Flux上。请记住：操作符能看到定义在其下方的Context值。我们用所选的键
CONTEXT_KEY，把来自 header 的值写入Context。
在这种配置下，restaurantService 可以在一个共享线程上发出数据，但日志仍会为每个请求引用正确的 X-UserId。
为完整起见，我们也看看一个错误日志辅助函数可以是什么样子：
public static Consumer<Signal<?>> logOnError(Consumer<Throwable> errorLogStatement) {
return signal -> {
if (!signal.isOnError()) return;
Optional<String> toPutInMdc = signal.getContextView().getOrEmpty("CONTEXT_KEY");
toPutInMdc.ifPresentOrElse(tpim -> {
try (MDC.MDCCloseable cMdc = MDC.putCloseable("MDC_KEY", tpim)) {
errorLogStatement.accept(signal.getThrowable());
}
},
() -> errorLogStatement.accept(signal.getThrowable()));
};
}
几乎没什么变化，只是我们会检查该 Signal 确实是一个 onError，并把该错误（一个 Throwable）提供给日志语句 lambda。
在控制器中应用这个辅助函数，与我们之前所做的非常相似：
@GetMapping("/byPrice")
public Flux<Restaurant> byPrice(@RequestParam Double maxPrice, @RequestHeader(required = false, name = "X-UserId") String userId) {
String apiId = userId == null ? "" : userId;
return restaurantService.byPrice(maxPrice))
.doOnEach(logOnNext(v -> LOG.info("found restaurant {}", v))
.doOnEach(logOnError(e -> LOG.error("error when searching restaurants", e)) 1
.contextWrite(Context.of("CONTEXT_KEY", apiId));
}
如果
restaurantService发出错误，它会在这里带上 MDC 上下文被记录下来。