Spring WebFlux : onErrorResume not being called when exception is thrown halfway during the webclient reactive chain

Question

我编写了使用 webclient 调用另一个端点的代码，并希望添加响应式(Reactive)错误处理。不过，看来我对 doOnError 或 onErrorResume 的理解可能不正确:

webClient
        .get()
        .uri(someUri)
        .retrieve()
        .bodyToFlux(Some.class)
        .onErrorResume(throwable -> {
            log.error("Error occurred when calling other service: {}", throwable.getMessage());
            return Flux.error(new RunTimeException("Exception type: " + throwable.getClass() + " Exception message: " + throwable.getMessage()));
        });

目的是，此调用实际上是调用此调用的更大 react 链的一部分，并且如果在运行 api 调用时引发异常 (.get().retrieve())， onErrorResume 应该抛出异常并将其传递给更高级别的 react 链调用者。

我尝试通过以下方式对其有效性进行单元测试:

Mockito.when(webClient.get().uri(URI.create(uri)).retrieve()).thenThrow(new RuntimeException("Hello world exception thrown"));

但注意到异常刚刚被抛出，并且代码在 react 链的 .retrieve 步骤终止，而不是继续到 onErrorResume 步骤。

Answer 1

这是因为您的测试在构造 retrieve Mono 时抛出错误，而不是返回订阅时立即发出错误的功能性 Mono。因此，错误的不是您的数据流，而是处理数据流的管道本身。

您可以通过返回Mono.error来解决这个问题:

Mockito.when(webClient.get().uri(URI.create(uri)).retrieve())
    .thenReturn(Mono.error(new RuntimeException("Hello world exception thrown")));