← 返回上一页

第6章: 「流语」实时对话推送引擎

基于 Spring WebFlux + SSE 实现生产级实时对话流式推送

📖 本章目标：构建「流语」实时对话推送引擎

本章带你完成 阶段 2：「流语」实时对话推送引擎 项目： 基于 Spring WebFlux + Reactor 构建一个具备流式响应、打字机效果、性能优化、多渠道适配的实时对话推送引擎。

✅ 完成本项目你将掌握以下8大核心能力

  • 1、流式响应架构设计：基于 Spring WebFlux + Reactor 设计实时流式对话架构，掌握 Mono/Flux 异步流模型。
  • 2、实时打字机效果实现：构建支持 SSE 的流式对话服务，实现逐字输出、打字机效果、用户中断等交互特性。
  • 3、响应式编程基础：掌握 Reactor 的 Mono/Flux 异步流模型；使用 StepVerifier 编写响应式测试用例。
  • 4、性能与稳定性优化：实现背压处理防止内存溢出，配置超时取消避免长时间等待；建立 P95/P99 延迟监控。
  • 5、实时交互体验优化：采用流式输出技术减少回复等待时长，配置高频问题快捷回复模板。
  • 6、多渠道适配与监控：支持 Web 端、移动端 APP、小程序等多渠道接入；集成服务监控面板。
  • 7、数据复盘与质量优化：自动采集流式交互日志，分析高频用户中断场景；输出优化建议。
  • 8、Spring 生态集成与部署：Spring AI 与 Spring WebFlux、Spring Security、Spring Cloud Gateway 整合。

🎯 前置要求

  • 已完成第4章：项目二 · 给聊天机器人添加记忆
  • 已完成第5章：流式响应测试
  • 理解 Reactor Flux
  • 理解对话记忆原理

🗂️ 项目结构（本章相关）

本章是在第4章“记忆聊天”项目基础上新增“流式 SSE 接口”。建议你先把下面这些文件在 IDE 里对照一遍， 再开始看后面的代码实现。

text
src/
├── main/
│   ├── java/
│   │   └── ...
│   │       ├── controller/
│   │       │   └── MemoryChatController.java        # 本章新增：/api/memory-chat/stream（SSE）
│   │       ├── service/
│   │       │   └── ConversationService.java         # 会话历史管理（第4章已有）
│   │       └── dto/
│   │           ├── MemoryChatRequest.java           # 请求：sessionId + message（第4章已有）
│   │           └── MemoryChatResponse.java          # 响应 DTO（第4章已有）
│   └── resources/
│       └── static/
│           └── data/
│               ├── demo.html                        # 演示页面（可用于手工体验）
│               └── demo-script.js                   # 前端交互脚本（可扩展对 SSE 的支持）
└── test/
    └── java/
        └── ...                                      #（可选）为流式接口补充集成测试

✅ 本章改动点

  • 在 MemoryChatController 中新增流式接口，返回 text/event-stream
  • 复用 ConversationService 做会话历史拼接与保存
  • 沿用第4章的 DTO：MemoryChatRequest / MemoryChatResponse

🎯 理解 Server-Sent Events (SSE)

什么是 SSE？

💡 核心概念

  • 服务器主动推送数据到客户端
  • 基于 HTTP 协议
  • 单向通信（服务器 → 客户端）
  • 适合实时数据流（如 AI 生成文本）

SSE vs WebSocket

特性 SSE WebSocket
通信方向 单向（服务器→客户端） 双向
协议 基于 HTTP 需要升级协议
实现难度 简单 复杂
适用场景 聊天机器人、实时通知 在线游戏、协作编辑

💡 对于 AI 聊天场景，SSE 是更好的选择。

📝 Step 1：在 MemoryChatController 添加流式接口

1.1 为什么在 MemoryChatController 中实现？

✅ 设计决策

  • 统一管理：所有记忆聊天功能在一个控制器中
  • 代码复用：共享会话管理逻辑
  • 简化架构：避免创建多个控制器
  • 一致性：流式和非流式使用相同的会话管理

1.2 添加流式聊天方法

在 MemoryChatController.java 中添加流式接口：

java
/**
 * 流式聊天接口
 * 
 * POST /api/memory-chat/stream
 */
@PostMapping(value = "/stream", produces = "text/event-stream")
public Flux<String> streamChat(@RequestBody MemoryChatRequest request) {
    try {
        // 验证请求
        request.validate();
        
        log.info("收到流式聊天请求 [会话: {}]: {}", 
            request.sessionId(), request.message());
        
        // 添加用户消息到历史
        conversationService.addUserMessage(
            request.sessionId(), 
            request.message()
        );
        
        // 获取对话历史
        List<Message> history = conversationService.getHistory(request.sessionId());
        
        // 调用流式 API
        Flux<ChatResponse> stream = chatModel.stream(new Prompt(history));
        
        // 使用 AtomicReference 来收集完整回复（线程安全）
        AtomicReference<String> fullResponse = new AtomicReference<>("");
        
        return stream
                .map(response -> response.getResult().getOutput().getText())
                .filter(content -> content != null && !content.isEmpty())
                .doOnNext(content -> {
                    // 累积内容
                    fullResponse.updateAndGet(current -> current + content);
                    log.debug("收到流式数据: [{}]", content);
                })
                .doOnComplete(() -> {
                    // 保存 AI 回复到历史
                    String finalResponse = fullResponse.get();
                    log.info("准备保存 AI 回复: {} 字符", finalResponse.length());
                    conversationService.addAssistantMessage(
                        request.sessionId(), 
                        finalResponse
                    );
                    log.info("流式回复完成 [会话: {}, 轮次: {}]: {} 字符",
                        request.sessionId(),
                        conversationService.getMessageCount(request.sessionId()) / 2,
                        finalResponse.length()
                    );
                })
                .doOnError(error -> {
                    log.error("流式响应错误", error);
                });
                
    } catch (Exception e) {
        log.error("处理流式聊天请求时发生错误", e);
        return Flux.just("错误: " + e.getMessage());
    }
}

⚠️ 重要说明

  • produces = "text/event-stream" - 指定返回 SSE 格式
  • Flux<String> - 响应式流类型
  • .filter() - 过滤空内容，避免发送无效数据块
  • AtomicReference - 线程安全地收集完整响应
  • doOnComplete() - 流完成时保存到历史
  • 不需要手动添加 data: 前缀！Spring 会自动处理

1.3 SSE 格式处理

❌ 错误做法（会导致双重前缀）

java
// ❌ 不要手动添加 SSE 格式！
return stream
    .map(content -> "data: " + content + "\n\n");  // 错误！

✅ 正确做法

java
// ✅ Spring 自动处理 SSE 格式
return stream
    .map(response -> response.getResult().getOutput().getText())
    .filter(content -> content != null && !content.isEmpty());

为什么？ Spring 的 produces = "text/event-stream" 会自动为每个元素添加 data: 前缀。如果手动添加，会变成 data:data: 内容（双重前缀），导致前端解析错误。

🧪 Step 2：使用 curl 测试流式响应

2.1 启动项目

bash
mvn spring-boot:run

2.2 测试健康检查

bash
curl http://localhost:8080/api/memory-chat/health

预期输出：

text
Memory Chat API is running! 🧠

2.3 测试流式聊天

bash
curl -N -X POST http://localhost:8080/api/memory-chat/stream \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "test-001",
    "message": "用50个字介绍 Spring AI"
  }'

💡 参数说明

  • -N - 禁用缓冲，立即显示数据
  • 没有 -N 会等待全部数据后才显示

预期输出（逐字显示）

text
data:Spring
data: AI
data: 是
data:Spring
data: 官方
data:的
data:AI
data:集成
data:框架
...

⚠️ 注意

  • 每行以 data: 开头（Spring 自动添加）
  • 冒号后可能有空格也可能没有
  • 这是标准的 SSE 格式

2.4 测试带记忆的流式对话

第1轮：告诉 AI 名字

bash
curl -N -X POST http://localhost:8080/api/memory-chat/stream \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "user-stream-001",
    "message": "我叫小明"
  }'

第2轮：询问名字（测试记忆）

bash
curl -N -X POST http://localhost:8080/api/memory-chat/stream \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "user-stream-001",
    "message": "我叫什么名字？"
  }'

预期：AI 记得名字，并且逐字显示回复

2.5 对比非流式和流式

非流式（等待后一次性显示）：

bash
time curl -X POST http://localhost:8080/api/memory-chat \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "test-002",
    "message": "用100个字介绍人工智能"
  }'

流式（逐字显示）：

bash
time curl -N -X POST http://localhost:8080/api/memory-chat/stream \
  -H "Content-Type: application/json" \
  -d '{
    "sessionId": "test-002",
    "message": "用100个字介绍人工智能"
  }'

观察：

🎨 Step 3：前端流式响应处理

3.1 SSE 解析示例

前端正确解析 SSE 格式的代码示例：

javascript
async function sendMessageStream(message) {
    // 创建 AI 消息容器
    const aiMessageId = appendMessage('ai', '');
    const aiTextDiv = document.getElementById(aiMessageId)
                              .querySelector('.message-text');
    
    // 添加打字机光标
    aiTextDiv.innerHTML = '<span class="typing-cursor">▋</span>';

    try {
        // 发起流式请求
        const response = await fetch(`${API_BASE_URL}/stream`, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
            },
            body: JSON.stringify({ 
                message: message,
                sessionId: sessionId 
            })
        });
        
        // 读取流式响应
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let fullText = '';

        while (true) {
            const { done, value } = await reader.read();
            
            if (done) {
                console.log('流式响应完成');
                break;
            }

            // 解码数据块
            const chunk = decoder.decode(value, { stream: true });
            console.log('收到数据块:', chunk);
            
            // SSE 格式：data:xxx 或 data: xxx
            // 提取实际内容
            const lines = chunk.split('\n');
            for (const line of lines) {
                // 只处理以 "data:" 开头的行
                if (line.startsWith('data:')) {
                    // 提取 data: 后面的内容（可能有空格也可能没有）
                    let content = line.substring(5);  // 跳过 "data:"
                    // 如果第一个字符是空格，去掉它
                    if (content.startsWith(' ')) {
                        content = content.substring(1);
                    }
                    // 只处理非空内容
                    if (content && content.trim()) {
                        console.log('提取内容:', content);
                        fullText += content;
                        // 实时更新显示（保留光标）
                        aiTextDiv.innerHTML = fullText + 
                            '<span class="typing-cursor">▋</span>';
                        // 自动滚动到底部
                        chatMessages.scrollTop = chatMessages.scrollHeight;
                    }
                }
            }
        }

        // 移除光标，显示最终文本
        aiTextDiv.innerHTML = fullText;

        // 刷新历史记录
        setTimeout(() => loadChatHistory(), 500);

    } catch (error) {
        console.error('流式响应失败:', error);
        aiTextDiv.innerHTML = '抱歉，发生了错误：' + error.message;
        throw error;
    }
}

3.2 SSE 解析兼容性

❌ 修复前（错误）

javascript
if (line.startsWith('data: ')) {  // 要求有空格
    const content = line.substring(6);  // 跳过 "data: "
}

问题：只能解析 data: 内容，无法解析 data:内容

✅ 修复后（正确）

javascript
if (line.startsWith('data:')) {  // 不要求空格
    let content = line.substring(5);  // 跳过 "data:"
    if (content.startsWith(' ')) {
        content = content.substring(1);  // 去掉可选的空格
    }
}

兼容性：

  • ✅ data:内容 - 正确解析
  • ✅ data: 内容 - 正确解析
  • ✅ 符合 SSE 标准

📁 项目结构更新

完成本章后的项目结构：

text
src/main/java/com/example/chatbot/
├── AiChatbotApplication.java
├── controller/
│   ├── ChatController.java              # 基础聊天
│   └── MemoryChatController.java        # 带记忆聊天（含流式接口）⭐
├── dto/
│   ├── ChatRequest.java
│   ├── ChatResponse.java
│   ├── MemoryChatRequest.java
│   └── MemoryChatResponse.java
└── service/
    └── ConversationService.java

📝 本章小结

🎓 你已经掌握：

  • 流式响应实现：使用 Server-Sent Events (SSE)
  • SSE 格式处理：Spring 自动处理，不要手动添加前缀
  • 流式 + 记忆：结合对话历史和流式响应
  • 线程安全：使用 AtomicReference 收集完整回复
  • 前端集成：正确解析 SSE 格式
  • 测试方法：使用 curl -N 测试流式响应

🚀 下一步学习

恭喜！你已经完成了 Spring AI 的核心功能学习。 现在你可以构建一个完整的、带记忆的、流式响应的 AI 聊天机器人了！

📝 本章小结

🎓 你完成了「流语」实时对话推送引擎的8大核心能力

  • 1、流式架构设计：基于 Spring WebFlux + Reactor 设计实时流式对话架构。
  • 2、打字机效果：构建支持 SSE 的流式对话服务，实现逐字输出、用户中断。
  • 3、响应式编程：掌握 Reactor 的 Mono/Flux 异步流模型。
  • 4、性能优化：实现背压处理、超时取消、P95/P99 延迟监控。
  • 5、交互优化：采用流式输出减少等待时长，配置快捷回复模板。
  • 6、多渠道适配：支持 Web 端、移动端 APP、小程序等多渠道接入。
  • 7、数据复盘：自动采集流式交互日志，分析用户中断场景。
  • 8、Spring 生态：Spring AI 与 WebFlux、Security、Gateway 整合。

🚀 下一步学习

你已掌握「流语」实时对话推送引擎的开发能力。 接下来，进入阶段 3：「智工」AI业务工具集成平台， 学习如何实现 Function Calling 和工具调用。

💡 常见问题

Q1：SSE 和 WebSocket 的区别？

A：

Q2：如何处理流式响应的错误？

A：

Q3：流式响应会增加成本吗？

A：

Q4：为什么不能手动添加 data: 前缀？

A：

Q5：如何测试流式响应？

A：

← 上一章 下一章 →