0

这是 Spring 5 Web Reactive 的后续问题- 我们如何使用 WebClient 在 Flux 中检索流数据？

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

在服务器端，代码是一个简单的控制器，暴露了 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 错误 99026 --- [ctor-http-nio-5] r.ipc.netty.channel.ChannelOperations：[HttpClient] 处理连接时出错。请求关闭频道

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.最终的]

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

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

4

1 回答 1

2

所有CommandLineRunnerbean 都在 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?
        }
    };
}
于 2017-02-28T09:20:29.523 回答