本节将提供基于 Spring WebFlux 和 SSE 实现类ChatGPT流式回复效果的完整代码示例，并详细说明所需的依赖和配置。
1. 项目配置
- 构建工具: Maven 或 Gradle
- 依赖:
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
2. 后端代码 (Spring WebFlux)
package com.example.ssedemo;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import org.springframework.http.codec.ServerSentEvent;import java.time.Duration;@SpringBootApplication
@RestController
public class SseDemoApplication {public static void main(String[] args) {SpringApplication.run(SseDemoApplication.class, args);}@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Flux<ServerSentEvent<String>> streamChatGPTReply(@RequestParam String message) {// 模拟调用 ChatGPT API 获取回复String reply = "收到消息: " + message + ". 正在思考...";return Flux.<String>create(sink -> {sink.next(reply); // 先发送初始回复// 模拟逐字生成回复for (int i = 0; i < reply.length(); i++) {try {Thread.sleep(100); // 模拟延迟sink.next(reply.substring(0, i + 1));} catch (InterruptedException e) {throw new RuntimeException(e);}}sink.complete();}).map(data -> ServerSentEvent.<String>builder().data(data).build()).delayElements(Duration.ofMillis(100)); // 每隔一段时间发送一个字符}
}
另外一种方式：
private final WebClient webClient; // 用于调用外部 APIpublic ChatController(WebClient.Builder webClientBuilder) {this.webClient = webClientBuilder.baseUrl("http://api.example.com").build();
}@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> streamChatGPTReply(@RequestParam String message) {// 使用 WebClient 异步调用外部 APIreturn webClient.post().uri("/api/external") .bodyValue(message).retrieve().bodyToFlux(String.class) // 假设 API 返回 String 类型数据.map(data -> ServerSentEvent.<String>builder().data(data) // 将 API 响应数据包装到 SSE 事件中.build()).delayElements(Duration.ofMillis(100));
}
3. 前端代码 (HTML & JavaScript)
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>SSE Demo</title>
</head>
<body><h1>SSE Chat Demo</h1><div id="output"></div><input type="text" id="message"><button onclick="sendMessage()">发送</button><script>const output = document.getElementById('output');function sendMessage() {const message = document.getElementById('message').value;const eventSource = new EventSource('/stream?message=' + message);eventSource.onmessage = (event) => {output.innerHTML += event.data + '<br>';};eventSource.onerror = (error) => {console.error('SSE 连接错误:', error);eventSource.close();};}</script>
</body>
</html>
4. 代码解析
- 后端:
@GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE): 指定响应类型为text/event-stream，这是 SSE 的标准 MIME 类型。Flux<ServerSentEvent<String>>: 使用 Spring WebFlux 的Flux类型返回数据流，并使用ServerSentEvent包装每个数据项.sink.next(): 向数据流中发送数据。sink.complete(): 通知数据流结束。
- 前端:
new EventSource('/stream'): 创建 EventSource 对象，连接到后端 SSE 接口.eventSource.onmessage: 监听message事件，接收后端推送的数据.eventSource.onerror: 监听连接错误.
5. 运行 & 测试
- 启动 Spring Boot 应用.
- 访问
http://localhost:8080(默认端口). - 在输入框中输入消息并点击发送，观察逐字显示的效果.
总结
本文详细介绍了如何使用 Spring WebFlux 和 SSE 实现类似 ChatGPT 的流式回复效果，并提供了完整的代码示例。希望读者能够通过本文掌握该技术，并在实际项目中灵活运用。