在完成之前开始处理来自服务器的 Flux 响应：这可能吗？

如何解决在完成之前开始处理来自服务器的 Flux 响应：这可能吗？

我有 2 个 Spring-Boot-Reactive 应用程序，一个服务器和一个客户端；客户端像这样调用服务器：

Flux<Thing> things = thingsApi.listThings(5);

我想把它作为一个列表供以后使用：

// "extractContent" operation takes 1.5s per "thing"
List<String> thingsContent = things.map(ThingConverter::extractContent)
        .collect(Collectors.toList())
        .block()

在服务器端，端点定义如下所示：

@Override
public Mono<ResponseEntity<Flux<Thing>>> listThings(
        @NotNull @Valid @RequestParam(value = "nbThings") Integer nbThings,ServerWebExchange exchange
) {
    // "getThings" operation takes 1.5s per "thing"
    Flux<Thing> things = thingsService.getThings(nbThings);
    return Mono.just(new ResponseEntity<>(things,HttpStatus.OK));
}

签名来自 Open-API 生成的代码（Spring-Boot 服务器，反应模式）。

---

我观察到的：客户端立即跳转到 things.map，但仅在服务器完成发送所有“事物”后才开始处理 Flux。

我想要的是：服务器应该在“事物”生成时发送，以便客户端可以在到达时开始处理它们 em>，有效地将处理时间减半。

有没有办法做到这一点？我在网上找到了很多关于服务器部分的教程，但没有一个带有 Java 客户端的教程。我听说过服务器发送的事件，但可以使用返回 Flux 的“经典”Open-API 端点定义来实现我的目标吗？

问题似乎太复杂，无法在问题正文中放置一个最小的可行示例；完整代码可供参考on Github。

编辑：合并提议的解决方案后将链接重定向到 main 分支

解决方法

我通过更改 2 个点来运行它：

• 首先：我已将 content 端点响应的 /things 类型更改为：

content:
  text/event-stream

不要忘记更改默认响应，否则客户端将期望类型为 application/json 并等待整个响应。

• 第二点：我已经将 ThingsService.getThings 的返回值改为 this.getThingsFromExistingStream（你注释掉的方法）

我将更改推送到您 Github 上的新分支 fix-flux-response，以便您可以直接测试它们。