Flux在TEXT_EVENT_STREAM_VALUE端点批量返回，如何实现实时流式输出？

1. 修复阻塞操作（核心原因） #

如果你的calculateStateById方法包含同步阻塞逻辑（比如JDBC查库、同步文件读取、第三方同步接口调用），会导致整个响应式流被阻塞，所有元素必须等全部计算完成后才能批量返回。

将阻塞逻辑包装到Mono.fromCallable中，并通过subscribeOn指定专门的阻塞操作线程池，避免占用WebFlux的非阻塞主线程：

private Mono<State> calculateStateById(Long id) {
    return Mono.fromCallable(() -> {
        // 这里放置原来的同步阻塞代码，比如：
        return stateDao.queryStateById(id);
    })
    // 使用boundedElastic线程池处理阻塞任务，不影响响应式流的非阻塞执行
    .subscribeOn(Schedulers.boundedElastic());
}

2. 调整flatMap的并发控制（可选优化） #

默认flatMap的并发订阅数是256，如果任务处理速度过快，可能会出现“批量返回”的视觉效果。你可以根据业务需求调整并发数，或者换成串行处理的concatMap：

• 自定义并发数：

public Flux<State> getStatesByIds(List<Long> ids) {
    return Flux.fromIterable(ids)
            // 比如设置并发数为10，可根据实际场景调整
            .flatMap(id -> calculateStateById(id), 10);
}

• 严格串行处理（每个元素计算完成后立即返回，再处理下一个）：

public Flux<State> getStatesByIds(List<Long> ids) {
    return Flux.fromIterable(ids)
            .concatMap(id -> calculateStateById(id));
}

3. 禁用客户端缓冲（前端侧问题） #

如果服务端日志显示每个State都已及时生成，但前端仍批量接收数据，可能是客户端（如浏览器）的默认缓冲导致的。可以在Controller中添加响应头禁用缓存：

@GetMapping(value = "/stream/{ids}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<State> getConfiguredStatesStream(@PathVariable List<Long> ids, ServerHttpResponse response) {
    response.getHeaders().setCacheControl("no-cache");
    return stateService.getStatesByIds(ids);
}

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