java - 在 Spring Boot 客户端中接收 Flux

Question

这是 Spring 5 Web Reactive - How can we use WebClient to retrieve streamed data in a Flux? 的后续问题

我尝试遵循如何使用 Spring WebClient 接收 Flux 的建议，但实际上遇到了网络问题。

在服务器端，代码是一个简单的 Controller ，公开 Mongo 存储库的 findAll 方法:

@RequestMapping("power")
public Flux<Power> getAll() {
    return powerRepository.findAll();
}

在客户端上的消费代码类似于上面给出的答案:

@Bean
public CommandLineRunner readFromApiServer() {
    return new CommandLineRunner() {
        @Override
        public void run(String... args) throws Exception {
            WebClient webClient = WebClient.create("http://localhost:8080");
            Flux<Power> flux = webClient.get().uri("/power").accept(MediaType.APPLICATION_STREAM_JSON).exchange()
                    .flatMap(response -> response.bodyToFlux(Power.class));
            flux.subscribe();
        }
    };
}

但这会引发异常:

2017-02-27 08:19:41.281 ERROR 99026 --- [ctor-http-nio-5] r.ipc.netty.channel.ChannelOperations : [HttpClient] Error processing connection. Requesting close the channel

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1 at io.netty.buffer.AbstractReferenceCountedByteBuf.release0(AbstractReferenceCountedByteBuf.java:101) ~[netty-all-4.1.8.Final.jar:4.1.8.Final]

我正在使用当前的 Spring Boot 2.0.0 BUILD-SNAPSHOT。

这个异常告诉我什么？我怎样才能做到正确？

Answer 1

所有 CommandLineRunner bean 都会在 Spring Boot 应用程序启动时执行。如果没有守护线程(即当前应用程序不是 Web 应用程序)，则应用程序将在所有运行程序执行完毕后关闭。

在您的情况下，仅使用flux.subscribe()“启动链并请求无界需求”(javadoc)，因此此方法调用不会阻塞。我怀疑这个命令行运行程序在您有机会对您的通量执行任何操作之前就返回了，应用程序关闭并且您的网络资源关闭。

此外，您没有对 HTTP 请求的结果执行任何操作。 我认为使用以下内容更新您的代码示例应该可以解决问题:

@Bean
public CommandLineRunner readFromApiServer() {
    return new CommandLineRunner() {
        @Override
        public void run(String... args) throws Exception {
            WebClient webClient = WebClient.create("http://localhost:8080");
            Flux<Power> flux = webClient.get().uri("/power").accept(MediaType.APPLICATION_STREAM_JSON).exchange()
                    .flatMap(response -> response.bodyToFlux(Power.class));
            List<Power> powerList = flux.collectList().block();
            // do something with the result?
        }
    };
}