项目反应堆：flatMap 后的 onErrorResume答案

【问题标题】：Project reactor: onErrorResume after flatMap项目反应堆：flatMap 后的 onErrorResume
【发布时间】：2018-09-01 01:28:53
【问题描述】：

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);

你好！

在这里，我制作了两个元素的通量，然后通过 flatMap 将第一个元素暴露给异常，将第二个元素暴露给另一个 Flux。

onErrorResume 我期待输出

b1
b2

但一无所获。谁能解释一下为什么会这样？

谢谢。

【问题讨论】：

标签： java project-reactor reactor reactive-streams

【解决方案1】：

鉴于此：

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() 正在应用于flatMap() 运算符返回的发布者。由于在“b”上 Publisher 发出失败信号，flatMap() Publisher 不再执行，onErrorResume() 运算符继续使用其后备发布。

documentation for onErrorResume() 清楚地表明原始 Publisher 因错误而完成，并且后备接管：

【讨论】：

嘿，要得到他期望的输出，即得到b1，b2作为输出，我们不能使用onErrorReturn()吗？否则我们怎样才能做到这一点？
@uneq95 看看我的补充答案。

【解决方案2】：

codependent 已经给出了这个问题的可靠答案，为什么会发生这种情况。回答一点题外话如何达到预期的输出：

onErrorResume 调用必须移动到flatMap：

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

【讨论】：