ParallelFlux doOnNext如何处理异常

Question

在我的项目中，我有这样的内容：

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

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

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

Answer 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不支持“handle”方法，所以我不得不添加一个try catch并使用Flux.error重新启动异常。我不知道这是不是一个好的练习，但这是唯一的方法，我得到它的工作。