从返回 Flux 的 WebClient 捕获 PrematureCloseException答案

【问题标题】：Catching PrematureCloseException from WebClient returning a Flux从返回 Flux 的 WebClient 捕获 PrematureCloseException
【发布时间】：2020-10-08 18:52:13
【问题描述】：

使用org.springframework.web.reactive.function.client.WebClient检索和无限reactor.core.publisher.Flux时如何捕获reactor.netty.http.client.PrematureCloseException？

为了重现这个问题，我创建了两个简单的 Spring Boot 应用程序。两者都基于org.springframework.boot:spring-boot-starter-webflux。

服务器代码是：

@RestController
public class TestserverRestController {

    @GetMapping(value="/huge-flux", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<Long> hugeFlux() {
        return Flux.range(1, Integer.MAX_VALUE).interval(Duration.ofSeconds(1));
    }

    @Autowired
    ConfigurableApplicationContext springContext;

    @GetMapping(value="/stop")
    public void stop() {
        springContext.close();
    }
}

客户端代码是：

@Component
public class TestclientRunner implements CommandLineRunner {
    @Override
    public void run(String... args) throws Exception {
        Flux<Long> flux = WebClient.create().get()
                .uri("http://localhost:9000/huge-flux")
                .accept(MediaType.APPLICATION_STREAM_JSON)
                .retrieve()
                .bodyToFlux(Long.class);
        flux.subscribe(val -> System.out.printf("Value %d%n", val));
    }
}

通过在网络浏览器中点击http://localhost:9000/stop，客户端被终止，并且这个错误出现在客户端的控制台上。

WARN 15635 --- [ctor-http-nio-2] reactor.netty.channel.FluxReceive : [id: 0x82a94759, L:0.0.0.0/0.0.0.0:61812] An exception has been observed post termination, use DEBUG level to see the full stack: reactor.core.Exceptions$ErrorCallbackNotImplemented: reactor.netty.http.client.PrematureCloseException: Connection prematurely closed DURING response

我希望能够捕获该错误以便我可以恢复。（在我的真实如果一个失败，项目将有备用服务器连接。）

我在客户端尝试过的事情包括：

flux.doOnCancel(()-> log.warning("CANCEL"));
flux.doOnTerminate(()-> log.warning("TERMINATE"));
flux.doOnComplete(()-> log.warning("COMPLETE"));
flux.doOnDiscard(Object.class, (o)-> log.warning("DISCARD"));
flux.doOnError((e)-> log.warning("ERROR"));

但是当服务器终止时，这些日志消息都不会被打印出来。

【问题讨论】：

你把flux.doOnError放在哪里了？在flux.subscribe 之前还是之后？

标签： spring-boot spring-webclient reactor-netty

【解决方案1】：

通过将客户端代码中的flux.subsribe() 行替换为：

            flux.subscribe(
                    val -> System.out.printf("Value %d%n", val),
                    ex -> System.err.printf("ERROR CONSUMER [%s] %s", ex.getClass(), ex.getMessage()),
                    () -> System.err.printf("COMPLETE CONSUMER"));

错误消费者可以捕获问题（打印“ERROR CONSUMER”）。

【讨论】：