如果通量为空，如何发出过滤掉的错误

Question

我有同步代码，我想用 reactor 进行非阻塞。

我想并行调用不同的 URI，调用可以返回响应、错误或什么都没有。

有3种情况：

• 一个请求返回一个响应，我返回它而不等待其他请求完成。 如果其他请求更早返回错误，我会删除错误
• 至少有一个请求返回错误，其他请求没有返回响应，我返回错误
• 所有请求均未返回（无响应、无错误），我没有返回任何内容

我已经以同步的方式做到了：

    AtomicReference<WebClientResponseException> responseException = new AtomicReference<>();
    
    String responseBody = Flux.fromIterable(uriList)
            .flatMap(url -> repo.sendRequest(uri)) 
            // sendRequest  returns a Mono that either emit a response, an error or nothing
            .onErrorContinue(WebClientResponseException.class, (error, element) -> {
                var webclientError = (WebClientResponseException) error;
                responseException.set(webclientError);
            })
            .blockFirst();
    
    return Pair.of(responseBody, responseException.get());

我想移除阻塞调用并返回一个 Mono

据我了解，我有点保持发生错误的状态，而我不能有反应器的状态。

我如何跟踪发生的错误但不发出它们，因为我想查看其他请求稍后是否发出结果？

这个版本好用吗？

AtomicReference<WebClientResponseException> responseException = new AtomicReference<>();
    
    return Flux.fromIterable(uriList)
            .flatMap(url -> repo.sendRequest(uri)) 
            // sendRequest  returns a Mono that either emit a response, an error or nothing
            .onErrorContinue(WebClientResponseException.class, (error, element) -> {
                var webclientError = (WebClientResponseException) error;
                responseException.set(webclientError);
            })
            .next()
            .switchIfEmpty(Mono.defer(() -> Mono.error(responseException.get())));

AtomicReference 会像闭包一样被关闭吗？

Answer 1

我认为flatMapDelayError 可能会实现您想要的，请参见以下示例：

int concurrency = 10;
int prefetch = 1;

Flux.just(
        Mono.error(new IOException("error")).delaySubscription(Duration.ofSeconds(2)),
        Mono.just("fast").delaySubscription(Duration.ofSeconds(4)),
        Mono.just("slow").delaySubscription(Duration.ofSeconds(6)))
    .flatMapDelayError(
        request -> request,
        concurrency,
        prefetch)
    .next()
    .doOnNext(result -> System.out.println("Result: " + result))

在此示例中，error 首先完成，但 -DelayError 运算符持有它，然后 fast 完成并作为结果发出。最后slow 永远不会完成，因为.next() 取消了剩余的请求，因为我们有一个结果。