JAVA 后端如何支持前端流式渲染 AI 回复？SSE 与分片推送实践

JAVA 后端支持前端流式渲染 AI 回复：SSE 与分片推送实践

大家好，今天我们来聊聊如何在 Java 后端支持前端流式渲染 AI 回复。这涉及到两个关键技术：Server-Sent Events (SSE) 和分片推送。我们将深入探讨它们的原理、实现方式以及最佳实践，并提供详细的代码示例。

1. 流式渲染的需求和挑战

传统的 AI 回复通常是后端一次性生成完整的结果，然后返回给前端。这种方式存在以下问题：

• 延迟高: 用户需要等待整个回复生成完毕才能看到内容。
• 用户体验差: 尤其对于长篇回复，用户会感到明显的等待，缺乏互动感。
• 资源浪费: 后端可能需要长时间占用资源生成完整的回复，即使前端用户已经不再关注。

流式渲染则可以有效解决这些问题。它允许后端将 AI 回复分解成多个片段，并逐步推送给前端，前端实时渲染这些片段，从而实现“边生成边显示”的效果。这可以显著降低延迟，提升用户体验，并更有效地利用后端资源。

2. Server-Sent Events (SSE) 简介

Server-Sent Events (SSE) 是一种服务器推送技术，它允许服务器向客户端单向推送数据。与 WebSocket 相比，SSE 更加轻量级，只需要一个 HTTP 连接，并且天然支持断线重连。非常适合用于实时数据更新，例如股票行情、新闻推送以及我们这里的 AI 回复流式渲染。

SSE 的特点：

• 基于 HTTP 协议: 易于部署和使用，不需要额外的协议支持。
• 单向通信: 只能从服务器向客户端推送数据。
• 文本格式: 通常使用文本格式传输数据，例如 JSON。
• 自动重连: 客户端会自动尝试重连断开的连接。

SSE 的数据格式：

SSE 数据由一系列事件组成，每个事件包含以下字段：

字段	描述
event	事件类型，可选。
data	事件数据，必须。
id	事件 ID，可选，用于跟踪事件顺序。
retry	重连延迟时间，可选，单位为毫秒。

每个字段以 字段名: 值n 的形式表示，事件之间用空行分隔。

示例：

data: This is the first chunk.nn
data: This is the second chunk.nn
data: This is the third chunk.nn

3. Java 后端 SSE 实现

在 Java 后端，我们可以使用 Spring Framework 提供的 SseEmitter 类来实现 SSE。 SseEmitter 允许我们将数据异步地发送到客户端。

3.1 添加依赖

首先，确保你的 Spring Boot 项目中包含了 Web 依赖。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

3.2 创建 Controller

创建一个 Controller 来处理 SSE 请求。

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 org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
public class AIController {

    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    @GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamAIResponse(@RequestParam("prompt") String prompt) {
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE); // 设置超时时间，可以根据实际情况调整

        executor.execute(() -> {
            try {
                // 模拟 AI 回复生成过程
                String aiResponse = generateAIResponse(prompt);
                String[] chunks = aiResponse.split("(?<=\G.{" + 50 + "})"); // 将回复分割成 50 个字符的片段

                for (String chunk : chunks) {
                    emitter.send(SseEmitter.event().data(chunk));
                    Thread.sleep(100); // 模拟生成延迟
                }

                emitter.send(SseEmitter.event().data("[DONE]")); // 发送结束标记
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }

    private String generateAIResponse(String prompt) {
        // 这里应该调用真正的 AI 模型进行推理
        // 为了演示，我们生成一个随机字符串
        String response = "AI Response: ";
        for (int i = 0; i < 200; i++) {
            response += (char) (Math.random() * 26 + 'a');
        }
        return response;
    }
}

代码解释：

• @GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE): 指定请求路径和响应类型为 text/event-stream，这是 SSE 的标准 MIME 类型。
• SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);: 创建一个 SseEmitter 实例，设置超时时间为 Long.MAX_VALUE，这意味着连接将一直保持，直到服务器主动关闭或发生错误。 可以根据实际场景调整超时时间，防止资源浪费。
• executor.execute(() -> { ... });: 使用线程池异步执行 AI 回复生成和推送过程，防止阻塞主线程。
• generateAIResponse(prompt): 模拟 AI 回复生成过程。 注意： 在实际应用中，这里应该调用真正的 AI 模型进行推理。
• String[] chunks = aiResponse.split("(?<=\G.{" + 50 + "})");: 将 AI 回复分割成多个片段。 这里使用了正则表达式 (?<=\G.{" + 50 + "}) 将字符串分割成每 50 个字符的片段。
• emitter.send(SseEmitter.event().data(chunk));: 使用 SseEmitter 将每个片段发送到客户端。 SseEmitter.event().data(chunk) 构建一个包含数据的 SSE 事件。
• Thread.sleep(100);: 模拟生成延迟。
• emitter.send(SseEmitter.event().data("[DONE]"));: 发送结束标记，通知客户端回复已完成。
• emitter.complete();: 完成 SSE 连接。
• emitter.completeWithError(e);: 如果发生错误，完成 SSE 连接并发送错误信息。

3.3 配置

如果你的应用使用了 Spring Security，需要配置允许访问 /ai/stream 接口。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
import org.springframework.security.web.SecurityFilterChain;

@Configuration
@EnableWebSecurity
public class SecurityConfig {

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        http
                .csrf().disable()
                .authorizeHttpRequests()
                .requestMatchers("/ai/stream").permitAll() // 允许匿名访问 /ai/stream
                .anyRequest().authenticated();

        return http.build();
    }
}

4. 前端 SSE 实现

在前端，我们可以使用 EventSource API 来建立 SSE 连接并接收服务器推送的数据。

4.1 HTML

首先，在 HTML 文件中创建一个用于显示 AI 回复的元素。

<!DOCTYPE html>
<html>
<head>
    <title>AI Stream</title>
</head>
<body>
    <h1>AI Stream</h1>
    <div id="ai-response"></div>
    <script src="script.js"></script>
</body>
</html>

4.2 JavaScript

然后，编写 JavaScript 代码来建立 SSE 连接并处理接收到的数据。

const aiResponseDiv = document.getElementById('ai-response');
const eventSource = new EventSource('/ai/stream?prompt=Hello');

eventSource.onmessage = (event) => {
    if (event.data === "[DONE]") {
        eventSource.close();
        console.log("Stream completed.");
    } else {
        aiResponseDiv.innerHTML += event.data;
    }
};

eventSource.onerror = (error) => {
    console.error("EventSource failed:", error);
    eventSource.close();
};

代码解释：

• const eventSource = new EventSource('/ai/stream?prompt=Hello');: 创建一个 EventSource 实例，连接到服务器的 SSE 接口。 prompt=Hello 是一个示例查询参数，用于传递用户输入。
• eventSource.onmessage = (event) => { ... };: 监听 message 事件，当服务器推送数据时，会触发该事件。
• if (event.data === "[DONE]") { ... }: 检查接收到的数据是否为结束标记 [DONE]。 如果是，则关闭 SSE 连接。
• aiResponseDiv.innerHTML += event.data;: 将接收到的数据追加到 ai-response 元素中，实现实时渲染。
• eventSource.onerror = (error) => { ... };: 监听 error 事件，当发生错误时，会触发该事件。

5. 分片推送的优化

虽然 SSE 已经可以实现流式渲染，但我们还可以通过一些优化来进一步提升性能和用户体验。

5.1 数据压缩

对于较大的 AI 回复，可以使用 Gzip 等压缩算法对数据进行压缩，减少网络传输量。

后端 (Java)：

可以使用 Spring Framework 提供的 GzipCompressingResponseBodyAdvice 来自动对响应进行 Gzip 压缩。

首先，添加以下依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

然后，创建一个 ResponseBodyAdvice 来对 SSE 响应进行压缩：

import org.springframework.core.MethodParameter;
import org.springframework.http.MediaType;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyAdvice;

@ControllerAdvice
public class GzipResponseBodyAdvice implements ResponseBodyAdvice<Object> {

    @Override
    public boolean supports(MethodParameter returnType, Class<? extends HttpMessageConverter<?>> converterType) {
        // 只对 text/event-stream 类型的响应进行压缩
        return returnType.getMethod().getReturnType().equals(SseEmitter.class);
    }

    @Override
    public Object beforeBodyWrite(Object body, MethodParameter returnType, MediaType selectedContentType,
                                  Class<? extends HttpMessageConverter<?>> selectedConverterType,
                                  ServerHttpRequest request, ServerHttpResponse response) {
        // 启用 Gzip 压缩
        response.getHeaders().set("Content-Encoding", "gzip");
        return body;
    }
}

注意： 前端需要支持 Gzip 解压缩。 大多数浏览器都支持自动解压缩，无需额外配置。

5.2 心跳机制

由于网络不稳定等原因，SSE 连接可能会断开。为了保持连接的活跃性，可以引入心跳机制。

后端 (Java)：

定期向客户端发送心跳事件。

@RestController
public class AIController {

    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    @GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamAIResponse(@RequestParam("prompt") String prompt) {
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

        executor.execute(() -> {
            try {
                // 发送心跳
                sendHeartbeat(emitter);

                // 模拟 AI 回复生成过程
                String aiResponse = generateAIResponse(prompt);
                String[] chunks = aiResponse.split("(?<=\G.{" + 50 + "})");

                for (String chunk : chunks) {
                    emitter.send(SseEmitter.event().data(chunk));
                    Thread.sleep(100);
                }

                emitter.send(SseEmitter.event().data("[DONE]"));
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }

    private void sendHeartbeat(SseEmitter emitter) {
        new Thread(() -> {
            try {
                while (true) {
                    emitter.send(SseEmitter.event().comment("heartbeat")); // 发送注释作为心跳
                    Thread.sleep(30000); // 每 30 秒发送一次心跳
                }
            } catch (IOException | InterruptedException e) {
                // 连接已关闭，停止发送心跳
            }
        }).start();
    }

    private String generateAIResponse(String prompt) {
        // ... (省略)
    }
}

前端 (JavaScript)：

前端无需特殊处理心跳事件，因为 EventSource 会自动忽略注释。

5.3 重试机制

如果 SSE 连接断开，客户端会自动尝试重连。 可以通过设置 retry 字段来控制重连延迟时间。

后端 (Java)：

emitter.send(SseEmitter.event().retry(5000).data(chunk)); // 设置重连延迟时间为 5 秒

5.4 错误处理

在后端和前端都需要进行适当的错误处理，以便在发生错误时能够及时通知用户并进行相应的处理。

6. 分片推送的其他方案

除了 SSE，还有其他一些技术可以用于实现分片推送，例如：

• WebSocket: WebSocket 是一种双向通信协议，可以实现服务器主动推送数据到客户端。 与 SSE 相比，WebSocket 更加灵活，但需要维护一个持久的连接，并且需要额外的协议支持。
• HTTP Long Polling: HTTP Long Polling 是一种模拟服务器推送的技术。 客户端向服务器发送一个请求，服务器保持连接，直到有新的数据可用时才返回响应。 HTTP Long Polling 实现简单，但会消耗服务器资源，并且可能存在延迟。

方案对比：

技术	优点	缺点	适用场景
SSE	轻量级，易于部署，天然支持断线重连，基于 HTTP 协议	单向通信	实时数据更新，例如股票行情、新闻推送、AI 回复流式渲染
WebSocket	双向通信，灵活	需要维护持久连接，需要额外协议支持	实时性要求高，需要双向通信的场景，例如在线聊天、多人游戏
HTTP Long Polling	实现简单	消耗服务器资源，可能存在延迟	实时性要求不高，对服务器资源消耗不敏感的场景

7. 总结：流式渲染AI回复的关键技术

我们讨论了如何使用 Server-Sent Events (SSE) 和分片推送技术在 Java 后端支持前端流式渲染 AI 回复。通过这些技术，可以显著降低延迟，提升用户体验，并更有效地利用后端资源。

8. 下一步：优化和扩展

未来，我们可以进一步优化和扩展这些技术，例如：

• 更智能的分片策略: 根据 AI 回复的内容动态调整分片大小，以获得更好的用户体验。
• 更强大的错误处理机制: 提供更详细的错误信息，并支持自动重试。
• 集成到更复杂的 AI 应用中: 将这些技术应用到更复杂的 AI 应用中，例如智能客服、聊天机器人等。