如何处理 Flux.fromStream 中的异常抛出

Question

    public Mono<ResponseEntity<Data>> getData(@RequestParam List<String> tagIds){
   
   Flux<S3Object> s3ObjectFlux = Flux.fromStream(tagIds.stream())
           .parallel()
           .runOn(Schedulers.boundedElastic())
           .flatMap(id -> fetchResources(id)) //S3Exception is thrown here 
           .flatMap(idS3Object -> Mono.just(s3Object))
           .doOnError((throwable) -> log.error(throwable))
           .ordered((u1, u2) -> u2.hashCode() - u1.hashCode());

   Mono<Data> data = s3ObjectFlux.collectList()
           .map(s3Objects -> new Data(s3Objects));  
}

我正在遍历 tagIds 并在此处获取 s3 对象，如果该对象不存在或引发任何异常，我想记录并忽略它并继续下一步。但是在这种情况下，如果在迭代时从 fetchResource 方法中抛出 S3Exception，则会将错误作为 500 抛出给用户，而不是我想要空列表。

我没有看到其他选项，例如 onErrorMap 或 onErrorReturn

Answer 1

你会想要使用类似onErrorContinue的东西。

另外，这个.ordered((u1, u2) -> u2.hashCode() - u1.hashCode()) 并没有什么意义，您可以使用sequential 之类的东西来合并“rails”。

另外，我不太明白这个：.flatMap(idS3Object -> Mono.just(s3Object))。首先，它不能编译，可能你有错字，你想要一些东西flatMap(idS3Object -> Mono.just(s3Object))。在这种形式下，从一个对象创建一个Mono 并对其执行flatMap 完全没有意义。

把所有这些放在一起，我假设你想要这样的东西：

Flux<S3Object> s3ObjectFlux = Flux.fromStream(tagIds.stream())
        .parallel()
        .runOn(Schedulers.boundedElastic())
        .flatMap(this::fetchResources)
        .sequential()
        .onErrorContinue((error, object) -> {
            log.error(error);
        });