项目Reactor：onErrorResume后平面图

Question

Flux.just("a", "b")
        .flatMap(s -> s.equals("a") ? Mono.error(new RuntimeException() : Flux.just(s + "1", s + "2"))
        .onErrorResume(throwable -> Mono.empty())
        .subscribe(System.out::println);

你好

在这里，我做了两个元素的通量，然后通过平面图公开第一个异常，第二个通量。

使用onErrorResume，我希望输出

b1
b2

但什么也得不到。谁能解释一下为什么会这样？

谢谢

Answer 1

这个问题已经有了一个可靠的答案，即为什么会发生这种情况。要回答一个有点离题的问题，即如何实现预期输出：

必须将onErrorResume调用移到平面图中：

Flux.just("a", "b")
    .flatMap(s ->
        (s.equals("a") ? Mono.error<RuntimeException>(RuntimeException()) : Flux.just(s + "1", s + "2"))
             .onErrorResume(ex -> Mono.empty())
    )
    .subscribe(System.out::println)

这样输出和预期的一样

b1
b2

Answer 2

鉴于此：

Flux.just("a", "b", "c")
        .flatMap { s ->
            if (s == "b") 
                Mono.error<RuntimeException>(RuntimeException()) 
            else 
                Flux.just(s + "1", s + "2")
        }.onErrorResume { throwable -> Mono.just("d") }.log()
        .subscribe { println(it) }

输出为：

12:35:19.673 [main] INFO reactor.Flux.OnErrorResume.1 - onSubscribe(FluxOnErrorResume.ResumeSubscriber)
12:35:19.676 [main] INFO reactor.Flux.OnErrorResume.1 - request(unbounded)
12:35:19.677 [main] INFO reactor.Flux.OnErrorResume.1 - onNext(a1)
a1
12:35:19.677 [main] INFO reactor.Flux.OnErrorResume.1 - onNext(a2)
a2
12:35:19.712 [main] INFO reactor.Flux.OnErrorResume.1 - onNext(d)
d
12:35:19.713 [main] INFO reactor.Flux.OnErrorResume.1 - onComplete()

这是怎么回事？onErrorResume（）正在应用于平面地图（）运算符返回的发布服务器。由于在“b”上发布服务器发出失败信号，因此平面地图（）发布服务器不再执行并且onErrorResume（）运算符继续使用其回退发布。

onErrorResume（）的文档清楚地表明，原始发布服务器由于错误而完成，并且回退接管：