时间戳

首页 > 解决方案 > 从 WebClient 流式传输到 Flux。阻塞超时抛出异常

spring-boot - 从 WebClient 流式传输到 Flux。阻塞超时抛出异常

问题描述

我在 Spring Boot 应用程序中使用 WebClient 调用流式 API。

我想检索元素，直到我收到 10 个元素或 10 秒过去。我希望请求被阻止，直到任何一个先发生。

        WebClient client = WebClient.builder().baseUrl(URL).build();

        List<Item> items = client
                .get()
                .retrieve()
                .bodyToFlux(Item.class)
                .limitRequest(10)
                .collectList()
                .block(Duration.ofSeconds(10));

如果在超时之前检索到 10 个项目，调用会很好地返回，并且我有一个包含 10 个项目的填充列表。

但是，如果超时首先通过，则会引发以下异常，并且不会返回任何项目。

java.lang.IllegalStateException: Timeout on blocking read for 10000 MILLISECONDS

如何读取长达 x 秒的流，然后使用 WebClient 返回检索到的项目？

标签： spring-bootreactive-programmingspring-webfluxproject-reactorspring-reactive

解决方案

我想检索元素，直到我收到 10 个元素或 10 秒过去。

听起来bufferTimeout()正是你所追求的。

每次缓冲区达到最大大小或 maxTime Duration 过去时，将传入的值收集到多个 List 缓冲区中，这些缓冲区将由返回的 Flux 发出。

您只需要这些缓冲区之一。在反应式上下文中，您只需调用next()生成的通量 - 因为您只想阻止，您可以调用blockFirst().

就像是：

List<Item> items = client
        .get()
        .retrieve()
        .bodyToFlux(Item.class)
        .bufferTimeout(10, Duration.ofSeconds(10)) //first parameter is max number of elements, second is timeout
        .blockFirst();

推荐阅读