Spring Boot中Flux.fromStream响应被缓冲，如何开启实时流？

当使用Flux.fromStream生成响应返回给客户端（包括终端curl请求）时，会批量发送所有消息，而非逐条推送；但单独运行该Flux时，却能每隔3秒打印一条消息，符合预期。而用Flux.interval实现的流，无论单独运行还是作为接口响应，都能实时逐条推送消息。

1. 单独运行Flux.fromStream（符合预期） #

Flux.fromStream(() -> {
    System.out.println("Started");
    return Stream.iterate(0, i -> i + 2).limit(3)
            .map( id -> {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException ex) {
                    return ResponseDTO.builder().message("Sleep failed").build();
                }
                return ResponseDTO.builder().message("Time is "+ id).build();
            });
}).subscribe(value -> System.out.println(value));

2. 作为REST接口返回（批量发送，不符合预期） #

return ResponseEntity.ok()
        .cacheControl(CacheControl.noCache())
        .contentType(MediaType.TEXT_EVENT_STREAM)
        .headers(httpHeaders -> httpHeaders.add("X-Accel-Buffering", "no"))
        .body(Flux.fromStream(() -> {
            System.out.println("Started");
            return Stream.iterate(0, i -> i + 2).limit(3)
                    .map( id -> {
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException ex) {
                            return ResponseDTO.builder().message("Sleep failed").build();
                        }
                        return ResponseDTO.builder().message("Time is "+ id).build();
                    });
        }));

3. Flux.interval实现（实时推送，符合预期） #

单独运行 #

Flux.interval(Duration.ofSeconds(3))
                        .map(id-> ResponseDTO.builder().message("Time is "+ id).build()).subscribe(value -> System.out.println(value));

作为REST接口返回 #

return ResponseEntity.ok()
        .cacheControl(CacheControl.noCache())
        .contentType(MediaType.TEXT_EVENT_STREAM)
        .headers(httpHeaders -> httpHeaders.add("X-Accel-Buffering", "no"))
        .body(Flux.interval(Duration.ofSeconds(3))
                        .map(id-> ResponseDTO.builder().message("Time is "+ id).build()));

解决方案 #

问题核心在于Stream.iterate的同步阻塞特性与Reactor调度逻辑的冲突：

• Stream.iterate是同步生成元素的，你在map中使用Thread.sleep会导致整个Stream在当前线程阻塞，直到3个元素全部生成完毕。单独运行时subscribe是同步消费，所以能逐个输出；但Spring WebFlux场景下，Stream的处理会被放到框架线程池，同步阻塞导致所有元素一次性生成后才被发送到响应流。
• Flux.interval是异步生成元素的，它会按指定间隔在调度器上逐个发射元素，天然适配实时流场景。

要修复Flux.fromStream的问题，需让元素生成过程异步化，确保逐个发射：

修复方案1：使用调度器隔离阻塞操作 #

return ResponseEntity.ok()
        .cacheControl(CacheControl.noCache())
        .contentType(MediaType.TEXT_EVENT_STREAM)
        .headers(httpHeaders -> httpHeaders.add("X-Accel-Buffering", "no"))
        .body(Flux.fromStream(() -> Stream.iterate(0, i -> i + 2).limit(3))
                .publishOn(Schedulers.boundedElastic()) // 切换到弹性调度器处理阻塞逻辑
                .map(id -> {
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException ex) {
                        return ResponseDTO.builder().message("Sleep failed").build();
                    }
                    return ResponseDTO.builder().message("Time is " + id).build();
                }));

修复方案2：改用Reactor异步延迟API（更推荐） #

避免直接使用Thread.sleep，用Reactor原生的异步延迟替代：

return ResponseEntity.ok()
        .cacheControl(CacheControl.noCache())
        .contentType(MediaType.TEXT_EVENT_STREAM)
        .headers(httpHeaders -> httpHeaders.add("X-Accel-Buffering", "no"))
        .body(Flux.range(0, 3)
                .map(i -> i * 2)
                .delayElements(Duration.ofSeconds(3)) // 异步延迟生成元素
                .map(id -> ResponseDTO.builder().message("Time is " + id).build()));

修改后，每个元素会间隔3秒被逐个发射，客户端就能收到实时的流推送。

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