Spring Boot控制器能否按线程完成顺序返回CompletableFuture结果？

直接返回CompletableFuture做不到每个线程完成就向客户端返回结果——因为HTTP是请求-响应模型，一次请求对应一次完整的响应，CompletableFuture会等所有异步任务完成后才把汇总结果打包成响应返回。

要实现“每个线程完成就推送结果”的需求，你需要用**服务器发送事件（SSE）**或者WebSocket，其中SSE更适配这种单向、分批推送结果的场景，实现起来也更轻量。

用SSE实现分批推送的思路 #

1. 控制器方法返回SseEmitter对象，它是Spring提供的SSE支持类，用于向客户端流式发送数据。
2. 接收到请求参数后，启动多个异步任务（可以用CompletableFuture或自定义线程池）。
3. 每个任务完成时，调用SseEmitter.send()将该任务的结果推送给客户端。
4. 所有任务执行完毕后，调用SseEmitter.complete()结束连接；如果任务出错，调用completeWithError()返回异常信息。

代码示例 #

@RestController
@RequestMapping("/text")
public class TextGeneratorController {

    // 自定义线程池，避免使用默认ForkJoinPool影响应用其他逻辑
    @Bean
    public ThreadPoolTaskExecutor asyncTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("text-generator-");
        executor.initialize();
        return executor;
    }

    @Autowired
    private ThreadPoolTaskExecutor asyncTaskExecutor;

    @GetMapping("/generate")
    public SseEmitter generateText(@RequestParam String keyword) {
        // 设置超时时间，比如5分钟，避免任务未完成就断开连接
        SseEmitter emitter = new SseEmitter(300000L);

        List<CompletableFuture<Void>> taskList = new ArrayList<>();

        // 快任务：1秒完成
        taskList.add(CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
                List<String> fastResult = List.of(keyword + "_快速结果1", keyword + "_快速结果2");
                emitter.send(fastResult);
            } catch (InterruptedException | IOException e) {
                emitter.completeWithError(e);
            }
        }, asyncTaskExecutor));

        // 慢任务：30秒完成
        taskList.add(CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(30000);
                List<String> slowResult = List.of(keyword + "_慢速结果1", keyword + "_慢速结果2");
                emitter.send(slowResult);
            } catch (InterruptedException | IOException e) {
                emitter.completeWithError(e);
            }
        }, asyncTaskExecutor));

        // 所有任务完成后关闭连接
        CompletableFuture.allOf(taskList.toArray(new CompletableFuture[0])).whenComplete((unused, throwable) -> {
            if (throwable != null) {
                emitter.completeWithError(throwable);
            } else {
                emitter.complete();
            }
        });

        return emitter;
    }
}

客户端接收示例（JavaScript） #

客户端可以用EventSource来监听SSE推送的结果：

const eventSource = new EventSource('/text/generate?keyword=test');

eventSource.onmessage = function(event) {
    const result = JSON.parse(event.data);
    console.log('收到新结果:', result);
    // 在这里处理每一批返回的结果
};

eventSource.onerror = function(error) {
    console.error('SSE连接出错:', error);
    eventSource.close();
};

注意事项 #

• SSE是单向通信，仅支持服务器向客户端推送数据，如果需要双向交互可以考虑WebSocket。
• 务必自定义线程池处理异步任务，避免使用默认的ForkJoinPool，防止任务阻塞影响应用其他功能。
• 根据任务耗时合理设置SseEmitter的超时时间，避免连接提前断开。

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