当前位置： 首页 > news >正文

【Java转AI实战】第3讲：流式输出——LLM的“分页查询“，附生产级SSE踩坑实录

news 2026/7/6 14:04:01

你一定用过SELECT * FROM table LIMIT 10000的痛——用户盯着白屏干等8秒。LLM也一样，一次性返回完整回答可能要20秒。流式输出就是LLM的"分页查询"：数据边生成边推送，用户第一个字0.5秒就能看到。

为什么LLM必须流式返回

先理解一个事实：LLM是逐token生成的。它不是像数据库那样"先算好再返回"，而是一个token一个token往外蹦。一次对话生成500个token，每个token耗时30-50ms，总共15-25秒。

如果等全部生成完再返回（非流式），用户体验是这样的：

用户提问 → 白屏等待15秒 → 突然蹦出一大段文字

如果边生成边推送（流式），用户体验变成：

用户提问 → 0.5秒后开始出字 → 像打字机一样逐字显示 → 15秒后全部显示完

同样15秒的总耗时，体感完全不同。这就是ChatGPT"打字机效果"背后的技术本质。

Java类比：Stream的惰性求值

如果你用过Java 8的Stream API，流式输出的概念就很好理解：

Java概念LLM流式输出共同点
list.stream().map(...)惰性求值LLM逐token生成不是一次性算完，而是"按需产出"
Stream.forEach(System.out::println)SSE逐事件推送数据产出一个就消费一个
Flux.fromIterable(list).delayElements(...)Token流异步、非阻塞、背压控制
BufferedReader.readLine()循环SSE事件循环逐行/逐事件读取，直到结束标志

用代码表达就是：

// 非流式 = collect再处理（用户干等）List<String>result=stream.collect(Collectors.toList());returnresult;// 全部收集完才返回// 流式 = forEach边产出边消费（用户边看边等）stream.forEach(token->pushToClient(token));// 产出一个就推一个

SSE协议：HTTP版的"消息推送"

LLM流式输出用的是**SSE（Server-Sent Events）**协议。你可以把它理解为"HTTP版的WebSocket单向版"——服务端持续向客户端推送数据，客户端只需要监听。

SSE vs WebSocket vs 轮询

特性SSEWebSocketHTTP轮询
通信方向单向（服务端→客户端）双向客户端主动拉
协议HTTP/1.1ws://HTTP
断线重连浏览器自动重连需手动实现天然支持
数据格式纯文本（text/event-stream）二进制/文本任意
适用场景LLM流式输出、通知推送聊天室、实时协作简单状态轮询

SSE数据格式

SSE的报文格式极其简单，只有4个字段：

event: chunk data: {"content":"你"} event: chunk data: {"content":"好"} event: done data: {"totalLength":2}

核心规则：

  • Content-Type必须是text/event-stream
  • 每个事件用空行分隔
  • data:后面跟数据内容（通常是JSON）
  • event:可选，指定事件类型（不写则默认message）

在Spring Boot中，SSE通过SseEmitter实现。

LangChain4j的流式输出

上一讲我们用的是ChatLanguageModel——同步调用，等全部生成完才返回。现在换成它的流式版本StreamingChatLanguageModel。

新增依赖

在第2讲的pom.xml基础上，不需要额外依赖——langchain4j-open-ai-spring-boot-starter已经同时支持同步和流式模型。

配置流式模型

langchain4j:open-ai:streaming-chat-model:api-key:${DEEPSEEK_API_KEY:sk-xxxxx}model-name:deepseek-chatbase-url:https://api.deepseek.com/v1

注意key的变化：chat-model→streaming-chat-model。Spring Boot会自动注入一个StreamingChatLanguageModelBean。

后端Controller：SSE流式接口

@RestController@RequestMapping("/api/chat")publicclassStreamChatController{privatefinalStreamingChatLanguageModelstreamingModel;publicStreamChatController(StreamingChatLanguageModelstreamingModel){this.streamingModel=streamingModel;}@GetMapping(value="/stream",produces=MediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterchat(@RequestParamStringmessage){// 超时5分钟，LLM长回答可能需要较长时间SseEmitteremitter=newSseEmitter(5*60*1000L);streamingModel.generate(message,newStreamingResponseHandler<AiMessage>(){@OverridepublicvoidonNext(Stringtoken){try{// 每生成一个token就推送一次emitter.send(SseEmitter.event().name("chunk").data(Map.of("content",token)));}catch(IOExceptione){emitter.completeWithError(e);}}@OverridepublicvoidonComplete(Response<AiMessage>response){try{emitter.send(SseEmitter.event().name("done").data(Map.of("totalTokens",response.tokenUsage().totalTokenCount())));emitter.complete();}catch(IOExceptione){emitter.completeWithError(e);}}@OverridepublicvoidonError(Throwableerror){emitter.completeWithError(error);}});returnemitter;}}

这个接口做的事情很简单：

  1. 创建一个SseEmitter，设置5分钟超时
  2. 调用streamingModel.generate()，传入回调处理器
  3. 每收到一个token就通过emitter.send()推送给前端
  4. 生成完毕调用emitter.complete()关闭连接

前端：打字机效果

前端用浏览器原生的EventSourceAPI接收SSE流。下面是一个完整的HTML页面，复制到src/main/resources/static/index.html即可运行：

<!DOCTYPEhtml><htmllang="zh-CN"><head><metacharset="UTF-8"><title>LLM流式对话</title><style>*{margin:0;padding:0;box-sizing:border-box;}body{font-family:-apple-system,sans-serif;background:#f5f5f5;display:flex;justify-content:center;padding-top:40px;}.container{width:700px;}.chat-box{background:white;border-radius:12px;padding:24px;min-height:300px;margin-bottom:16px;box-shadow:0 2px 8pxrgba(0,0,0,0.08);}#response{white-space:pre-wrap;line-height:1.8;color:#333;}.cursor{display:inline-block;width:2px;height:1em;background:#333;animation:blink 0.8s infinite;vertical-align:text-bottom;}@keyframesblink{50%{opacity:0;}}.input-row{display:flex;gap:8px;}input{flex:1;padding:12px 16px;border:1px solid #ddd;border-radius:8px;font-size:15px;outline:none;}button{padding:12px 24px;background:#1a73e8;color:white;border:none;border-radius:8px;font-size:15px;cursor:pointer;}button:disabled{background:#ccc;}.status{margin-top:8px;font-size:13px;color:#999;}</style></head><body><divclass="container"><divclass="chat-box"><divid="response">等待提问...</div></div><divclass="input-row"><inputid="input"placeholder="输入你的问题..."/><buttonid="sendBtn"onclick="sendMessage()">发送</button></div><divclass="status"id="status"></div></div><script>leteventSource=null;functionsendMessage(){constinput=document.getElementById('input');constmessage=input.value.trim();if(!message)return;// 重置UIdocument.getElementById('response').innerHTML='';document.getElementById('sendBtn').disabled=true;document.getElementById('status').textContent='生成中...';input.value='';// 关闭上一个连接if(eventSource)eventSource.close();// 建立SSE连接consturl=`/api/chat/stream?message=${encodeURIComponent(message)}`;eventSource=newEventSource(url);constresponseEl=document.getElementById('response');// 监听chunk事件——每个token到达时触发eventSource.addEventListener('chunk',function(e){constdata=JSON.parse(e.data);responseEl.textContent+=data.content;});// 监听done事件——生成完毕eventSource.addEventListener('done',function(e){constdata=JSON.parse(e.data);document.getElementById('status').textContent=`生成完毕，共${data.totalTokens}tokens`;document.getElementById('sendBtn').disabled=false;eventSource.close();});// 错误处理eventSource.onerror=function(){document.getElementById('status').textContent='连接断开';document.getElementById('sendBtn').disabled=false;eventSource.close();};}// 回车发送document.getElementById('input').addEventListener('keydown',function(e){if(e.key==='Enter')sendMessage();});</script></body></html>

启动Spring Boot应用，访问http://localhost:8080，输入问题，你就能看到LLM的回答像打字机一样逐字蹦出来。

关键代码解读：

  • new EventSource(url)建立SSE长连接，浏览器原生支持
  • addEventListener('chunk', ...)监听自定义事件类型，对应后端SseEmitter.event().name("chunk")
  • eventSource.close()生成完毕后主动关闭连接，释放资源

生产实录

实录1：WebSocket vs SSE双通道——为什么不是"二选一"

在我们的AI助手平台中，我们同时使用了WebSocket和SSE两种协议，各管各的活儿：

通道协议端点用途
主对话WebSocket/chat/main（MainChatWebsocketServer）主AI助手对话
主对话WebSocketSimpleChatWebsocketServer简洁模式对话
主对话WebSocketPlanChatWebsocketServer教案生成对话
主对话WebSocketMultimodalChatWebsocketServer多模态对话
辅助功能SSEMindMapSseController录音思维导图生成

为什么主对话用WebSocket而不是SSE？核心原因是对话场景需要双向通信——用户可以随时中断生成（点"停止"按钮），这需要客户端主动发消息给服务端。SSE是单向的，做不到。

为什么思维导图用SSE而不是WebSocket？因为它是"触发一次→等结果"的单向推送场景，不需要用户中途交互。SSE比WebSocket轻量得多：不需要维护长连接、不需要心跳、用标准HTTP就能走——一行@GetMapping(produces = TEXT_EVENT_STREAM_VALUE)搞定。

选型结论：需要双向通信（对话、协作）→ WebSocket；只需服务端推送（进度条、流式生成）→ SSE。不要为了"统一"而强行只用一种。

实录2：客户端断了怎么办——SafeSseEmitterRef的clientGone模式

生产环境有一个高频场景：用户触发了思维导图生成，但中途关掉了页面。此时SSE连接断开，如果你的代码直接往SseEmitter.send()里写，会抛IOException。更危险的是，如果异常没处理好，LLM的流式调用也跟着被中断——结果没入库，用户下次打开还得重新生成。

生产项目中，我们封装了一个SafeSseEmitterRef来解决这个问题。核心思路：

// 示意实现：SafeSseEmitterRef.javaclassSafeSseEmitterRef{privatefinalSseEmitteremitter;privatevolatilebooleanclientGone=false;// 关键标记privatevoidsendInternal(StringeventName,StringdataJson){if(clientGone){return;// 客户端已断开，直接跳过，不抛异常}try{emitter.send(SseEmitter.event().name(eventName).data(dataJson));}catch(IOExceptione){clientGone=true;// 标记断连，后续所有send变成no-oplog.warn("SSE send 失败，标记 clientGone，后续不再发送");}catch(IllegalStateExceptione){clientGone=true;// emitter已被complete/timeout}}}

调用方的逻辑也很简单——不管客户端在不在，LLM流都跑完、结果都入库：

// 示意实现：MindMapService.javallmApi.chat(SceneConstant.Scene.DEFAULT_LLM).userInput(userInput).consumer(resp->{Stringdelta=resp.getContent();if(StringUtils.isNotEmpty(delta)){acc.append(delta);// 无论如何都累积ref.sendChunk(delta);// clientGone时自动no-op}}).callStream();// 流结束后：结果入库，不受客户端断连影响recordingMapper.markMindMapCompleted(taskId,acc.toString());log.info("思维导图入库完成, taskId={}, clientGone={}",taskId,ref.isClientGone());

这个模式的本质是把"推送"和"持久化"解耦：推送失败不影响持久化，用户下次打开直接查库拿结果。

实录3：SSE中间层透传——另一个项目的中间层转发架构

我参与的另一个AI助手项目的架构更典型：前端不直接连LLM，中间有一层中间层转发服务。架构如下：

前端 ←─SSE─→ 中间层转发服务 ←─HTTP─→ LLM Gateway ←─→ 大模型

中间层转发服务的ChatService做的是SSE透传：从LLM Gateway读取流式响应，经过处理后逐事件转发给前端：

// 示意实现：ChatService.java（简化）// 1. 从LLM Gateway读取SSE流BufferedReaderreader=newBufferedReader(newInputStreamReader(inputStream,StandardCharsets.UTF_8));SseEventParserparser=newSseEventParser();OutputSafetyFiltersafetyFilter=newOutputSafetyFilter(sensitiveValues,brandName);booleanclientAlive=true;// 2. 逐行读取、过滤、转发while((line=reader.readLine())!=null){fullResponseBuilder.append(line).append('\n');// 无论如何都累积完整响应StringeventData=parser.feedLine(line);if(eventData==null)continue;Stringfiltered=safetyFilter.processEvent(eventData);// 敏感信息过滤if(!clientAlive)continue;// 客户端断了就不推了，但流要读完try{outputStream.write(filtered.getBytes(StandardCharsets.UTF_8));outputStream.flush();}catch(IOExceptionclientAbortEx){clientAlive=false;// 和SafeSseEmitterRef同样的思路log.warn("客户端已断开，继续消费响应完成入库");}}

中间层的价值不只是"转发"，更重要的是可以在流中注入和过滤：

  • 用户画像注入：ChatContextAssembler在请求发给LLM之前，把用户角色、学校类型等上下文拼进prompt
  • 敏感信息过滤：OutputSafetyFilter实时过滤LLM回答中的敏感内容（系统路径、凭据等）
  • 正则安全过滤：StreamingRegexFilter用滑动窗口机制，处理跨token边界的正则匹配——这在流式场景下不是trivial的问题
  • 流式缓存：断线后前端可通过pending接口轮询拿到已生成的内容，实现断线续看

本讲要点

  1. LLM是逐token生成的，流式输出不是优化手段，而是匹配其生成机制的自然选择
  2. SSE协议= HTTP长连接 +text/event-stream+data:行格式，Spring Boot用SseEmitter实现
  3. LangChain4j流式调用：把ChatLanguageModel换成StreamingChatLanguageModel，在onNext回调中逐token推送
  4. 前端用EventSource接收SSE，addEventListener监听自定义事件类型
  5. 生产必须处理客户端断连：用clientGone标记模式，让LLM流跑完、结果入库，推送失败只是静默跳过
  6. SSE vs WebSocket选型：单向推送用SSE，双向通信用WebSocket，不要为了统一强行二选一

动手练习

  1. 基础：跑通本讲的完整demo（Spring Boot + 前端），体验打字机效果
  2. 进阶：给流式接口加一个"停止生成"功能。提示：前端eventSource.close()断开连接后，后端SseEmitter的onCompletion回调会触发——在这里中断LLM调用
  3. 挑战：参考SafeSseEmitterRef的设计，封装一个你自己的SafeSseEmitter——send不抛异常、自动标记断连、提供isClientGone()查询方法

下一讲预告

第4讲：多模型策略——LLM的"负载均衡"——生产环境不会只用一个模型。下一讲我们用Java的策略模式 + YAML配置，实现多模型路由和降级切换。你会发现，这和你做过的支付通道路由、多数据源切换是一模一样的套路。

「Java转AI实战」完整目录（共16讲）

讲次标题
第1讲Java工程师的AI转型地图
第2讲开发环境搭建与第一次LLM调用
第3讲流式输出——LLM的"分页查询"（本文）
第4讲多模型策略——LLM的"负载均衡"
第5讲Prompt Engineering——给LLM写"需求文档"
第6讲RAG全景——为什么LLM需要"开卷考试"
第7讲文档解析——把非结构化数据变成可检索的文本
第8讲智能切分——文档的"分库分表"
第9讲Embedding向量化——文本的"序列化"
第10讲检索策略——从SQL查询到语义搜索
第11讲端到端RAG应用——把前5讲串起来
第12讲从对话到行动——Agent是什么
第13讲MCP协议——AI的标准API网关
第14讲Agent实战——构建一个能执行任务的AI助手
第15讲从Demo到生产——你的Java经验在这里爆发
第16讲转型行动指南——从"会了"到"找到AI工作"

前4讲免费阅读，第5讲起进入RAG核心区和Agent实战，在知识星球持续更新。

获取完整课程

流式输出只是AI应用的"门面功夫"，真正决定AI应用质量的是RAG检索链路——文档怎么切分、向量怎么检索、结果怎么精排。这些都在第6-11讲里，配合真实生产项目的代码逐步拆解。

完整16讲 + 持续更新的AI实战内容，都在知识星球**「Java转AI实战内参」**里：

  • 完整课程16讲，从LLM调用到Agent实战一站式覆盖
  • 真实生产项目的架构设计和代码级拆解
  • Java转AI的面试准备和简历优化
  • 社群答疑，有问必答

星球定价199/年，首年限时优惠99元。

微信搜索**「Java转AI实战内参」**加入星球，或者私信我获取加入方式。

http://www.cnnetsun.cn/news/3185270.html

相关文章：

  • 告别数据分析小白！okbiye AI写作，一键搞定论文数据实证分析
  • 踩坑十多款AIGC检测工具后，我给实验室搭了套全本地的论文内容自查系统
  • Dify_搭建智能销售 Agent 工作流
  • 第38篇：数据组织（一）：数据所有者的角色与责任
  • 第40篇：IT部门的重组：从“系统维护中心”到“能力集成中心”
  • G. Summmon （Codeforces 2241）
  • 华为MetaERP Oracle EBS COA（会计科目表）完整深度解析一、COA 核心设计哲学（财务架构师视角）COA 全称Chart of Accounts，在 EBS 底层由会计关键弹性域
  • 06-运放运用电路-PH测试
  • 别人卷SEO，我用AI吃搜索红利
  • VCG 拉普拉斯Mesh变形（三）
  • 如何永久保存你的网络小说？这款开源下载器支持200+网站
  • 测试零基础抓包前端调试学习路线、线上课程（不会F12/Charles/Fiddler专用）
  • 阶段二：AI手工测试2 Web/APP抓包：网络协议与抓包工具深度解析，测试执行的辅助技能
  • 【2026更新】Python基础学习指南（AI版）——09面向对象
  • 2026 最新 CTF 知识库：Web 漏洞、历年 WP、脚本工具与 AI 知识库
  • Terraform 状态文件深度解析：从原理到远程Backend的6大核心作用
  • CloudFormation Stack 概念全解析：从资源集合到自动化编排的核心引擎
  • 第48期 OpenAI的钱都花哪了？Capex支出全景调研
  • 用 100 行代码手搓一个 MCP Server，让 LLM 直接读你本地文件
  • 一个能优化skill的skill能不能优化它自己？——20 亿 token 烧完，我才看懂 AI 在作弊
  • 小米 NAS 火了：它抢的不是群晖，是你家的数据入口
  • 2026年最新八字排盘应用推荐：天乙八字排盘、命枢、问真八字等怎么选？
  • 《从 300 行到 30 万行代码，我学到的 5 个重构教训》
  • 大模型工具调用安全：从开放 API 到可控 Agent 的权限治理
  • MediaCrawler技术解析：免逆向设计的多平台数据采集实战指南
  • 边缘计算在预测性维护中的应用：数据处理不下云
  • Transformer/BERT 位置编码对比：3种主流方案原理与代码实现差异
  • 5步掌握炉石传说脚本：告别繁琐操作，实现智能自动化对战
  • 有一个很好的思路记录一下，创建一个辅助学习的skill
  • 5分钟找回QQ空间全部历史说说的终极指南：GetQzonehistory帮你永久保存青春记忆