从projectreactor使用Flux时生成和处理异常的正确方法是什么

我正在使用 io.projectreactor 3 (reactor-core 3.2.6.RELEASE)，我注意到错误处理方面存在一些差异。不幸的是，官方文档没有提供足够的细节来解决我的问题。

我有以下 4 个片段。在某些情况下，异常将被忽略，而在其他情况下，它将进一步抛出。实际产生和消费异常的方式是怎样的呢？

片段1

在这种情况下，异常将被忽略，并且 main() 将完成而不会收到异常。

import reactor.core.publisher.Flux;

class Scratch {
    public static void main(String[] args) throws Throwable {
        Flux.push(sink -> {
            sink.next(1);
            sink.next(2);
        }).doOnNext(e -> {
            throw new RuntimeException("HELLO WORLD");
        }).subscribe(System.out::println, e -> {
            throw new RuntimeException(e);
        });
        System.out.println("DONE");
    }
}

Output:

DONE

片段2

与上面的示例类似，只是我们不使用 Flux.push 而是使用 Flux.just。 Main() 将收到异常。

import reactor.core.publisher.Flux;

class Scratch {
    public static void main(String[] args) throws Throwable {
        Flux.just(
                1
        ).doOnNext(e -> {
            throw new RuntimeException("HELLO WORLD");
        }).subscribe(System.out::println, e -> {
            throw new RuntimeException(e);
        });
        System.out.println("DONE");
    }
}

Output:

Exception in thread "main" java.lang.RuntimeException: java.lang.RuntimeException: HELLO WORLD
    at Scratch.lambda$main$1(scratch_15.java:10)
...

片段3

我们通过调用sink.error来发出异常信号。 Main() 不会收到异常。

import reactor.core.publisher.Flux;

class Scratch {
    public static void main(String[] args) throws Throwable {
        Flux.push(sink -> {
            sink.next(1);
            sink.next(2);
            sink.error(new RuntimeException("HELLO WORLD"));
        }).subscribe(System.out::println, e -> {
            throw new RuntimeException(e);
        });
        System.out.println("DONE");
    }
}

Output:

1
2
DONE

片段4

我们直接抛出异常。 Main() 将收到异常。

import reactor.core.publisher.Flux;

class Scratch {
    public static void main(String[] args) throws Throwable {
        Flux.push(sink -> {
            sink.next(1);
            sink.next(2);
            throw new RuntimeException("HELLO WORLD");
        }).subscribe(System.out::println, e -> {
            throw new RuntimeException(e);
        });
        System.out.println("DONE");
    }
}

Output

1
2
Exception in thread "main" java.lang.RuntimeException: java.lang.RuntimeException: HELLO WORLD
    at Scratch.lambda$main$1(scratch_15.java:10)
...

使用反应核心时处理异常的正确方法是什么？唯一可靠的方法似乎根本不使用错误回调，而是用 try/catch 包围 Flux.subscribe。但在那种情况下我总是收到UnsupportedOperationException而不是原来的异常，然后我需要使用Exceptions.isErrorCallbackNotImplemented要检查它是否来自反应式，请提取嵌套异常然后抛出它。

这当然可以做到，但需要在我们订阅 Flux 的每个地方一致地完成。这对我来说看起来不太好。我在这里缺少什么？

在你所有的例子中，问题都是从.subscribe(...)错误处理 lambda。

如果您希望在主块中引发异常，请使用block()变体。

如果您想测试错误是否在整个管道中传播，请使用StepVerifier.create(pipeline).expectError(...).verify().

.subscribe一般来说是为了获得“终端”状态的通知，而不是为了恢复或重新抛出错误（使用onError*上游运营商）。

The just基于 -的示例似乎正确地传播了异常，因为它们在订阅时不执行用户提供的代码，因此在执行期间没有 try/catchsubscribe(Consumer<Throwable>).

push, like generate/create/defer and compose，执行一些用户定义的逻辑（Consumer<FluxSink>），在订阅时。他们防范整个Consumer抛出异常并尝试传播它（作为onErrorsignal）而不是直接抛出它。

但如果失败了Consumer是在执行其中之一时引起的sink的方法，如果subscriber重新抛出：我们进入递归，其中调用接收器调用接收器。当我们检测到水槽的递归排水时，我们通过退出来防止这种无限情况。

这就是为什么push基于示例之后触发错误sink.next or in sink.error（示例 1 和 3）无法在 main 中产生异常：

1. Consumer被申请;被应用
2. sink.next被调用并且下一个运算符创建异常 1，或者sink.error叫做
3. 异常1到达subscribe并作为异常 2 重新抛出
4. 这会短路Consumer.apply，异常 2 被传递给sink.error
5. 接收器已经被调用，所以我们突破以避免无限递归
6. 例外2从未见过

另一方面，在示例 4 中，我们不再调用接收器的方法，并且原始异常不会首先到达订阅者：

1. Consumer被申请;被应用
2. 直接抛出异常1
3. 这会短路Consumer.apply异常 1 被传递给sink.error
4. 传播到订阅者
5. 将其重新抛出为异常 2
6. 异常2出现在main方法中