⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处 blog.csdn.net/lssjzmn/article/details/151361288 「lssjzmn」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

引言

在构建高性能 Web 应用时，异步响应机制是提升系统吞吐量和用户体验的关键。本文将深入探讨 Spring Web 中两种常用的异步响应方式：CompletableFuture 和 ResponseBodyEmitter，并结合前后端交互、Nginx 配置、连接管理优化等方面，提供一套完整的实战指南。

一、使用 CompletableFuture 实现异步接口

控制器代码：

@RestController
public class AsyncController {
 
    @Autowired
    private AsyncService asyncService;
 
    @GetMapping("/async")
    public CompletableFuture<String> async() {
        return asyncService.doAsyncTask();
    }
}

服务层代码：

@Service
public class AsyncService {
 
    @Async
    public CompletableFuture<String> doAsyncTask() {
        try {
            Thread.sleep(3000); // 模拟耗时任务
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("任务完成！");
    }
}

配置类启用异步支持：

@Configuration
@EnableAsync
public class AsyncConfig {
    @Bean
    public Executor taskExecutor() {
        return Executors.newCachedThreadPool();
    }
}

解释：

@Async： 表示该方法将在独立线程中异步执行。
CompletableFuture： 表示异步任务的结果容器，Spring 会挂起请求直到任务完成。
@EnableAsync： 启用 Spring 的异步方法支持。
taskExecutor()： 定义线程池，避免每次都创建新线程。

Spring MVC 中 CompletableFuture 的响应行为详解

服务端处理流程：

1.客户端发起请求：

请求到达 Spring MVC 的 DispatcherServlet。

2.控制器返回 CompletableFuture：

Spring 检测到返回类型是 CompletableFuture，会将请求挂起（非阻塞），等待异步任务完成。

3.异步任务执行中：

通常由线程池（如 @Async 配置的 TaskExecutor）执行任务。

4.任务完成后：

Spring 会自动将 CompletableFuture 的结果作为响应体序列化为 JSON 或文本，并发送给客户端。

5.连接关闭：

响应发送完毕后，HTTP 连接关闭。

客户端体验：

虽然服务端是异步处理，但客户端并不需要特殊处理。它只会感受到响应时间稍长，但最终仍是一个标准的 HTTP 响应。

示例：浏览器或前端框架中的行为

fetch('/async')
  .then(response => response.text()) // 或 .json() 取决于返回类型
  .then(data => {
    console.log("收到响应:", data);
  });

无需轮询： 客户端只需等待响应即可。
无需特殊协议： 不像 WebSocket 或 SSE，不需要建立特殊连接。
适合一次性结果： 比如异步计算、远程调用、数据库查询等。

使用 CompletableFuture 的好处

1.提升系统吞吐量与并发能力

接口返回 CompletableFuture 后，Spring 会将请求挂起，不占用当前线程。
后端任务可以在独立线程池中异步执行，释放 Web 容器线程资源。
更适合高并发场景，如 API 网关、微服务聚合接口。

2.非阻塞式编程模型

避免传统同步调用中的线程阻塞。
支持链式调用（如 .thenApply(), .thenCompose()），更易于构建复杂异步逻辑。

3.更好的用户体验

虽然服务端异步处理，但客户端仍然是标准 HTTP 请求，使用体验一致。
响应时间更短，尤其在多个子任务并发执行时。

4.与 @Async 配合使用，解耦业务逻辑

可以将耗时任务封装在 @Async 的 Service 层中，控制器层保持简洁。
易于测试和维护，逻辑清晰。

5.支持异常处理与超时控制

return asyncService.doAsyncTask()
    .orTimeout(5, TimeUnit.SECONDS)
    .exceptionally(ex -> "任务失败：" + ex.getMessage());

可以优雅地处理任务失败、超时等情况，提升系统稳定性。

6.适用场景建议

场景	是否推荐使用 `CompletableFuture`
微服务聚合调用	✅ 非常适合
异步数据库或远程调用	✅ 推荐
文件上传/下载	❌ 不推荐（使用 `StreamingResponseBody` 更合适）
实时推送	❌ 不推荐（使用 `SseEmitter` 或 `WebSocket` 更合适）
任务进度反馈	✅ 可用，但 `ResponseBodyEmitter` 更灵活

底层机制补充（Spring 5+）

Spring 5 引入了对响应式编程的支持（如 Mono 和 Flux），但 CompletableFuture 仍然是非常常用的异步返回类型，尤其在非响应式项目中。

Spring 会通过 DeferredResult 或 WebAsyncTask 等机制将 CompletableFuture 包装成异步响应对象，挂起请求直到任务完成。

二、使用 ResponseBodyEmitter 实现流式推送

ResponseBodyEmitter 是 Spring MVC 提供的一种异步响应机制，允许服务端在一个 HTTP 请求中分段发送数据，而不是一次性返回完整响应。它适用于需要长连接或实时推送的场景。

控制器代码：

@RestController
public class StreamController {
 
    @Autowired
    private EmitterManager emitterManager;
 
    @GetMapping("/stream/{id}")
    public ResponseBodyEmitter stream(@PathVariable String id) {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter();
        emitterManager.register(id, emitter);

        Executors.newSingleThreadExecutor().submit(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    if (emitterManager.shouldStop(id)) {
                        break;
                    }
                    emitter.send("进度：" + i * 10 + "%\n", MediaType.TEXT_PLAIN);
                    Thread.sleep(1000);
                }
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });
 
        return emitter;
    }
 
    @PostMapping("/stop/{id}")
    public ResponseEntity<Void> stop(@PathVariable String id) {
        emitterManager.stop(id);
        return ResponseEntity.ok().build();
    }
}

解释：

ResponseBodyEmitter： 允许服务端在一个请求中多次发送数据。
emitter.send()： 每次发送一段数据，客户端可实时接收。
emitter.complete()： 表示数据发送完毕，关闭连接。
emitter.completeWithError()： 发送异常时关闭连接。
EmitterManager： 用于管理连接状态，支持外部停止。

注意事项：

ResponseBodyEmitter 默认超时时间为 30 秒，可通过构造函数设置。
异步线程必须手动调用 complete() 或 completeWithError()。

核心特性：

非阻塞异步响应
支持多次写入数据
与 Servlet 3.0+ 异步处理兼容
可与 @ResponseBody 或 @RestController 一起使用

前端接收方式

使用 fetch + ReadableStream：

fetch('/progress/stream')
  .then(response => {
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
 
    function read() {
      reader.read().then(({ done, value }) => {
        if (done) return;
        console.log(decoder.decode(value));
        read();
      });
    }
 
    read();
  });

解释：

response.body.getReader()： 获取响应流的读取器。
TextDecoder： 将字节流解码为字符串。
read()： 递归读取数据直到连接关闭。

使用 EventSource（适用于 text/event-stream）：

const source = new EventSource('/sse/stream');
source.onmessage = function(event) {
  console.log("Received:", event.data);
};

Nginx 配置

location /stream {
    proxy_pass http://localhost:8080;
    proxy_buffering off;         # 禁用缓冲，确保实时推送
    proxy_read_timeout 3600s;    # 设置长连接超时时间
    chunked_transfer_encoding on;# 启用分块传输
}

解释：

proxy_buffering off： 防止 Nginx 缓冲响应，确保客户端实时接收。
proxy_read_timeout： 避免连接被 Nginx 提前关闭。
chunked_transfer_encoding on： 允许分块传输数据。

EmitterManager和 EmitterSession

@Component
public class EmitterManager {
 
    // 存储所有活跃的 emitter 会话
    private final Map<String, EmitterSession> sessions = new ConcurrentHashMap<>();
 
    /**
     * 注册一个新的 emitter 会话
     */
    public void register(String id, ResponseBodyEmitter emitter) {
        EmitterSession session = new EmitterSession(id, emitter);
        sessions.put(id, session);
 
        // 添加连接关闭或异常的监听器
        emitter.onCompletion(() -> cleanup(id));
        emitter.onTimeout(() -> cleanup(id));
        emitter.onError((Throwable t) -> cleanup(id));
    }
 
    /**
     * 外部触发停止推送
     */
    public void stop(String id) {
        EmitterSession session = sessions.get(id);
        if (session != null) {
            session.getStopFlag().set(true);
            session.getEmitter().complete();
            cleanup(id);
        }
    }
 
    /**
     * 判断是否应该停止发送
     */
    public boolean shouldStop(String id) {
        EmitterSession session = sessions.get(id);
        return session != null && session.getStopFlag().get();
    }
 
    /**
     * 获取 emitter 实例
     */
    public ResponseBodyEmitter getEmitter(String id) {
        EmitterSession session = sessions.get(id);
        return session != null ? session.getEmitter() : null;
    }
 
    /**
     * 清理资源
     */
    private void cleanup(String id) {
        sessions.remove(id);
    }
 
    /**
     * 获取当前活跃连接数
     */
    public int activeCount() {
        return sessions.size();
    }
}
public class EmitterSession {
    private final String id;
    private final ResponseBodyEmitter emitter;
    private final AtomicBoolean stopFlag;
    private final Instant connectedAt;
 
    public EmitterSession(String id, ResponseBodyEmitter emitter) {
        this.id = id;
        this.emitter = emitter;
        this.stopFlag = new AtomicBoolean(false);
        this.connectedAt = Instant.now();
    }
 
    public String getId() {
        return id;
    }
 
    public ResponseBodyEmitter getEmitter() {
        return emitter;
    }
 
    public AtomicBoolean getStopFlag() {
        return stopFlag;
    }
 
    public Instant getConnectedAt() {
        return connectedAt;
    }
}

三、使用场景对比：CompletableFuture vs ResponseBodyEmitter

维度	CompletableFuture	ResponseBodyEmitter
响应方式	一次性返回结果	多次推送数据
是否阻塞容器线程	否（异步执行）	否（异步推送）
客户端处理方式	普通 HTTP 请求	需处理流式数据（如 `ReadableStream`）
适合场景	异步计算、远程调用、微服务聚合	实时推送、任务进度反馈、日志流
服务端控制	自动完成	手动控制发送与结束
连接生命周期	请求完成即关闭	可长时间保持连接
资源管理复杂度	低	高（需管理连接状态）
扩展性	支持任务组合、异常处理	支持连接控制、分组推送
部署兼容性	与传统 HTTP 完全兼容	需配置 Nginx 支持流式传输

使用建议总结

场景	推荐方案
异步数据库查询	✅ CompletableFuture
微服务聚合调用	✅ CompletableFuture
实时任务进度反馈	✅ ResponseBodyEmitter
日志流推送	✅ ResponseBodyEmitter
文件下载	❌ 使用 StreamingResponseBody 更合适
聊天、协同编辑	❌ 使用 WebSocket 更合适

芋道源码 —— 纯源码解析博客

Spring Web 异步响应实战：CF → RBE 全链路指南

引言

一、使用 CompletableFuture 实现异步接口

控制器代码：

服务层代码：

配置类启用异步支持：

解释：

Spring MVC 中 CompletableFuture 的响应行为详解

服务端处理流程：

客户端体验：

使用 CompletableFuture 的好处

底层机制补充（Spring 5+）

二、使用 ResponseBodyEmitter 实现流式推送

控制器代码：

解释：

注意事项：

核心特性：

前端接收方式

Nginx 配置

EmitterManager和 EmitterSession

三、使用场景对比：CompletableFuture vs ResponseBodyEmitter