ParallelFlux doOnNext 如何处理异常答案

【问题标题】：ParallelFlux doOnNext how to handle ExceptionParallelFlux doOnNext 如何处理异常
【发布时间】：2019-10-17 13:46:54
【问题描述】：

在我的项目中，我有这个：

ParallelFlux<Device> flux = Flux.fromIterable(children)
                .delayElements(Duration.ofMillis(10))
                .parallel(18)
                .runOn(Schedulers.elastic(), 10)
                .doOnNext(c -> recursiveValidationThroughoutChildren(c, tracker)
                });

其中 recursiveValidationThroughoutChildren 是具有此声明的方法：

boolean recursiveValidationThroughoutChildren(Device d, NodeChangesTracker tracker) throws Exception;

我不明白的是如何处理最后一个方法抛出的异常。我希望将异常传播到 ParallelFlux 之外。是否可以？正确的处理方法是什么？

【问题讨论】：

相信您应该可以在这里找到问题的答案：stackoverflow.com/questions/53595420/…
我同意@Rozart - 但如果不同意，请告诉我们为什么它不能解决它，这样我们就可以避免将其作为骗子关闭:-)

标签： spring-boot project-reactor reactor

【解决方案1】：

我按照@Rozart 建议的链接进行操作，但无法按照说明应用该解决方案。我不得不稍微改变一下：

ParallelFlux<Device> flux = Flux.fromIterable(children)
                .delayElements(Duration.ofMillis(10))
                .parallel(18)
                .runOn(Schedulers.elastic(), 10)
                .doOnNext(child -> {
                    try {
                        recursiveValidationThroughoutChildren(child, tracker);
                    } catch (Exception ex) {
                        Flux.error(ex);
                    }
                });

需要进行更改，因为 ParallelFlux 不支持“句柄”方法，因此我必须添加一个 try catch 并使用 Flux.error 重新启动异常。我不知道这是否是一种好的做法，但这是我让它发挥作用的唯一方法。

【讨论】：