Spring Webflux流式返回大列表 消费者写入文件无法逐秒生效问题

• 服务端未指定流式响应的媒体类型：默认情况下Spring WebFlux返回Flux时如果没有明确设置流式媒体类型，会启用响应缓冲，攒到一定大小才批量发送数据，不会按delayElements的节奏逐次返回。
• 客户端文件写入未实时刷盘：FileWriter自带内存缓冲区，你当前代码仅在所有数据接收完成的doOnComplete阶段才执行flush，之前写入的内容都存在JVM缓冲区中，不会实时落盘，所以观察不到逐行写入的效果。
• 可选影响因素：部分版本的WebClient默认存在小的响应数据缓冲区，也可能导致少量数据攒包后才向下游发射。

1 服务端接口调整 #

给接口增加流式媒体类型声明，强制服务端每产生一条数据就立刻推送，不做缓冲，推荐使用TEXT_EVENT_STREAM（Server-Sent Events标准）或者APPLICATION_NDJSON（换行分隔JSON）：

import org.springframework.http.MediaType;

@GetMapping(value = "/path", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<Something> streamAPI(){
    List<Something> object = // 1000+ Records
    return Flux.fromIterable(object).delayElements(Duration.ofSeconds(1));
}

2 客户端代码调整 #

在每次写入文件后立即执行flush操作，把缓冲区的内容立刻刷入磁盘，同时补充异常场景的文件流关闭逻辑避免资源泄漏：

FileWriter jsonFileWriter = new FileWriter("example.json");
return webclient.get().uri(builder -> builder.scheme("http").host("localhost").port(8888)
                .path("/path").build())
        .retrieve()
        .onStatus(HttpStatus::is4xxClientError, response -> {
            throw new NotFoundException("Not Found");
        })
        .onStatus(HttpStatus::is5xxServerError, response -> {
            throw new InternalServerError("Internal Server Error");
        })
        .bodyToFlux(Something.class)
        .doOnNext(data -> {
            try {
                jsonFileWriter.write(data.toString() + "\n");
                // 每次写入后立即刷盘
                jsonFileWriter.flush();
            } catch (IOException e) {
                e.printStackTrace();
            }
        })
        .doOnComplete(() -> {
            try {
                jsonFileWriter.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        })
        .doOnError(e -> {
            try {
                jsonFileWriter.close();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        });

可选优化（如果上述调整仍有攒包） #

如果调整后仍存在少量攒包，可在创建WebClient时调小响应缓冲区阈值：

WebClient webClient = WebClient.builder()
        .codecs(configurer -> configurer.defaultCodecs()
                // 调小最大缓冲区，避免攒包
                .maxInMemorySize(1024))
        .build();

内容的提问来源于stack exchange，提问作者developer