spring-webflux - 当通量从 Spring 网 Controller 返回时会发生什么？-6ren

spring-webflux - 当通量从 Spring 网 Controller 返回时会发生什么？

转载作者：行者123 更新时间：2023-12-02 04:25:05

我对响应式(Reactive) API 比较陌生，并且对当我们从 Web Controller 返回 Flux 时幕后发生的事情感到好奇。
根据 spring-web 文档

Reactive return values are handled as follows:

A single-value promise is adapted to, similar to using DeferredResult. Examples include Mono (Reactor) or Single (RxJava).

A multi-value stream with a streaming media type (such as application/stream+json or text/event-stream) is adapted to, similar to using ResponseBodyEmitter or SseEmitter. Examples include Flux (Reactor) or Observable (RxJava). Applications can also return Flux or Observable.

A multi-value stream with any other media type (such as application/json) is adapted to, similar to using DeferredResult<List<?>>.

我创建了两个 API，如下所示:

@GetMapping("/async-deferredresult")
public DeferredResult<List<String>> handleReqDefResult(Model model) {
    LOGGER.info("Received async-deferredresult request");
    DeferredResult<List<String>> output = new DeferredResult<>();

    ForkJoinPool.commonPool().submit(() -> {
        LOGGER.info("Processing in separate thread");
        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10000   ; i++) {
            list.add(String.valueOf(i));
        }
        output.setResult(list);
    });

    LOGGER.info("servlet thread freed");
    return output;
}


@GetMapping(value = "/async-flux",produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    LOGGER.info("Received async-deferredresult request");
    List<String> list = new ArrayList<>();
    list.stream();
    for (int i = 0; i < 10000   ; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list);
}

所以异常(exception)是两个 API 的行为应该与多值流(Flux)的行为相同，应该与返回 DeferredResult 的行为相似。但是在返回延迟结果的 API 中，整个列表在浏览器上一次性打印，而在 API 中，Flux 返回按顺序(一个接一个)打印的数字。
当我从 Controller 返回 Flux 时到底发生了什么？

最佳答案

当我们从服务端点返回 Flux 时，可能会发生很多事情。但是我假设您想知道当 Flux 观察到来自该端点客户端的事件流时发生了什么。
场景一 :通过添加 'application/json' 作为端点的内容类型，Spring 将与客户端通信以期待 JSON 正文。

@GetMapping(value = "/async-flux", produces = MediaType.APPLICATION_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    List<String> list = new ArrayList<>();
    for (int i = 0; i < 10000; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list);
}

客户端的输出将是一整套数字。一旦响应传递，连接将被关闭。即使您使用了 Flux 作为响应类型，您仍然受到 HTTP over TCP/IP 工作原理的约束。端点收到一个 HTTP 请求，执行逻辑并使用包含最终结果的 HTTP 响应进行响应。

结果，您看不到响应式(Reactive) api 的真正值(value)。
场景二 :通过添加 'application/stream+json' 作为端点的内容类型，Spring 开始将 Flux 流的结果事件视为单独的 JSON 项。当一个项目被序列化时，HTTP 响应缓冲区被刷新，从服务器到客户端的连接保持打开，直到事件序列完成。
为了让它工作，我们可以按如下方式稍微修改您的原始代码。

@GetMapping(value = "/async-flux",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<String> handleReqDefResult1(Model model) {
    List<String> list = new ArrayList<>();
    for (int i = 0; i < 10000   ; i++) {
        list.add(String.valueOf(i));
    }
    return Flux.fromIterable(list)
            // we have 1 sec delay to demonstrate the difference of behaviour. 
            .delayElements(Duration.ofSeconds(1));
}

这次我们可以看到响应式(Reactive) api 端点的真正值(value)，它能够在日期可用时向其客户端提供结果。