第5章: 流式响应体验拉满：打字机效果与中断处理

📋 本章目标

本章通过 8 个完整的测试用例，带你掌握 Spring AI 的流式响应功能。你将学会如何使用 Reactor Flux 处理流式数据，实现打字机效果的 AI 回复。

✨ 你将学到什么？

✅ 理解流式响应的概念和优势
✅ 使用 stream() 方法获取流式数据
✅ 掌握 Reactor Flux 的基本用法
✅ 使用 StepVerifier 测试响应式流
✅ 统计和分析流式数据块
✅ 处理超时和取消订阅
✅ 对比流式和非流式的性能

📚 测试用例索引

测试 5.1：基础流式响应
测试 5.2：使用 StepVerifier 测试
测试 5.3：统计数据块数量
测试 5.4：对比流式和非流式
测试 5.5：流式响应完整性验证
测试 5.6：流式响应的超时处理
测试 5.7：流式响应的取消

🎯 前置准备

1. 确认已完成前面的章节

✅ 完成第1章：Spring AI 基础知识测试
✅ 理解 ChatModel 的基本用法
✅ 理解 Prompt 和 ChatResponse

2. 理解流式响应

什么是流式响应？

text

非流式（传统方式）：
用户提问 → 等待... → 一次性返回完整回答

流式（打字机效果）：
用户提问 → "你" → "好" → "！" → "我" → "是" → ...

✅ 流式响应的优势

用户体验更好（即时反馈）
减少等待时间的感觉
可以提前看到部分内容
类似 ChatGPT 的打字机效果

3. 理解 Reactor Flux

💡 Flux 是什么？

Reactor 框架的响应式流
可以发射 0-N 个元素
支持异步、非阻塞处理
Spring AI 使用 Flux 实现流式响应

📝 创建测试类

1. 测试类基本结构

创建文件：src/test/java/com/example/chatbot/knowledge/Test05_StreamingResponse.java

java
package com.example.chatbot.knowledge;

import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.chat.prompt.Prompt;
import org.springframework.ai.chat.messages.UserMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.jupiter.api.Assertions.*;

/**
 * 测试类：流式响应
 * 
 * 本测试类演示如何使用流式响应实现打字机效果
 * 
 * 核心概念：
 * - Flux：响应式流，可以发射多个元素
 * - stream()：ChatModel 的流式调用方法
 * - StepVerifier：测试响应式流的工具
 */
@SpringBootTest(classes = com.example.chatbot.AiChatbotApplication.class)
@DisplayName("知识点5：流式响应")
public class Test05_StreamingResponse {

    @Autowired
    private ChatModel chatModel;
    
    // 测试方法将在这里添加
}

📦 新增导入说明

Flux - 响应式流
StepVerifier - 测试响应式流的工具
AtomicInteger - 线程安全的计数器
Duration - 时间间隔

🧪 测试 5.1：基础流式响应

第一个流式响应测试，了解基本用法。

测试代码

java
/**
 * 测试 5.1：基础流式响应
 * 
 * 知识点：
 * - stream() 方法返回 Flux<ChatResponse>
 * - Flux 会逐块发射数据
 * - 使用 subscribe() 订阅流
 */
@Test
@DisplayName("5.1 基础流式响应")
public void test01_BasicStreaming() {
    System.out.println("\n========== 测试 5.1：基础流式响应 ==========");
    
    // 步骤1：创建 Prompt
    Prompt prompt = new Prompt("用一句话介绍 Spring AI");
    System.out.println("👤 用户: " + prompt.getContents());
    
    // 步骤2：调用流式 API
    System.out.println("🤖 AI (流式): ");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 步骤3：订阅并打印每个数据块
    StringBuilder fullResponse = new StringBuilder();
    
    stream.subscribe(
        // onNext: 每收到一块数据
        response -> {
            String content = response.getResult().getOutput().getText();
            System.out.print(content);  // 打印这一块
            fullResponse.append(content);
        },
        // onError: 发生错误
        error -> {
            System.err.println("\n❌ 错误: " + error.getMessage());
        },
        // onComplete: 流结束
        () -> {
            System.out.println("\n✅ 流式响应完成");
            System.out.println("📝 完整回复: " + fullResponse.toString());
        }
    );
    
    // 步骤4：等待流完成
    try {
        Thread.sleep(5000);  // 等待5秒
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    // 步骤5：验证
    assertFalse(fullResponse.toString().isEmpty(), 
        "应该收到完整的回复");
    
    System.out.println("✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test01_BasicStreaming

预期输出

text

========== 测试 5.1：基础流式响应 ==========
👤 用户: 用一句话介绍 Spring AI
🤖 AI (流式): 
Spring AI 是...（逐字显示）
✅ 流式响应完成
📝 完整回复: Spring AI 是一个...
✅ 测试通过

💡 理解要点

stream() 返回 Flux<ChatResponse>
subscribe() 订阅流，处理每个数据块
onNext 处理每个数据块
onComplete 流结束时调用

🧪 测试 5.2：使用 StepVerifier

学习使用 StepVerifier 测试响应式流。

测试代码

java
/**
 * 测试 5.2：使用 StepVerifier
 * 
 * 知识点：
 * - StepVerifier 是测试响应式流的专用工具
 * - 可以验证流的行为
 * - 更适合单元测试
 */
@Test
@DisplayName("5.2 使用 StepVerifier")
public void test02_WithStepVerifier() {
    System.out.println("\n========== 测试 5.2：使用 StepVerifier ==========");
    
    // 步骤1：创建流
    Prompt prompt = new Prompt("说一个数字");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 步骤2：使用 StepVerifier 验证
    System.out.println("🔍 使用 StepVerifier 验证流...");
    
    StepVerifier.create(stream.take(1))
        // 只验证“至少能收到一个 chunk”，避免流式多次 onNext 导致 expectComplete 失败
        .expectNextMatches(resp -> {
            String text = resp.getResult().getOutput().getText();
            return text != null && !text.isEmpty();
        })
        .expectComplete()
        // 验证（最多等待30秒）
        .verify(Duration.ofSeconds(30));
    
    System.out.println("✅ StepVerifier 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test02_WithStepVerifier

✅ StepVerifier 的作用

StepVerifier.create() 创建验证器
expectNextCount() 期望元素数量
expectComplete() 期望流正常结束
verify() 执行验证

🧪 测试 5.3：统计数据块数量

统计流式响应的数据块数量和字符数。

测试代码

java
/**
 * 测试 5.3：统计数据块数量
 * 
 * 知识点：
 * - 流式响应会分多个块发送
 * - 可以统计收到的块数
 * - 了解流式传输的特点
 */
@Test
@DisplayName("5.3 统计数据块数量")
public void test03_CountChunks() {
    System.out.println("\n========== 测试 5.3：统计数据块数量 ==========");
    
    // 步骤1：创建计数器
    AtomicInteger chunkCount = new AtomicInteger(0);
    AtomicInteger totalChars = new AtomicInteger(0);
    
    // 步骤2：创建流
    Prompt prompt = new Prompt("用50个字介绍 Java");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 步骤3：订阅并统计
    System.out.println("📊 开始统计...");
    
    stream.subscribe(
        response -> {
            String content = response.getResult().getOutput().getText();
            int count = chunkCount.incrementAndGet();
            totalChars.addAndGet(content.length());
            
            System.out.println("📦 块 " + count + ": \"" + content + 
                "\" (长度: " + content.length() + ")");
        },
        error -> System.err.println("❌ 错误: " + error.getMessage()),
        () -> {
            System.out.println("\n📊 统计结果:");
            System.out.println("  - 总块数: " + chunkCount.get());
            System.out.println("  - 总字符数: " + totalChars.get());
            System.out.println("  - 平均每块: " + 
                (totalChars.get() / chunkCount.get()) + " 字符");
        }
    );
    
    // 步骤4：等待完成
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    // 步骤5：验证
    assertTrue(chunkCount.get() > 0, "应该收到至少一个数据块");
    assertTrue(totalChars.get() > 0, "应该收到一些字符");
    
    System.out.println("✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test03_CountChunks

💡 理解要点

使用 AtomicInteger 进行线程安全的计数
每个数据块可能包含不同数量的字符
流式传输会分多次发送

🧪 测试 5.4：对比流式和非流式

对比流式和非流式响应的区别。

测试代码

java
/**
 * 测试 5.4：对比流式和非流式
 * 
 * 知识点：
 * - 流式：逐块返回，体验更好
 * - 非流式：一次性返回，等待时间长
 * - 内容相同，只是返回方式不同
 */
@Test
@DisplayName("5.4 对比流式和非流式")
public void test04_StreamingVsNonStreaming() {
    System.out.println("\n========== 测试 5.4：对比流式和非流式 ==========");
    
    String question = "用一句话介绍 Spring Boot";
    Prompt prompt = new Prompt(question);
    
    // 步骤1：非流式调用
    System.out.println("\n【非流式调用】");
    System.out.println("👤 用户: " + question);
    
    long startTime1 = System.currentTimeMillis();
    ChatResponse response1 = chatModel.call(prompt);
    long endTime1 = System.currentTimeMillis();
    
    String content1 = response1.getResult().getOutput().getText();
    System.out.println("🤖 AI: " + content1);
    System.out.println("⏱️  耗时: " + (endTime1 - startTime1) + "ms");
    System.out.println("💡 特点: 等待后一次性返回");
    
    // 步骤2：流式调用
    System.out.println("\n【流式调用】");
    System.out.println("👤 用户: " + question);
    System.out.println("🤖 AI (流式): ");
    
    long startTime2 = System.currentTimeMillis();
    StringBuilder content2 = new StringBuilder();
    
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    stream.subscribe(
        response -> {
            String chunk = response.getResult().getOutput().getText();
            System.out.print(chunk);
            content2.append(chunk);
        },
        error -> {},
        () -> {
            long endTime2 = System.currentTimeMillis();
            System.out.println("\n⏱️  耗时: " + (endTime2 - startTime2) + "ms");
            System.out.println("💡 特点: 逐字显示，体验更好");
        }
    );
    
    // 等待流完成
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    // 步骤3：验证内容相似
    System.out.println("\n📊 对比结果:");
    System.out.println("  - 非流式长度: " + content1.length());
    System.out.println("  - 流式长度: " + content2.length());
    
    assertFalse(content1.isEmpty());
    assertFalse(content2.toString().isEmpty());
    
    System.out.println("✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test04_StreamingVsNonStreaming

🧪 测试 5.5：流式响应完整性验证

使用 StepVerifier 验证流的完整性。

测试代码

java
/**
 * 测试 5.5：流式响应完整性
 * 
 * 知识点：
 * - 使用 StepVerifier 验证流的完整性
 * - expectComplete() 确保流正常结束
 * - 验证流没有中断或出错
 */
@Test
@DisplayName("5.5 流式响应完整性")
public void test05_StreamingCompleteness() {
    System.out.println("\n========== 测试 5.5：流式响应完整性 ==========");
    
    Prompt prompt = new Prompt("说'你好'");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 使用 StepVerifier 验证流的完整性
    StepVerifier.create(stream)
        .thenConsumeWhile(response -> {
            String content = response.getResult().getOutput().getText();
            System.out.print(content);
            return true;
        })
        .expectComplete()
        .verify(Duration.ofSeconds(30));
    
    System.out.println("\n\n💡 知识点：");
    System.out.println("- StepVerifier 可以验证流的完整性");
    System.out.println("- expectComplete() 确保流正常结束");
    System.out.println("- 如果流中断或出错，测试会失败");
    
    System.out.println("\n✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test05_StreamingCompleteness

🧪 测试 5.6：流式响应的超时处理

学习如何设置超时时间。

测试代码

java
/**
 * 测试 5.6：流式响应的超时处理
 * 
 * 知识点：
 * - 可以设置超时时间
 * - 超时后会触发错误
 * - 使用 timeout() 操作符
 */
@Test
@DisplayName("5.6 流式响应的超时处理")
public void test06_StreamingTimeout() {
    System.out.println("\n========== 测试 5.6：流式超时处理 ==========");
    
    // 步骤1：创建流
    Prompt prompt = new Prompt("讲一个长故事");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 步骤2：设置超时（10秒）
    System.out.println("⏰ 设置10秒超时...");
    
    stream
        .timeout(Duration.ofSeconds(10))
        .subscribe(
            response -> {
                String content = response.getResult().getOutput().getText();
                System.out.print(content);
            },
            error -> {
                if (error instanceof java.util.concurrent.TimeoutException) {
                    System.out.println("\n⏰ 超时！");
                } else {
                    System.err.println("\n❌ 错误: " + error.getMessage());
                }
            },
            () -> {
                System.out.println("\n✅ 在超时前完成");
            }
        );
    
    // 等待
    try {
        Thread.sleep(12000);  // 等待12秒
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    System.out.println("✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test06_StreamingTimeout

🧪 测试 5.7：流式响应的取消

学习如何取消正在进行的流。

测试代码

java
/**
 * 测试 5.7：流式响应的取消
 * 
 * 知识点：
 * - 可以取消正在进行的流
 * - 使用 Disposable 控制订阅
 * - 节省资源
 */
@Test
@DisplayName("5.7 流式响应的取消")
public void test07_StreamingCancellation() {
    System.out.println("\n========== 测试 5.7：流式取消 ==========");
    
    // 步骤1：创建流
    Prompt prompt = new Prompt("讲一个很长的故事");
    Flux<ChatResponse> stream = chatModel.stream(prompt);
    
    // 步骤2：订阅并保存 Disposable
    System.out.println("📡 开始订阅...");
    
    AtomicInteger count = new AtomicInteger(0);
    java.util.concurrent.atomic.AtomicReference<reactor.core.Disposable> disposableRef = new java.util.concurrent.atomic.AtomicReference<>();
    
    reactor.core.Disposable disposable = stream.subscribe(
        response -> {
            int current = count.incrementAndGet();
            if (current > 5) {
                return;
            }

            String content = response.getResult().getOutput().getText();
            String printable = content == null ? "null" : content.replace("\n", "\\n");
            System.out.println("\n📦 第" + current + "块: " + printable);
            
            // 收到5个块后取消
            if (current == 5) {
                System.out.println("\n🛑 收到5个块，立即取消订阅...");
                reactor.core.Disposable d = disposableRef.get();
                if (d != null && !d.isDisposed()) {
                    d.dispose();
                }
            }
        },
        error -> System.err.println("\n❌ 错误: " + error.getMessage()),
        () -> System.out.println("\n✅ 流完成")
    );
    disposableRef.set(disposable);
    
    // 步骤3：等待一会儿后取消
    try {
        Thread.sleep(2000);  // 等待2秒
        if (!disposable.isDisposed()) {
            System.out.println("🛑 取消订阅");
            disposable.dispose();
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    // 步骤4：验证已取消
    assertTrue(disposable.isDisposed(), "订阅应该已被取消");
    
    System.out.println("✅ 测试通过");
}

运行测试

bash
mvn test -Dtest=Test05_StreamingResponse#test07_StreamingCancellation

🚀 运行所有测试

运行完整测试类，验证所有流式响应功能：

bash
mvn test -Dtest=Test05_StreamingResponse

预期输出

text

Tests run: 7, Failures: 0, Errors: 0, Skipped: 0

✅ 所有测试通过！

测试用例总结

测试编号	测试名称	核心知识点
5.1	基础流式响应	stream() 和 subscribe()
5.2	使用 StepVerifier	测试响应式流
5.3	统计数据块数量	AtomicInteger 计数
5.4	对比流式和非流式	性能和体验对比
5.5	流式响应完整性	验证流正常结束
5.6	超时处理	timeout() 操作符
5.7	取消订阅	Disposable 控制

📚 课后习题与常见面试题

请完成下面的练习，巩固本章的“流式响应”核心机制与测试验证能力

⚠️ 提示：本章高频点集中在 Flux、流式分块、StepVerifier、超时/取消、以及 SSE/WebFlux 的工程化落地。

🎯 课后习题（动手为主）

1) 写一个 StepVerifier 测试：验证 stream() 至少发出 1 个 chunk 并正常 complete

要求：对 chatModel.stream(prompt) 使用 StepVerifier，至少 expectNextCount(1)，并 expectComplete()。附加：设置合理的 verify(Duration) 超时，避免测试挂死。

参考答案（展开/收起）

java
Flux<String> stream = chatModel.stream(prompt)
    .map(resp -> resp.getResult().getOutput().getText())
    .filter(s -> s != null && !s.isBlank());

// 用 take(1) 保证一定会 complete，避免真实 stream 长时间不断流导致用例挂死
StepVerifier.create(stream.take(1))
    .expectNextMatches(s -> !s.isBlank())
    .verifyComplete();

说明：如果你直接对原始 stream 写 expectNextCount(1).expectComplete()，在“模型会继续吐后续 chunk”的情况下，可能会因为流迟迟不结束而导致测试卡住。

2) 统计“chunk 数量 + 首块耗时”：输出 chunkCount、firstChunkTime、totalTime

要求：使用 AtomicInteger 统计 chunkCount；记录首块到达耗时（TTFB）与总耗时；最终打印对比结果：为什么流式的“首块时间”对体验更重要。

参考答案（展开/收起）

java
AtomicInteger chunkCount = new AtomicInteger();
AtomicLong firstChunkMs = new AtomicLong(-1);
long start = System.currentTimeMillis();

Flux<String> stream = chatModel.stream(prompt)
    .map(resp -> resp.getResult().getOutput().getText())
    .filter(s -> s != null && !s.isBlank())
    .doOnNext(s -> {
        int idx = chunkCount.incrementAndGet();
        if (idx == 1) {
            firstChunkMs.set(System.currentTimeMillis() - start);
        }
    });

stream.doFinally(sig -> {
    long total = System.currentTimeMillis() - start;
    System.out.println("chunkCount=" + chunkCount.get());
    System.out.println("firstChunkTimeMs=" + firstChunkMs.get());
    System.out.println("totalTimeMs=" + total);
}).blockLast();

说明：TTFB（首块耗时）决定了用户“多久看到第一个字”，通常比 totalTime 更影响体验。

3) 加入 timeout 保护：把 stream() 包一层 timeout，并在超时后给出兜底提示

要求：对流式结果增加 timeout(Duration.ofSeconds(x))；超时后给出一条兜底信息（例如“本次响应超时，请稍后重试”）。用测试验证：超时场景下不会无限等待。

参考答案（展开/收起）

java
Flux<String> safe = stream
    .timeout(Duration.ofSeconds(10))
    .onErrorResume(java.util.concurrent.TimeoutException.class,
        e -> Flux.just("本次响应超时，请稍后重试"));

StepVerifier.create(safe)
    .thenConsumeWhile(s -> true)
    .verify(Duration.ofSeconds(12));

说明：timeout 是“超过时限就失败”，通常要配合 onErrorResume 给用户一个可读的兜底输出。

4) 演示“取消订阅”：只取前 N 个 chunk 后取消，验证后续不再继续打印

任选方式： 方式 A：stream.take(N)； 方式 B：保留 Disposable 并手动 dispose()。重点说明：为什么取消是必要能力（用户离开页面、网络断开、节省资源）。

参考答案（展开/收起）

java
// 方式 A：take(N) 自动取消上游订阅
stream.take(5)
    .doOnNext(s -> System.out.println("chunk=" + s))
    .doOnCancel(() -> System.out.println("已取消订阅"))
    .blockLast();

说明：取消是“消费者主动停止”，可避免继续消耗模型配额/网络资源。

5) 写一个 WebFlux SSE 端点：把 Flux 的内容以 text/event-stream 输出到前端

要求：实现一个返回 Flux<String> 或 Flux<ServerSentEvent<String>> 的接口；前端使用 EventSource 接收并逐字渲染，形成“打字机效果”。

参考答案（展开/收起）

java
@GetMapping(value = "/sse", produces = org.springframework.http.MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> sse() {
    return chatModel.stream(prompt)
        .map(resp -> resp.getResult().getOutput().getText())
        .filter(s -> s != null && !s.isBlank());
}

javascript
const es = new EventSource('/api/sse');
es.onmessage = (e) => { document.querySelector('#out').textContent += e.data; };

说明：SSE 是基于 HTTP 的“服务端推送”，天然适合“逐块文本”场景。

💼 常见面试题（概念 + 场景）

1) 什么是“流式响应”？它和一次性返回有什么本质区别？

关键点：流式是“分块、边生成边发送”，能显著降低首字节时间（TTFB），提升交互体验； 非流式是“生成完再返回”，总耗时可能相近，但用户会一直等待。

参考答案（展开/收起）

定义：

流式响应是服务端把生成结果按 chunk 连续推送；一次性返回是生成完成后再整体返回。

核心差异：

1) 体验：流式首字节更快（TTFB），用户能更早看到输出。 2) 控制能力：流式更强调取消/超时/背压；一次性返回通常只有超时。 3) 工程形态：流式通常是 Flux + SSE/WebFlux；一次性返回是普通 JSON/文本。

适用场景：

长文本生成、打字机效果、需要“边生成边展示”的交互式聊天。

注意点：

要处理客户端断开与取消；要做空 chunk 过滤/拼接；要设置超时与资源释放，避免后台继续生成。

2) Reactor 的 Flux 是什么？它和 List/Iterator 有什么不同？

Flux 是响应式流：支持异步、非阻塞、按需推送（背压）； List/Iterator 是同步拉取模型，无法天然表达“逐步产生 + 逐步消费”的异步链路。

参考答案（展开/收起）

定义：

Flux 是 Reactor 的响应式 Publisher，表示 0..N 个元素的异步流（Mono 表示 0..1）。

与 List/Iterator 的区别：

1) 推/拉模型：Flux 是推送（带背压语义），Iterator 是同步拉取。 2) 线程模型：Flux 默认非阻塞 + 可通过 Scheduler 控制；List/Iterator 通常阻塞式遍历。 3) 信号语义：Flux 有 onNext/onComplete/onError/cancel；Iterator 只有 next/hasNext。

为什么适合流式聊天：

模型输出天然是“逐步产生”的，Flux 可以表达逐 chunk 到达、取消、超时与链路编排。

3) StepVerifier 适合测试哪些内容？常用断言有哪些？

StepVerifier 适合测试 Flux/Mono 的“行为”，包括：发出元素数量、元素内容、完成信号、错误信号、取消行为。常用：expectNextCount、expectNextMatches、expectComplete、expectError、thenCancel。

参考答案（展开/收起）

适合测试什么：

StepVerifier 用来验证 Flux/Mono 的行为与信号序列：发了哪些元素、何时完成、是否报错、是否被取消。

常用断言：

expectNext/expectNextMatches（内容）， expectNextCount（数量）， expectComplete/expectError（终止信号）， thenCancel（取消）， verify(Duration)（防挂死）。

流式测试注意点：

真实流可能“很久才 complete”，通常用 take(n) 或 thenCancel() 限制测试范围，并设置超时。

4) timeout 和 cancel 有什么区别？为什么流式接口必须处理取消？

timeout 是“超过时限就失败/中断”；cancel 是“消费者主动停止”。流式接口如果不处理取消：用户离开页面仍继续生成、占用网络/线程/模型配额，导致资源浪费。

参考答案（展开/收起）

区别：

timeout：超过时限触发 error，属于系统保护（防止无限等待）。 cancel：消费者主动停止订阅（触发 cancel），属于用户/业务行为（比如用户点停止/关闭页面）。

为什么必须处理取消：

流式链路如果忽略 cancel，上游可能继续生成并占用连接/线程/模型配额；正确做法是收到 cancel 就尽快释放资源（或终止上游）。

落地做法：

前端 abort/关闭连接；后端用 doOnCancel/take(n)/Disposable.dispose() 等方式停止订阅并记录一次日志。

5) SSE 和 WebSocket 怎么选？在“流式聊天”场景各有什么优缺点？

SSE：单向（服务端→浏览器）、基于 HTTP、简单、天然适合“逐块文本”；缺点是双向能力弱。 WebSocket：全双工、适合复杂交互；缺点是实现与运维更复杂。

参考答案（展开/收起）

结论先行：“流式聊天输出”优先 SSE；确实需要双向实时通道再选 WebSocket。

SSE：

HTTP 单向推送（服务端→浏览器），实现简单，浏览器支持 EventSource；适合逐块文本输出。缺点：双向弱（发消息仍需 HTTP）。

WebSocket：

全双工，适合高频双向交互/协作；缺点是协议升级、连接治理、心跳与运维复杂度更高。

选型要点：

关注是否需要双向、代理/网关支持、断线重连、并发连接数与资源治理。

📝 本章小结

🎓 你已经掌握：

流式响应基础：stream() 方法和 Flux 的使用
响应式编程：Reactor Flux 的概念和操作
StepVerifier：测试响应式流的专用工具
流式响应特性：数据块统计、错误处理、超时控制
订阅管理：取消订阅和资源释放
性能优化：首字节时间和用户体验提升

🚀 下一步学习

继续学习 第6章：项目三 · 流式聊天API，将本章学到的流式响应知识应用到实际项目中，实现打字机效果的聊天接口！