Spring Boot Vue 实战如何用SSE实现Dify大模型的流式输出附完整代码最近在做一个内部知识库问答系统后端用Spring Boot前端用Vue 3。核心需求是接入Dify平台的大模型能力实现类似ChatGPT那种逐字输出的“打字机”效果。一开始觉得不就是个HTTP请求吗真做起来才发现从传统的请求-响应模式切换到流式输出Streaming前后端都得动点“手术”。特别是当Dify的响应模式设置为streaming时它返回的不是一个完整的JSON而是一个持续的、分块的数据流。这对习惯了axios和RestController的Java开发者来说是个挺有意思的挑战。你需要处理长连接、管理事件流还要确保前端能平滑地接收并渲染这些陆续到达的数据块。本文将分享我如何用Spring Boot的SseEmitter和Vue配合microsoft/fetch-event-source一步步搭建起这个流式通道并附上踩坑后优化的完整代码。1. 理解流式输出与SSE为何不用WebSocket在开始敲代码之前有必要先厘清几个概念。当我们谈论大模型的“流式输出”时指的是服务器将生成的文本内容分成多个小块chunks持续地发送给客户端而不是等全部内容生成完毕后再一次性返回。这种模式能极大提升用户体验用户无需等待漫长的生成过程可以实时看到思考的“痕迹”。实现这种持续通信常见的有两种技术WebSocket和Server-Sent Events。WebSocket全双工通信协议连接建立后客户端和服务器可以随时相互发送消息。功能强大适合聊天室、实时游戏等需要高频双向交互的场景。Server-Sent Events简称SSE是一种服务器向客户端推送文本消息的轻量级协议。它基于普通的HTTP协议是单向的服务器到客户端。那么为什么在接入Dify这类大模型时我选择了SSE而非WebSocket呢特性Server-Sent EventsWebSocket通信方向单向服务器-客户端双向客户端-服务器协议基于HTTP/HTTPS独立的ws://或wss://协议连接管理自动重连、事件ID追踪需手动实现数据格式文本流通常为UTF-8格式为data: content\n\n二进制或文本帧格式自定义复杂度简单前端可使用原生EventSource相对复杂需处理握手、帧解析等适用场景实时通知、日志流、大模型流式输出在线聊天、协作编辑、实时游戏注意Dify等多数大模型平台在流式模式下返回的数据格式正是符合SSE规范的。我们的后端角色更像一个“代理”或“适配器”接收Dify的SSE流并将其转发给我们的前端。这是一个典型的单向数据流场景SSE的简单性和对HTTP生态的友好性使其成为更合适的选择。Spring Boot为SSE提供了两种主要的编程模型一种是基于Servlet API的SseEmitter属于spring-webmvc另一种是基于响应式编程的ServerSentEvent和Flux属于spring-webflux。本文将重点介绍更易于理解和集成在传统Spring MVC项目中的SseEmitter方案。2. 后端搭建Spring Boot与SseEmitter深度集成我们的后端目标很明确提供一个接收前端请求的HTTP端点这个端点去调用Dify的流式聊天接口然后将Dify返回的SSE事件流原样地、实时地转发给前端。2.1 项目依赖与环境准备首先确保你的pom.xml包含了必要的依赖。除了基础的Web依赖我们还需要引入spring-boot-starter-webflux。这是因为我们需要使用WebClient来调用Dify的接口而WebClient是Spring WebFlux的核心组件之一。别担心这不会把你的项目变成完全的响应式项目我们只是局部使用它。dependencies !-- Spring Boot Web MVC (用于SseEmitter) -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- Spring Boot WebFlux (用于WebClient) -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency !-- 其他常用依赖如Lombok、JSON处理等 -- dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /dependency /dependencies2.2 核心Service层连接Dify并转发事件流Service层是整个流程的中枢。这里我将创建一个DifyStreamService其核心方法streamChat负责与Dify交互。import lombok.extern.slf4j.Slf4j; import org.springframework.http.MediaType; import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.stereotype.Service; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import reactor.netty.http.client.HttpClient; import java.time.Duration; import java.util.function.Consumer; Service Slf4j public class DifyStreamService { // Dify API的基础URL和密钥建议配置在application.yml中 private final String difyBaseUrl https://api.dify.ai/v1; private final String apiKey app-YourApiKeyHere; private final WebClient webClient; public DifyStreamService() { // 配置WebClient设置合理的超时时间 HttpClient httpClient HttpClient.create() .responseTimeout(Duration.ofSeconds(30)); // 响应超时 this.webClient WebClient.builder() .clientConnector(new ReactorClientHttpConnector(httpClient)) .baseUrl(difyBaseUrl) .defaultHeader(Authorization, Bearer apiKey) .defaultHeader(Content-Type, application/json) .build(); } public SseEmitter streamChat(ChatRequest request) { // 1. 创建SseEmitter设置超时时间例如5分钟 SseEmitter emitter new SseEmitter(5 * 60 * 1000L); // 2. 构建请求Dify的请求体 MapString, Object requestBody new HashMap(); requestBody.put(inputs, request.getInputs()); // 上下文参数 requestBody.put(query, request.getQuery()); // 用户问题 requestBody.put(response_mode, streaming); // 关键设置为流式 requestBody.put(conversation_id, request.getConversationId()); requestBody.put(user, request.getUserId()); log.info(调用Dify流式接口请求参数: {}, requestBody); // 3. 发起对Dify的流式请求 webClient.post() .uri(/chat-messages) .bodyValue(requestBody) .accept(MediaType.TEXT_EVENT_STREAM) // 接受SSE流 .retrieve() .bodyToFlux(String.class) // 将响应体转换为字符串流 .subscribe( data - { // 4. 成功接收到Dify返回的一个数据块 log.debug(收到Dify数据块: {}, data); try { // 将数据块通过SseEmitter发送给前端 // 注意Dify返回的数据可能包含data: 前缀和JSON结构需要解析 if (data.startsWith(data: )) { String jsonData data.substring(6); // 去掉data: emitter.send(jsonData); // 发送纯JSON字符串给前端 } } catch (IOException e) { log.error(向客户端发送SSE消息失败, e); emitter.completeWithError(e); } }, error - { // 5. 处理Dify调用或流处理中的错误 log.error(从Dify接收流数据时发生错误, error); emitter.completeWithError(error); }, () - { // 6. Dify流正常结束 log.info(Dify流式响应结束); emitter.complete(); } ); // 7. 设置SseEmitter的生命周期回调 emitter.onCompletion(() - log.info(SSE连接完成)); emitter.onTimeout(() - { log.warn(SSE连接超时); emitter.complete(); }); emitter.onError(err - log.error(SSE连接发生错误, err)); return emitter; } }代码关键点解析WebClient配置我们使用WebClient而非RestTemplate因为它天然支持响应式流处理SSE响应更加优雅。responseTimeout至关重要防止挂起的长连接。SseEmitter超时new SseEmitter(5 * 60 * 1000L)设置了一个较长的超时时间因为大模型生成可能需要数十秒。bodyToFlux(String.class)这是将HTTP响应体转换为一个FluxString流。每个String就是Dify服务器发来的一个事件行例如data: {message: hello}。数据清洗Dify返回的SSE数据行通常以data:开头。我们在转发给前端前去掉了这个前缀只发送内部的JSON字符串方便前端直接JSON.parse。错误传播无论是Dify侧的错误还是向前端发送数据时的IO异常都需要捕获并通过emitter.completeWithError()告知前端。资源清理在onCompletion和onTimeout回调中可以进行必要的日志记录和资源清理。2.3 Controller层提供SSE端点Controller层的实现非常简洁主要职责是定义API路径、接收参数并调用Service。import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; RestController RequestMapping(/api/chat) public class ChatController { private final DifyStreamService difyStreamService; public ChatController(DifyStreamService difyStreamService) { this.difyStreamService difyStreamService; } PostMapping(value /stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter streamChat(RequestBody ChatRequest request) { return difyStreamService.streamChat(request); } }核心注解说明PostMapping由于请求体可能较大包含上下文通常使用POST。produces MediaType.TEXT_EVENT_STREAM_VALUE这是声明该端点返回text/event-stream类型的响应这是SSE的标准MIME类型。浏览器或EventSourceAPI识别到这个类型就会以SSE的方式处理响应。2.4 处理Dify返回的复杂事件流上面的示例假设Dify返回的是简单的data: {json}格式。实际上Dify的流可能更复杂包含不同事件类型例如event: message表示一个普通消息块event: end表示流结束event: error表示错误为了更健壮地处理我们可以定义一个DTO来匹配Dify的事件结构并使用bodyToFlux(ServerSentEvent.class)。import org.springframework.http.codec.ServerSentEvent; import reactor.core.publisher.Flux; // 在Service方法中使用ParameterizedTypeReference ParameterizedTypeReferenceServerSentEventString typeRef new ParameterizedTypeReferenceServerSentEventString() {}; FluxServerSentEventString eventStream webClient.post() .uri(/chat-messages) .bodyValue(requestBody) .retrieve() .bodyToFlux(typeRef); eventStream.subscribe( sse - { if (message.equals(sse.event())) { String data sse.data(); // 解析data中的JSON提取实际内容 emitter.send(extractContentFromData(data)); } else if (end.equals(sse.event())) { emitter.complete(); } else if (error.equals(sse.event())) { emitter.completeWithError(new RuntimeException(sse.data())); } }, error - emitter.completeWithError(error), () - emitter.complete() // 流正常结束但Dify可能已通过event:end通知 );这种方式能更精确地控制不同事件的处理逻辑。3. 前端实现Vue 3与fetch-event-source的完美配合前端的目标是创建一个可以发送问题、并实时显示流式回答的界面。原生EventSourceAPI不支持POST方法和自定义请求头这对于需要传递认证信息和复杂请求体的我们来说是不可接受的。因此我们选择微软的microsoft/fetch-event-source库它基于Fetch API功能更强大。3.1 安装依赖与基础准备在你的Vue 3项目中安装必要的包。npm install microsoft/fetch-event-source我们将在一个Vue组件中实现核心逻辑。假设我们有一个简单的聊天界面包含一个输入框、一个发送按钮和一个显示区域。3.2 核心组件实现template div classchat-container div classmessage-display refdisplayArea div v-for(msg, index) in messages :keyindex :class[message, msg.role] {{ msg.content }} /div !-- 当前流式响应的临时显示行 -- div v-ifstreamingText classmessage assistant streaming {{ streamingText }} /div /div div classinput-area textarea v-modeluserInput keydown.enter.exact.preventsendMessage placeholder输入您的问题.../textarea button clicksendMessage :disabledisLoading发送/button button clickabortRequest v-ifisLoading停止/button /div /div /template script setup import { ref, nextTick } from vue; import { fetchEventSource } from microsoft/fetch-event-source; const userInput ref(); const messages ref([]); // 存储历史消息 {role: user|assistant, content: string} const streamingText ref(); // 存储当前正在接收的流式文本 const isLoading ref(false); const abortController ref(null); // 用于中止请求 const API_URL http://your-backend-host/api/chat/stream; const sendMessage async () { if (!userInput.value.trim() || isLoading.value) return; const userMessage userInput.value.trim(); userInput.value ; // 添加用户消息到历史 messages.value.push({ role: user, content: userMessage }); // 准备开始流式接收 streamingText.value ; isLoading.value true; abortController.value new AbortController(); const requestBody { query: userMessage, // 可以根据需要传递上下文、会话ID等 // inputs: {}, // conversation_id: some-id, // user: user-123 }; try { await fetchEventSource(API_URL, { method: POST, headers: { Content-Type: application/json, // 如果需要可以在这里添加认证头 // Authorization: Bearer ${yourToken} }, body: JSON.stringify(requestBody), signal: abortController.value.signal, // 用于中止请求 openWhenHidden: true, // 即使页面隐藏也保持连接可选 async onopen(response) { // 连接成功建立 console.log(SSE连接已打开状态码:, response.status); if (response.ok) { return; // 一切正常 } else { throw new Error(服务器响应错误: ${response.status}); } }, onmessage(event) { // 收到服务器发送的一个事件 // 我们的后端发送的是纯JSON字符串所以直接解析 try { const parsedData JSON.parse(event.data); // 假设Dify返回的数据结构里答案在 answer 或 content 字段 // 实际情况需要根据Dify的API文档调整 const chunk parsedData.answer || parsedData.content || parsedData.message || ; if (chunk) { // 将新的文本块追加到当前流式文本中 streamingText.value chunk; // 滚动到显示区域底部 scrollToBottom(); } } catch (e) { console.error(解析SSE数据失败:, e, 原始数据:, event.data); } }, onclose() { // 连接被服务器关闭 console.log(SSE连接已关闭); finalizeStreamingResponse(); }, onerror(err) { // 发生错误网络错误、解析错误、或我们在onopen中抛出的错误 console.error(SSE连接发生错误:, err); // 不要自动重试除非你明确想要这个行为 throw err; // 抛出错误会停止fetchEventSource触发finally块 }, }); } catch (error) { console.error(请求过程发生错误:, error); // 可以在这里给用户一个错误提示 messages.value.push({ role: assistant, content: 请求出错: ${error.message} }); } finally { finalizeStreamingResponse(); } }; const finalizeStreamingResponse () { if (streamingText.value) { // 将完整的流式响应存入历史消息 messages.value.push({ role: assistant, content: streamingText.value }); } streamingText.value ; isLoading.value false; abortController.value null; }; const abortRequest () { if (abortController.value) { abortController.value.abort(); console.log(用户中止了请求); finalizeStreamingResponse(); } }; // 工具函数滚动到底部 const displayArea ref(null); const scrollToBottom () { nextTick(() { if (displayArea.value) { displayArea.value.scrollTop displayArea.value.scrollHeight; } }); }; /script style scoped .chat-container { display: flex; flex-direction: column; height: 600px; } .message-display { flex: 1; overflow-y: auto; border: 1px solid #ccc; padding: 10px; margin-bottom: 10px; } .message { margin-bottom: 10px; padding: 8px; border-radius: 5px; } .message.user { background-color: #e3f2fd; text-align: right; } .message.assistant { background-color: #f5f5f5; } .message.streaming { /* 可以为流式响应添加一些特殊样式比如闪烁的光标 */ border-right: 2px solid #333; animation: blink 1s infinite; } keyframes blink { 50% { border-color: transparent; } } .input-area { display: flex; } .input-area textarea { flex: 1; height: 60px; resize: none; } .input-area button { margin-left: 10px; } /style前端代码关键点解析fetchEventSource配置这是库的核心函数。我们配置了method: POST、headers和body使其能像普通Fetch请求一样工作同时又具备处理SSE流的能力。signal我们将AbortController的signal传入这使得我们可以通过调用abortController.abort()来主动中止一个正在进行的流式请求比如用户点击了“停止”按钮。onmessage(event)这是最重要的回调。每次后端推送一个SSE事件即我们通过emitter.send()发送的数据到这里event.data就包含了我们发送的字符串。我们将其解析为JSON并提取出文本内容块累加到streamingText响应式变量中。实时渲染由于streamingText是响应式的Vue会自动更新DOM实现逐字打印的效果。scrollToBottom函数确保新内容总是可见。连接生命周期管理onopen、onclose、onerror回调让我们能妥善处理连接的各种状态比如在onerror中抛出错误以停止重试在onclose或finally块中完成本次流式响应的收尾工作将临时文本存入历史消息。用户体验优化提供了“停止”按钮允许用户中断生成过程。为流式响应添加了闪烁光标的CSS动画增强“正在输入”的感知。4. 进阶优化与生产环境考量上面的代码已经可以跑通一个基础的流式对话。但要投入生产环境还需要考虑更多。4.1 后端优化连接管理与错误处理连接池与超时大量并发流式请求可能耗尽服务器资源。需要合理配置WebClient的连接池如HttpClient的maxConnections和SseEmitter的超时时间。背压处理如果前端处理速度慢于后端推送速度可能导致内存问题。SseEmitter本身处理能力有限在极端情况下可以考虑使用响应式的WebFlux控制器返回FluxServerSentEvent它能更好地处理背压。结构化日志与监控为SSE连接的生命周期创建、发送消息、完成、超时、错误添加详细的、结构化的日志并集成到你的APM应用性能监控系统中便于排查问题。身份认证与授权在Controller或Interceptor中验证前端请求的合法性如JWT Token确保只有授权用户才能建立昂贵的SSE连接。4.2 前端优化用户体验与健壮性自动重连策略fetchEventSource库内置了重试逻辑。你可以通过retry相关参数如retryInterval,maxRetries配置更智能的重连策略例如指数退避。历史记录与会话管理将messages数组持久化到localStorage或Pinia/Vuex store中实现页面刷新后聊天记录不丢失。管理conversation_id实现多轮对话上下文。更丰富的数据展示Dify的流可能返回包含引用来源、思考过程等元数据。前端可以解析这些数据高亮显示引用或者展示一个“正在思考...”的占位符。网络状态感知监听onerror和onclose当网络中断或服务器异常时给用户友好的提示并提供“重新连接”或“重新发送”的选项。4.3 部署与跨域问题在开发环境下前端(localhost:8080)访问后端(localhost:8081)会遇到跨域问题。SSE请求同样受同源策略限制。后端解决方案Spring Boot配置CORSimport org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.servlet.config.annotation.CorsRegistry; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; Configuration public class CorsConfig implements WebMvcConfigurer { Override public void addCorsMappings(CorsRegistry registry) { registry.addMapping(/api/**) // 针对你的API路径 .allowedOrigins(http://localhost:8080) // 你的前端地址 .allowedMethods(GET, POST, PUT, DELETE, OPTIONS) .allowedHeaders(*) .allowCredentials(true) .maxAge(3600); // 特别注意对于SSE可能不需要额外配置但确保OPTIONS预检请求通过。 } }前端代理方案Vue CLI / Vite如果你使用开发服务器配置代理可以避免CORS。// vue.config.js (Vue CLI) module.exports { devServer: { proxy: { /api: { target: http://localhost:8081, // 后端地址 changeOrigin: true, } } } }; // vite.config.js (Vite) export default defineConfig({ server: { proxy: { /api: { target: http://localhost:8081, changeOrigin: true, } } } })配置好后前端请求/api/chat/stream就会被代理到http://localhost:8081/api/chat/stream从而规避跨域。整个流程走下来从后端的WebClient订阅Dify流到SseEmitter桥接再到前端的fetchEventSource逐块渲染形成了一个完整的数据管道。这种模式不仅适用于Dify对于任何提供SSE格式流式响应的AI API如OpenAI兼容接口都通用。关键在于理解SSE这种“长轮询”变体的特性并在前后端做好连接的生命周期管理和错误处理。在实际项目中根据流量规模可能还需要考虑引入消息队列或专门的流式代理服务来解耦和扩容但上述核心代码已经为你打下了坚实的地基。