目录

1. 为什么普通接口不像ChatGPT？

2. 流式输出到底是什么？

3. 为什么AI聊天更适合使用SSE？

3.1 AI聊天的通信方向

4. Spring AI的流式调用链

4.1 同步调用

5. 使用Spring AI实现最简单的流式输出

5.1 创建ChatClient

6. 前端实现ChatGPT式打字效果

7. 从“能够运行”升级到“工程可用”

8. 改造前端事件处理

9. 原生EventSource为什么不适合直接发送复杂POST请求？

10. Spring MVC项目如何使用SseEmitter？

11. SSE和WebSocket应该怎么选？

12. 流式输出最常见的几个问题

13. 企业AI项目中的流式输出应该怎样设计？

14. 一次完整流式请求的执行过程

前言

上一篇文章，我们使用Spring AI的ChatClient，把原来调用大模型需要编写的大量HTTP请求代码，缩减成了几行链式调用。

如果还没有看过上一篇，可以先阅读：

📖 Java转AI第三课：Spring AI如何把100行HTTP代码缩减到10行？

上一课中，我们实现的接口大致是这样的：

String answer = chatClient.prompt()
        .user(question)
        .call()
        .content();

代码确实很简单。

但是运行之后，很快就会发现一个问题：

为什么ChatGPT可以一边生成、一边显示，而我们的接口必须等待几十秒，最后一次性返回全部内容？

假设模型生成一篇较长的分析报告需要20秒。

普通接口的体验是：

用户发送问题
    ↓
页面等待20秒
    ↓
一次性显示完整回答

在这20秒里，用户看不到任何反馈。

他不知道系统正在处理，还是接口已经卡住。

而ChatGPT的体验是：

用户发送问题
    ↓
模型生成第一部分内容
    ↓
前端立即显示
    ↓
后续内容持续追加

即使完整回答仍然需要20秒，用户可能在第1秒或者第2秒就已经看到第一段内容。

模型并没有变快，但用户感受到的等待时间明显缩短了。

这就是我们这一课要实现的功能：

使用Spring AI、Flux和SSE，让大模型生成的内容实时返回到浏览器，形成类似ChatGPT的打字效果。

本文不会只停留在一个能运行的Demo。

除了最基础的流式输出，我们还会继续处理：

  • 如何区分内容、完成和失败事件；
  • 前端怎样正确关闭SSE连接；
  • 模型调用超时后怎样返回提示；
  • Spring MVC项目能不能使用Flux；
  • 什么场景应该使用SseEmitter；
  • Nginx为什么可能把流式响应变成一次性返回；
  • 企业AI项目怎样设计阶段流和结果流。

1. 为什么普通接口不像ChatGPT？

先来看一下上一课中的同步调用：

String answer = chatClient.prompt()
        .user(question)
        .call()
        .content();

这里使用的是：

call()

它表示按照同步方式调用模型。

完整过程是：

Controller接收请求
    ↓
ChatClient调用大模型
    ↓
等待模型生成完整答案
    ↓
获得String
    ↓
Controller返回响应

只要模型还没有生成完，Controller就无法拿到最终的String。

因此，浏览器看到的通常是：

请求一直处于Pending状态

直到模型生成全部内容，接口才会一次性返回。

这种方式并没有错误。

对于分类、意图识别、结构化数据提取等内容较短的任务，同步调用反而更加简单。

例如：

判断用户意图
提取订单号
生成一个标题
返回一个JSON对象

这些任务可能只需要几百毫秒或者几秒，不一定需要流式输出。

但对于下面这些场景，同步返回的体验就比较差：

  • AI聊天；
  • 长文本生成；
  • RAG知识问答；
  • 案件或者合同分析；
  • 周报和报告生成；
  • Agent长任务执行；
  • 多步骤Workflow。

这些任务的共同特点是：

完整结果生成较慢，但前面已经生成的内容可以提前展示。

因此，我们需要把同步调用改成流式调用。

2. 流式输出到底是什么？

很多人第一次看到流式输出，会把它理解成：

模型一个字一个字地返回。

这个理解并不完全准确。

模型服务返回的内容，可能按照Token、词语或者文本片段进行传输。

例如模型最终要返回：

Spring AI可以帮助Java开发者快速接入大模型。

服务端实际收到的片段可能是：

Spring
 AI
可以帮助
Java
开发者
快速接入
大模型
。

也可能是：

Spring AI
可以帮助Java开发者
快速接入大模型。

Spring AI会把这些持续到达的内容包装成一个响应流。

后端收到一个片段，就立即向前端发送一个片段。

前端再把所有片段依次拼接起来：

answer += event.data;

最终就形成了我们看到的“打字效果”。

因此，流式输出的本质不是前端播放了一段预设动画，而是：

模型生成多少，服务端就传输多少，前端接收到多少就显示多少。

3. 为什么AI聊天更适合使用SSE？

SSE的全称是：

Server-Sent Events

中文通常翻译为：

服务器发送事件

它允许服务器通过一个持续保持的HTTP连接，不断向浏览器发送事件。

SSE响应的媒体类型是：

Content-Type: text/event-stream

一个最简单的SSE事件格式如下：

data: 第一段内容

data: 第二段内容

data: 第三段内容

每个事件之间使用空行分隔。

SSE还可以指定事件名称：

event: delta
data: 第一段模型内容

event: done
data: [DONE]

前端可以根据不同事件执行不同操作：

eventSource.addEventListener("delta", handleContent);
eventSource.addEventListener("done", handleComplete);

SSE标准就是为服务端持续向页面推送数据设计的，Spring MVC也直接提供了SseEmitter支持。

3.1 AI聊天的通信方向

AI聊天看起来是双向交互：

用户提问 → AI回答

但单次请求的真实通信过程是：

浏览器先发送一次问题
    ↓
服务器持续返回回答

在生成回答期间，主要的数据方向是：

服务器 → 浏览器

这正好符合SSE单向推送的特点。

3.2 SSE仍然基于HTTP

SSE没有引入一套完全不同的通信协议。

它仍然基于HTTP，因此可以继续使用：

  • Controller；
  • 网关；
  • Cookie；
  • Token鉴权；
  • 跨域配置；
  • Nginx反向代理；
  • 日志和调用链追踪。

对于已经使用Spring Boot开发业务系统的Java程序员来说，接入成本相对较低。

3.3 浏览器原生支持EventSource

浏览器提供了EventSource对象，可以直接连接SSE接口：

const eventSource = new EventSource("/api/ai/chat/stream");

不需要额外引入通信框架。

连接建立后，服务器每发送一次事件，前端就能收到一次通知。

4. Spring AI的流式调用链

Spring AI的ChatClient同时支持同步调用和流式调用。

同步调用使用：

call()

流式调用使用：

stream()

最常见的两段代码如下。

4.1 同步调用

String answer = chatClient.prompt()
        .user(question)
        .call()
        .content();

返回类型是：

String

4.2 流式调用

Flux<String> answer = chatClient.prompt()
        .user(question)
        .stream()
        .content();

返回类型变成了：

Flux<String>

Spring AI官方文档中，stream().content()返回Flux<String>，还可以通过chatResponse()获得包含模型元数据的Flux<ChatResponse>。

两段代码的核心区别只有一处：

- .call()
+ .stream()

但返回模型已经完全不同：

call().content()
    ↓
等待完整结果
    ↓
String

stream().content()
    ↓
持续接收结果
    ↓
Flux<String>

4.3 Flux是什么？

Flux来自Project Reactor。

可以先把它简单理解为：

一个未来可能持续产生多个数据的响应序列。

普通的List<String>要求所有数据都准备完成以后，才能得到完整集合。

而Flux<String>可以在每个数据生成后立即向下游发送：

第一个文本片段生成 → 立即发送
第二个文本片段生成 → 立即发送
第三个文本片段生成 → 立即发送

这正是大模型流式输出需要的能力。

Spring AI的StreamingChatModel底层同样通过Flux暴露流式响应。

5. 使用Spring AI实现最简单的流式输出

接下来进入代码部分。

本文继续沿用前三课中的Spring AI项目和大模型配置，不再重复介绍API Key和模型接入过程。

为了直接返回响应流，需要确保项目中包含WebFlux相关能力：

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

Spring AI模型Starter继续沿用之前项目中的配置即可。

5.1 创建ChatClient

@Configuration
public class AiConfig {

    @Bean
    public ChatClient chatClient(ChatClient.Builder builder) {
        return builder.build();
    }
}

Spring Boot会根据已经配置的ChatModel，自动提供ChatClient.Builder。

我们只需要调用：

builder.build()

即可获得ChatClient。

5.2 编写最简单的流式接口

@RestController
@RequestMapping("/api/ai/chat")
public class AiChatController {

    private final ChatClient chatClient;

    public AiChatController(ChatClient chatClient) {
        this.chatClient = chatClient;
    }

    @GetMapping(
            value = "/simple-stream",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public Flux<String> simpleStream(@RequestParam String message) {

        return chatClient.prompt()
                .user(message)
                .stream()
                .content();
    }
}

核心代码只有：

return chatClient.prompt()
        .user(message)
        .stream()
        .content();

接口通过下面的配置声明返回SSE数据：

produces = MediaType.TEXT_EVENT_STREAM_VALUE

它等价于：

Content-Type: text/event-stream

调用接口：

GET /api/ai/chat/simple-stream?message=请介绍一下Spring AI

浏览器就可以陆续收到模型生成的文本片段。

5.3 为什么Controller可以直接返回Flux？

Spring WebFlux可以识别Flux返回值，并在每个数据生成后写入HTTP响应。

因此我们不需要手动：

  • 创建循环；
  • 不断刷新输出流；
  • 拼接SSE协议格式；
  • 判断模型是否生成了新内容。

Spring AI负责把模型响应转换成Flux。

Spring WebFlux负责把Flux中的内容持续写入HTTP响应。

整体调用链是：

大模型流式响应
    ↓
Spring AI
    ↓
Flux<String>
    ↓
Spring WebFlux
    ↓
text/event-stream
    ↓
浏览器

6. 前端实现ChatGPT式打字效果

后端已经可以持续返回内容，接下来需要在前端接收SSE事件。

为了方便演示，先使用一个最简单的HTML页面。

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>Spring AI流式聊天</title>

    <style>
        body {
            max-width: 800px;
            margin: 40px auto;
            padding: 0 20px;
            font-family: Arial, sans-serif;
        }

        .input-area {
            display: flex;
            gap: 12px;
            margin-bottom: 24px;
        }

        #question {
            flex: 1;
            padding: 12px;
            font-size: 16px;
        }

        button {
            padding: 0 24px;
            cursor: pointer;
        }

        #answer {
            min-height: 200px;
            padding: 20px;
            border: 1px solid #ddd;
            border-radius: 8px;
            white-space: pre-wrap;
            line-height: 1.8;
        }
    </style>
</head>

<body>

<h2>Spring AI流式聊天</h2>

<div class="input-area">
    <input
        id="question"
        placeholder="请输入你的问题"
    />
    <button onclick="sendMessage()">发送</button>
</div>

<div id="answer"></div>

<script>
    let eventSource = null;

    function sendMessage() {
        const questionInput = document.getElementById("question");
        const answerElement = document.getElementById("answer");
        const question = questionInput.value.trim();

        if (!question) {
            alert("请输入问题");
            return;
        }

        if (eventSource) {
            eventSource.close();
        }

        answerElement.textContent = "";

        const url =
            "/api/ai/chat/simple-stream?message="
            + encodeURIComponent(question);

        eventSource = new EventSource(url);

        eventSource.onmessage = function (event) {
            answerElement.textContent += event.data;
        };

        eventSource.onerror = function () {
            eventSource.close();
            eventSource = null;
        };
    }
</script>

</body>
</html>

最核心的代码是：

eventSource.onmessage = function (event) {
    answerElement.textContent += event.data;
};

每收到一个文本片段，就追加到原来的回答后面：

原来的内容 + 新收到的内容

这样就形成了持续输出效果。

6.1 为什么使用textContent？

示例中使用的是：

answerElement.textContent += event.data;

而不是：

answerElement.innerHTML += event.data;

因为模型输出本质上是不可信文本。

如果直接把模型内容写入innerHTML，可能引入XSS风险。

真实项目如果需要渲染Markdown，应该经过：

Markdown解析
    ↓
HTML安全过滤
    ↓
页面渲染

不能直接把模型生成的HTML写进页面。

6.2 当前版本还有一个问题

目前的接口虽然可以运行，但它没有明确告诉前端：

回答什么时候正常结束？
回答什么时候发生异常？

EventSource连接在服务端关闭后可能进入错误或者重连处理。

如果前端不能区分“正常结束”和“网络失败”，就容易出现：

  • 回答完成后仍然自动重连；
  • 页面错误地显示连接失败；
  • 同一个问题被重复请求；
  • 加载状态一直不结束。

所以，真实项目不能只发送文本内容。

我们还需要设计完整的事件协议

7. 从“能够运行”升级到“工程可用”

接下来，我们给SSE增加三个事件类型：

事件 作用
delta 模型生成的文本片段
done 回答正常完成
failure 调用失败或者超时

这样前端就能准确判断当前状态。

7.1 返回结构化SSE事件

@Slf4j
@RestController
@RequestMapping("/api/ai/chat")
public class AiStreamController {

    private static final Duration STREAM_TIMEOUT =
            Duration.ofSeconds(90);

    private final ChatClient chatClient;

    public AiStreamController(ChatClient chatClient) {
        this.chatClient = chatClient;
    }

    @GetMapping(
            value = "/stream",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public Flux<ServerSentEvent<String>> stream(
            @RequestParam String message) {

        if (!StringUtils.hasText(message)) {
            return Flux.just(
                    buildEvent("failure", "问题不能为空")
            );
        }

        Flux<ServerSentEvent<String>> contentStream =
                chatClient.prompt()
                        .system("""
                                你是一名专业、严谨的AI助手。
                                请直接回答用户问题。
                                不要编造无法确认的信息。
                                """)
                        .user(message)
                        .stream()
                        .content()
                        .filter(StringUtils::hasText)
                        .map(content ->
                                buildEvent("delta", content)
                        );

        return contentStream
                .concatWithValues(
                        buildEvent("done", "[DONE]")
                )
                .timeout(STREAM_TIMEOUT)
                .onErrorResume(
                        TimeoutException.class,
                        exception -> {
                            log.warn(
                                    "AI流式响应超时，message={}",
                                    message
                            );

                            return Flux.just(
                                    buildEvent(
                                            "failure",
                                            "模型响应超时，请稍后重试"
                                    )
                            );
                        }
                )
                .onErrorResume(exception -> {
                    log.error(
                            "AI流式响应失败，message={}",
                            message,
                            exception
                    );

                    return Flux.just(
                            buildEvent(
                                    "failure",
                                    "模型服务暂时不可用，请稍后重试"
                            )
                    );
                })
                .doOnCancel(() ->
                        log.info("客户端主动取消流式连接")
                )
                .doFinally(signalType ->
                        log.info(
                                "流式请求结束，signal={}",
                                signalType
                        )
                );
    }

    private ServerSentEvent<String> buildEvent(
            String event,
            String data) {

        return ServerSentEvent.<String>builder()
                .event(event)
                .data(data)
                .build();
    }
}

需要导入的核心类：

import java.time.Duration;
import java.util.concurrent.TimeoutException;

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;

7.2 delta事件

每当模型生成一个新的文本片段，就发送：

buildEvent("delta", content)

实际SSE数据类似：

event: delta
data: Spring AI

event: delta
data: 是Spring生态中的

event: delta
data: AI应用开发框架

7.3 done事件

当模型正常生成完成后，追加：

buildEvent("done", "[DONE]")

前端收到done后，就可以：

  • 停止加载动画；
  • 恢复发送按钮；
  • 保存完整会话；
  • 主动关闭EventSource；
  • 记录本次回答完成。

7.4 failure事件

发生超时或者异常时，返回：

buildEvent(
        "failure",
        "模型服务暂时不可用，请稍后重试"
)

不要把下面这些内容直接返回给用户：

完整异常堆栈
模型供应商内部错误
API Key信息
网关地址
数据库或者服务地址

详细错误应该记录在服务端日志中。

前端只显示用户能够理解的提示。

7.5 为什么要设置超时？

大模型调用可能因为以下原因长时间没有结束：

  • 模型供应商负载过高；
  • 网络连接异常；
  • Prompt过长；
  • 模型正在执行复杂推理；
  • 上游服务没有正确关闭响应流；
  • Agent或者Tool调用阻塞。

如果没有超时控制，一条连接可能一直占用系统资源。

这里设置：

Duration.ofSeconds(90)

只是一个示例值。

真实项目应该根据业务类型分别配置：

普通聊天：30～90秒
RAG问答：60～120秒
复杂分析：120～180秒
Agent长任务：结合阶段流单独设计

超时时间不应该全部写死在Controller中，可以放入配置文件：

ai:
  stream:
    timeout-seconds: 90

8. 改造前端事件处理

后端已经有了明确的事件类型，前端也需要分别监听。

let eventSource = null;
let fullAnswer = "";

function sendMessage() {
    const question = document
        .getElementById("question")
        .value
        .trim();

    const answerElement =
        document.getElementById("answer");

    if (!question) {
        alert("请输入问题");
        return;
    }

    closeCurrentConnection();

    fullAnswer = "";
    answerElement.textContent = "正在连接模型……";

    const url =
        "/api/ai/chat/stream?message="
        + encodeURIComponent(question);

    eventSource = new EventSource(url);

    eventSource.onopen = function () {
        answerElement.textContent = "";
    };

    eventSource.addEventListener(
        "delta",
        function (event) {
            fullAnswer += event.data;
            answerElement.textContent = fullAnswer;
        }
    );

    eventSource.addEventListener(
        "done",
        function () {
            closeCurrentConnection();
            console.log("回答生成完成");
        }
    );

    eventSource.addEventListener(
        "failure",
        function (event) {
            answerElement.textContent =
                fullAnswer
                ? fullAnswer + "\n\n" + event.data
                : event.data;

            closeCurrentConnection();
        }
    );

    eventSource.onerror = function () {
        if (eventSource) {
            closeCurrentConnection();

            if (!fullAnswer) {
                answerElement.textContent =
                    "连接中断，请稍后重试";
            }
        }
    };
}

function closeCurrentConnection() {
    if (eventSource) {
        eventSource.close();
        eventSource = null;
    }
}

浏览器原生EventSource连接会保持打开状态，调用close()才会主动关闭。

因此，收到done或者failure事件后，一定要执行：

eventSource.close();

否则可能出现无意义的重连和资源占用。

8.1 页面销毁时也要关闭

用户刷新页面或者离开当前页面时，也应该清理连接：

window.addEventListener("beforeunload", function () {
    closeCurrentConnection();
});

在Vue或者React中，应在组件卸载阶段执行相同操作。

例如React：

useEffect(() => {
    return () => {
        eventSourceRef.current?.close();
    };
}, []);

8.2 防止重复提交

用户连续点击发送按钮时，应该先关闭上一次连接：

closeCurrentConnection();

更完整的做法是：

发送后禁用按钮
    ↓
接收流式内容
    ↓
done或者failure
    ↓
恢复按钮

这样可以避免同一页面同时建立多条无意义连接。

9. 原生EventSource为什么不适合直接发送复杂POST请求？

前面的示例使用的是：

GET /api/ai/chat/stream?message=用户问题

这是为了让读者最容易看到运行效果。

但真实AI聊天请求通常不只有一个message。

还可能包含：

{
  "conversationId": "c_10001",
  "message": "分析这份合同",
  "model": "reasoning-model",
  "knowledgeBaseId": "kb_001",
  "enableSearch": true
}

原生EventSource构造器主要接收URL和凭证选项，并没有提供配置请求体的能力。

因此，复杂场景通常有三种做法。

9.1 方案一：参数较少时继续使用GET

例如：

GET /api/ai/chat/stream
    ?conversationId=c_10001
    &message=你好

适合：

  • 学习Demo；
  • 简单聊天；
  • 参数较少；
  • 内容不敏感。

不适合：

  • Prompt很长；
  • 参数很多；
  • URL不能出现业务内容；
  • 需要复杂请求体。

9.2 方案二：使用fetch读取响应流

浏览器的fetch可以发送POST请求，同时通过Response.body读取流式响应。

基本结构是：

const response = await fetch(
    "/api/ai/chat/stream",
    {
        method: "POST",
        headers: {
            "Content-Type": "application/json",
            "Accept": "text/event-stream"
        },
        body: JSON.stringify(request)
    }
);

const reader = response.body.getReader();

这种方式可以发送完整JSON请求，但前端需要自行解析响应流和SSE事件边界。

9.3 方案三：先创建任务，再建立SSE连接

企业项目中可以拆成两个接口。

第一步，提交问题并创建任务：

POST /api/ai/chat/tasks

返回：

{
  "taskId": "task_202607150001"
}

第二步，通过任务ID建立SSE连接：

GET /api/ai/chat/tasks/task_202607150001/stream

这种方式的优势是：

  • POST负责接收复杂参数；
  • GET负责建立SSE连接；
  • 可以记录任务状态；
  • 可以支持断线恢复；
  • 可以查询历史结果；
  • 可以取消任务；
  • 更适合Agent和长时间Workflow。

普通聊天Demo没有必要一开始就设计成任务系统。

但长任务、复杂分析和Agent执行，建议采用任务化设计。

10. Spring MVC项目如何使用SseEmitter？

看到Flux以后，一些Java开发者可能会产生疑问：

我的项目一直使用Spring MVC，难道为了实现流式输出，必须把整个系统改成WebFlux吗？

答案是不需要。

Spring MVC本身支持异步返回和流式响应，也可以使用：

SseEmitter

Spring官方文档中，SseEmitter是ResponseBodyEmitter的子类，专门用于生成符合SSE格式的事件。

10.1 创建SseEmitter接口

@Slf4j
@RestController
@RequestMapping("/api/ai/workflow")
public class AiWorkflowController {

    private static final long TIMEOUT = 180_000L;

    private final Executor aiExecutor;

    public AiWorkflowController(
            @Qualifier("aiExecutor") Executor aiExecutor) {

        this.aiExecutor = aiExecutor;
    }

    @PostMapping(
            value = "/analysis",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public SseEmitter analysis(
            @RequestBody AnalysisRequest request) {

        SseEmitter emitter = new SseEmitter(TIMEOUT);

        emitter.onTimeout(() -> {
            log.warn("AI分析任务超时");
            emitter.complete();
        });

        emitter.onCompletion(() ->
                log.info("AI分析连接完成")
        );

        emitter.onError(exception ->
                log.warn(
                        "AI分析连接异常",
                        exception
                )
        );

        aiExecutor.execute(() ->
                executeAnalysis(request, emitter)
        );

        return emitter;
    }

    private void executeAnalysis(
            AnalysisRequest request,
            SseEmitter emitter) {

        try {
            sendEvent(
                    emitter,
                    "stage",
                    "正在识别用户意图"
            );

            // Intent识别
            String intent = recognizeIntent(request);

            sendEvent(
                    emitter,
                    "stage",
                    "正在获取业务数据"
            );

            // 查询业务系统、知识库等数据
            String context = loadContext(request, intent);

            sendEvent(
                    emitter,
                    "stage",
                    "正在生成分析结果"
            );

            // 调用模型生成最终结果
            String result = generateResult(
                    request,
                    context
            );

            sendEvent(
                    emitter,
                    "result",
                    result
            );

            sendEvent(
                    emitter,
                    "done",
                    "[DONE]"
            );

            emitter.complete();

        } catch (Exception exception) {
            log.error("AI分析任务执行失败", exception);

            try {
                sendEvent(
                        emitter,
                        "failure",
                        "分析任务执行失败，请稍后重试"
                );

                emitter.complete();

            } catch (Exception sendException) {
                log.warn(
                        "发送失败事件异常",
                        sendException
                );
            }
        }
    }

    private void sendEvent(
            SseEmitter emitter,
            String event,
            Object data) throws IOException {

        emitter.send(
                SseEmitter.event()
                        .name(event)
                        .data(data)
        );
    }
}

10.2 SseEmitter更适合什么场景？

SseEmitter非常适合传统Spring MVC项目中的阶段流：

正在识别意图
正在检索知识库
正在查询业务数据
正在执行工具
正在生成分析结果
任务完成

这和单纯的模型Token流不完全相同。

模型Token流主要展示：

模型正在生成的正文

阶段流主要展示：

系统当前执行到哪一步

例如我的AI办案助手使用的就是阶段化SSE设计：

识别用户意图
    ↓
加载案件信息
    ↓
检索相关法律知识
    ↓
调用案件工具
    ↓
执行综合研判
    ↓
返回完整结果

这种场景下，系统需要主动控制事件名称、执行阶段、业务状态和最终结果。

SseEmitter通常比简单返回Flux<String>更加直观。

10.3 Flux和SseEmitter不是互相替代

可以按照下面的方式理解：

方案 更适合的场景
Flux<String> 直接转发模型生成的文本流
Flux<ServerSentEvent<?>> 响应式项目中的结构化SSE
SseEmitter Spring MVC项目、阶段流、命令式Workflow
任务系统＋SSE Agent、长任务、需要恢复和状态查询

Spring MVC也可以接收Controller返回的Flux等响应式类型，但底层响应写入仍然是阻塞式的；WebFlux则采用非阻塞I/O。

因此，没有必要为了一个流式接口盲目重构整个项目。

应该根据现有技术栈和业务场景选择。

11. SSE和WebSocket应该怎么选？

提到实时通信，很多人第一反应是WebSocket。

那么AI聊天为什么经常使用SSE，而不是WebSocket？

先看两者的核心区别。

对比项 SSE WebSocket
通信方向 服务器向客户端单向推送 双向全双工通信
底层方式 HTTP长连接 独立WebSocket协议
浏览器API EventSource WebSocket
自动重连 浏览器原生支持 通常需要自己实现
数据格式 主要是文本事件 文本和二进制
接入复杂度 较低 相对较高
AI流式回答 非常适合 可以实现但通常没有必要
即时通信 能力有限 更适合
多人协作 不适合 更适合

11.1 使用SSE的典型场景

  • AI回答流式输出；
  • 模型生成报告；
  • 任务进度通知；
  • Agent执行阶段；
  • 日志实时展示；
  • 服务端事件通知。

11.2 使用WebSocket的典型场景

  • 即时聊天；
  • 多人协作编辑；
  • 实时游戏；
  • 双向语音；
  • 高频状态同步；
  • 客户端需要持续向服务端发送数据。

AI问答的一次请求通常是：

用户提交一次问题
服务器持续返回一次回答

并不需要在模型生成过程中持续双向发送数据。

因此，SSE往往已经足够。

不要因为WebSocket看起来更加“实时”，就默认所有AI聊天都必须使用WebSocket。

技术选型应该匹配真实通信模型，而不是选择功能最重的方案。

 

12. 流式输出最常见的几个问题

代码在本地能够流式返回，不代表部署到服务器以后仍然能够正常运行。

下面这些问题，在真实项目中非常常见。

12.1 问题一：本地是流式，经过Nginx后变成一次性返回

现象是：

本地调用正常逐段显示
部署后等待很久
最后一次性显示全部内容

常见原因是Nginx对上游响应进行了缓冲。

Nginx官方文档说明，关闭代理缓冲后，响应会在接收到时同步传递给客户端，而不是尝试先读取完整响应。

可以针对流式接口关闭缓冲：

location /api/ai/ {
    proxy_pass http://ai-app:8080;

    proxy_http_version 1.1;

    proxy_buffering off;
    proxy_cache off;

    proxy_read_timeout 180s;
    proxy_send_timeout 180s;
}

也可以由上游响应设置：

X-Accel-Buffering: no

实际配置应结合现有网关和Nginx版本验证。

12.2 问题二：接口有数据，但前端一直不显示

需要检查以下内容：

响应Content-Type是否为text/event-stream
Controller是否真的返回Flux或SseEmitter
中间网关是否缓冲响应
前端是否逐段读取
服务端是否等全部内容拼接后才返回

错误写法：

StringBuilder result = new StringBuilder();

chatClient.prompt()
        .user(message)
        .stream()
        .content()
        .doOnNext(result::append)
        .blockLast();

return result.toString();

虽然调用了：

stream()

但后面又使用：

blockLast()

等待整个流结束。

最终还是一次性返回。

这相当于把流式调用重新变回了同步调用。

12.3 问题三：回答完成以后接口重复请求

原生EventSource具有连接恢复机制。

如果服务端关闭连接，而前端没有判断这是正常结束，浏览器可能尝试重新连接。

解决方法是发送明确的完成事件：

event: done
data: [DONE]

前端收到后主动执行：

eventSource.close();

12.4 问题四：浏览器关闭后，后端任务仍然继续执行

浏览器页面关闭，并不意味着所有上游模型请求都一定能够立即终止。

需要在后端关注：

.doOnCancel(...)
.doFinally(...)

对于复杂任务，还应该设计：

  • 任务取消状态；
  • 上游请求取消；
  • Tool调用中断；
  • 资源释放；
  • 未完成任务清理；
  • Token成本记录。

不能只关闭前端页面，却让后台继续执行一个高成本模型任务。

12.5 问题五：长时间没有Token，连接被网关断开

有些模型会长时间推理后才返回第一段内容。

Agent执行Tool时，也可能几十秒没有正文输出。

这时，网关或者负载均衡可能认为连接已经空闲。

可以设计心跳事件：

event: heartbeat
data: ping

前端收到后不展示，只用于维持连接和更新连接状态。

Spring官方文档也建议流式连接定期发送数据，以便发现客户端断开并维持连接活动。

12.6 问题六：每个Token都写数据库

不要在每次收到文本片段时执行：

写MySQL
写Redis
记录完整日志
更新会话表

模型一次回答可能产生大量片段。

每个片段都写数据库会带来：

  • 大量数据库请求；
  • 锁竞争；
  • IO放大；
  • 日志膨胀；
  • 响应速度下降。

更合理的方式是：

流式阶段：在内存中拼接完整回答
    ↓
回答完成：一次性保存最终结果

或者按较大批次异步落库。

12.7 问题七：直接记录完整Prompt和回答

AI请求日志中可能包含：

  • 用户隐私；
  • 企业内部文档；
  • 案件信息；
  • 身份信息；
  • API返回数据；
  • Tool执行结果。

生产环境不应该默认记录完整Prompt和完整回答。

建议记录：

traceId
conversationId
userId
业务场景
模型名称
Token数量
首包耗时
总耗时
执行结果
失败类型

正文是否记录，需要根据业务安全要求进行脱敏、截断或者完全关闭。

13. 企业AI项目中的流式输出应该怎样设计？

一个简单聊天机器人，返回模型正文即可。

但真实企业AI项目的执行链路可能是：

用户问题
    ↓
权限校验
    ↓
意图识别
    ↓
查询业务数据
    ↓
RAG检索
    ↓
Tool调用
    ↓
模型分析
    ↓
结果校验
    ↓
返回用户

如果整个过程只展示模型最后生成的Token，用户仍然会在前面的业务阶段长时间等待。

因此，企业项目通常需要区分两类流。

13.1 第一类：模型内容流

用于展示模型正在生成的正文：

delta：根据当前案件信息……
delta：综合相关法律规定……
delta：建议重点核查以下内容……

13.2 第二类：业务阶段流

用于展示系统当前执行状态：

stage：正在识别问题类型
stage：正在检索企业知识库
stage：正在获取案件信息
stage：正在调用时间线工具
stage：正在生成分析结果

13.3 统一事件协议

可以为整个AI系统设计统一的SSE事件：

事件 含义
connected 连接建立
stage 业务执行阶段
delta 模型文本片段
result 最终结构化结果
citation 引用来源
usage Token和耗时
done 正常完成
failure 执行失败
heartbeat 心跳

一次完整的AI请求可能是：

event: connected
data: {"traceId":"trace_001"}

event: stage
data: {"name":"INTENT","message":"正在识别用户意图"}

event: stage
data: {"name":"RETRIEVAL","message":"正在检索相关知识"}

event: delta
data: 根据当前检索结果，

event: delta
data: 可以从以下三个方面进行分析。

event: citation
data: {"documentId":"doc_001","title":"业务规范"}

event: usage
data: {"totalTokens":1820,"latencyMs":12600}

event: done
data: [DONE]

这样做的价值不只是页面更好看。

它还可以支持：

  • 前端统一处理不同AI业务；
  • 展示Workflow执行阶段；
  • 记录Agent调用链；
  • 返回RAG引用来源；
  • 统计模型Token；
  • 定位失败阶段；
  • 对接统一AI Gateway；
  • 构建调用可观测性。

14. 一次完整流式请求的执行过程

现在把本文内容连接起来。

用户发送问题：

请解释Spring AI中的ChatClient

前端建立连接：

new EventSource(
    "/api/ai/chat/stream?message=..."
);

Spring Boot接收请求：

@GetMapping(
    value = "/stream",
    produces = MediaType.TEXT_EVENT_STREAM_VALUE
)

Spring AI调用模型：

chatClient.prompt()
        .user(message)
        .stream()
        .content();

模型持续生成内容：

Spring AI
是Spring生态中的
AI应用开发框架

后端将内容转换成事件：

event: delta
data: Spring AI

前端持续拼接：

fullAnswer += event.data;

模型生成完成：

event: done
data: [DONE]

前端关闭连接：

eventSource.close();

最终完整链路是：

用户输入
    ↓
EventSource建立SSE连接
    ↓
Spring Boot接收请求
    ↓
ChatClient.stream()
    ↓
大模型持续生成
    ↓
Flux持续发出文本片段
    ↓
Spring发送SSE事件
    ↓
前端逐段追加
    ↓
收到done事件
    ↓
关闭连接并保存会话

总结

这一课，我们完成了Spring AI聊天机器人中非常重要的一次升级：

一次性返回
    ↓
流式返回

核心代码其实非常简单：

chatClient.prompt()
        .user(message)
        .stream()
        .content();

但真正把流式输出用于项目，还需要继续处理：

  • text/event-stream响应类型；
  • Flux<String>流式数据；
  • EventSource前端连接；
  • delta、done、failure事件；
  • 连接主动关闭；
  • 模型调用超时；
  • 用户取消请求；
  • Nginx代理缓冲；
  • 心跳和长连接；
  • 阶段流和模型内容流；
  • Token、耗时和调用链记录。

这也是AI Demo和真实AI应用之间的区别。

Demo关注的是：

能不能把模型内容显示出来？

真实项目还要继续考虑：

连接是否会被正确关闭，异常是否能够感知，长任务是否能够追踪，模型成本是否能够统计，系统经过网关和Nginx以后是否仍然可以稳定运行？

对于Java开发者来说，Spring AI最有价值的地方，不只是减少了几行调用模型的代码。

它让我们可以继续使用熟悉的Spring生态，把模型能力接入现有的Controller、网关、业务服务、日志、监控和部署体系。

模型负责生成内容。

Spring负责把这份不确定的生成能力，接入一个可以管理、可以观测、可以治理的企业系统。

下一篇预告

流式输出解决了AI回答的实时体验问题。

但现在的聊天机器人仍然存在一个明显问题：

用户：我叫小明。

AI：你好，小明。

用户：我刚才说我叫什么？

AI：我不知道。

原因是大模型接口本身通常是无状态的。

每次请求结束以后，模型并不会自动记住之前的对话。

下一篇文章，我们将继续实现：

Java转AI第五课：Spring AI实现多轮对话记忆，让AI真正记住上下文

下一课会重点介绍：

  • 大模型为什么没有记忆；
  • ChatMemory是什么；
  • 如何根据conversationId隔离会话；
  • MessageWindowChatMemory如何控制上下文窗口；
  • 对话记录应该放内存、Redis还是数据库；
  • 为什么不能把所有历史消息全部发送给模型；
  • 企业项目中的短期记忆、摘要记忆和长期记忆。

📚 推荐专栏

🤖 AI 技术专栏

从 0 到 1 学习 AI 应用开发，持续分享 Spring AI、RAG、Agent、企业级 AI 项目实战。

👉 https://blog.csdn.net/qupengkun/category_13184360.html

🚀 AI 转型日记

记录一名 10 年 Java 开发者从传统后端转向 AI 应用开发的全过程。

👉 https://blog.csdn.net/qupengkun/category_13183497.html

🧩 AI 开源项目实战

持续分享Java AI业务项目、AI Coding效率工具和工程治理实践。

👉 https://blog.csdn.net/qupengkun/category_13189373.html

👨‍💻 关于作者

QCoding

专注 AI 应用开发与 Java 技术实践。

持续分享 Spring AI、RAG、Agent、企业级 AI 项目实战、架构设计与职业成长。

# java # 人工智能 # spring # websocket
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

【AI踩坑实录】让AI写了3个月代码，这7个坑让我返工了整整2周——附完整避坑清单

avatar AI编程社区
cover

导出文档选 AI 导出鸭，彻底解决：Claude 复制到 Word 为什么有星号？各类格式兼容难题

avatar AI编程社区
cover

Opus 爱质疑、Sonnet 爱鼓励——Anthropic 解剖了 Claude 的人格，和我用了一年的感受一模一样

avatar AI编程社区