深圳市企业网站建设高要区公路建设规划局网站

深圳市企业网站建设,高要区公路建设规划局网站,沙洋县住房和城乡建设局网站,手机网站 asp一、什么是流式输出#xff1f;流式输出是一种数据传输模式#xff0c;在这种模式下#xff0c;数据不是作为一个完整的、单一的包裹在一次响应中发送给客户端#xff0c;而是被分成许多小的数据块 (chunks) #xff0c;并在服务器端生成的同时#xff0c;持续不断、逐块…一、什么是流式输出流式输出是一种数据传输模式在这种模式下数据不是作为一个完整的、单一的包裹在一次响应中发送给客户端而是被分成许多小的数据块 (chunks)并在服务器端生成的同时持续不断、逐块地推送到客户端。例如下面的Gemini回答过程——二、为什么我们需要流式输出流式输出的核心价值在于改变了用户对延迟的感知将原本漫长的等待转化为即时的内容消费。1.极大地提升用户体验 (UX)在传统的请求-响应模型中用户必须等待服务器生成并返回全部数据才能看到结果。例如一个 AI 回复可能需要 5-10 秒。流式输出则实现了内容逐块、逐字到达和显示。用户感知到的延迟从“总生成时间”缩短为“首个数据块到达时间”。这种实时反馈机制让用户感觉程序在即时响应显著降低了等待的焦虑感。在 AI/LLM 场景中流式输出是必需品它将数秒的枯燥等待变成了持续的阅读过程是用户留存和产品体验的基石。2.提高系统和网络效率流式输出技术尤其是SSE和WebSocket通过建立持久连接减少了重复建立和关闭 HTTP 连接的开销。 数据生成多少推送多少网络带宽得到更有效的利用特别是在处理大量异步或长时间运行的任务时效率优势更为明显。简单来说流式输出的重要性在于它把“等待”变成了“消费”是现代交互式应用和实时数据平台的标配。三、主流的流式输出实现流式输出主要有两种方案第一种是SSE基于HTTP的单向流式输出第二种是基于WebSocket的双向流式输出。1.基于SSE的单向流式输出1后端实现SSE 流式输出的后端代码是有固定的格式和严格的要求。这种格式是Server-Sent Events 规范的核心也是客户端浏览器能够正确解析流的关键。A. 基础格式这是最常用、最核心的格式用于传输数据块。字段格式作用示例Datadata: [内容]包含要发送的实际数据。客户端event.data接收到的就是:后面的内容。data: Hello world\n\n分隔符\n\n至关重要必须以两个换行符标记一个事件块的结束。B. 完整格式SSE 规范还允许其他可选字段用于更复杂的流控制字段格式作用示例Eventevent: [事件名]允许发送不同类型的事件客户端可以通过eventSource.addEventListener(事件名, ...)监听。event: update\nIDid: [唯一标识]允许为每个事件分配一个唯一 ID。如果客户端断线重连它会发送Last-Event-ID服务器可以从该 ID 恢复推送。id: 12345\nRetryretry: [毫秒数]全局设置。客户端断开连接后浏览器等待该毫秒数后再尝试重连。retry: 10000\n\n一个包含所有字段的完整事件块示例如下event: system-update id: 999 retry: 5000 data: {status: ok, progress: 80}注意即使使用了多个字段最后也必须以\n\n结束。const express require(express); const app express(); const PORT 3000; app.get(/api/stream-sse, (req, res) { // ------------------------------------------------------------------ // A. 固定的 HTTP 响应头设置 (关键!) // ------------------------------------------------------------------ res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); // 允许跨域访问 res.setHeader(Access-Control-Allow-Origin, *); // 可选设置断线重连时间间隔单位毫秒 // res.write(retry: 5000\n\n); console.log(--- SSE Connection Established ---); // 模拟数据流式生成 let counter 0; const intervalId setInterval(() { if (counter 8) { // 4. 结束流发送 [DONE] 标记常见实践并关闭连接 res.write(data: [DONE]\n\n); res.end(); clearInterval(intervalId); console.log(--- SSE Stream Ended ---); return; } counter; const payload 这是第 ${counter} 块数据时间${new Date().toLocaleTimeString()}; // ------------------------------------------------------------------ // B. 固定的数据体格式 (严格要求!) // ------------------------------------------------------------------ const sseData // 1. data 字段必须以 data: 开头 data: ${payload}\n // 2. 两个换行符必须以 \n\n 结尾标志一个事件的结束 \n; // 3. 实时写入数据到响应流 res.write(sseData); console.log(Pushed: ${payload}); }, 1500); // 每 1.5 秒推送一次 // 处理客户端断开连接的清理工作 req.on(close, () { clearInterval(intervalId); console.log(Client closed connection.); }); }); app.listen(PORT, () { console.log(SSE Server running on http://localhost:${PORT}); });设置正确的 Header必须设置Content-Type: text/event-stream。实时写入使用 Node.js 的res.write()方法而不是等待数据全部收集完毕后使用res.send()。res.send()会尝试关闭连接不适用于流式输出。遵守data: ...\n\n格式任何偏离这个格式的输出都可能导致客户端EventSource无法正确触发onmessage事件。连接清理监听req.on(close, ...)事件确保在客户端断开连接时服务器能停止不必要的定时器或资源占用。2前端实现A.浏览器原生实现EventSourceEventSource是浏览器专为 SSE 规范设计的高级 API。它自动处理了底层 HTTP 连接、数据格式解析、事件触发以及断线重连等所有复杂逻辑。优势最简洁只需要几行代码即可开始监听流。可靠性高内置自动重连机制非常健壮。无需手动解析自动将data:字段的内容提取出来作为event.data。局限性只能 GET 请求。不支持自定义 Header无法在请求头中传递 Token只能通过 URL query 参数。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleEventSource 客户端示例/title style body { font-family: Segoe UI, Tahoma, Geneva, Verdana, sans-serif; padding: 20px; } h1 { color: #333; } #status { font-weight: bold; margin-bottom: 15px; padding: 8px; border-radius: 4px; } .status-connecting { color: orange; background-color: #fff3e0; } .status-open { color: green; background-color: #e8f5e9; } .status-closed { color: red; background-color: #ffebee; } #output-area { border: 1px solid #ddd; padding: 15px; min-height: 150px; white-space: pre-wrap; /* 保持换行和空格 */ background-color: #fcfcfc; overflow-y: auto; } .chunk { margin-right: 5px; color: #00796b; } .divider { color: #bdbdbd; } /style /head body h1EventSource 实时流接收端/h1 p idstatus classstatus-connecting连接状态: 正在初始化.../p h3接收到的消息流:/h3 div idoutput-area/div button onclickcloseStream() 停止接收流/button script // 请确保这里的 URL 与您的 Node.js 后端 SSE 路由匹配 const STREAM_URL http://localhost:3000/api/stream-sse; const statusElement document.getElementById(status); const outputElement document.getElementById(output-area); let eventSource; function updateStatus(text, className) { statusElement.textContent text; statusElement.className ; statusElement.classList.add(className); } function initializeSSE() { outputElement.innerHTML ; // 清空旧内容 // 1. 创建 EventSource 实例并建立连接 eventSource new EventSource(STREAM_URL); updateStatus(连接状态: 正在连接..., status-connecting); // 2. 监听连接打开事件 eventSource.onopen () { updateStatus(连接状态: ✅ 已建立, status-open); console.log(SSE connection opened successfully.); }; // 3. 监听接收到数据事件 (核心逻辑) eventSource.onmessage (event) { const chunk event.data; console.log(Received chunk:, chunk); // 检查结束标记与后端定义的 [DONE] 匹配 if (chunk [DONE]) { eventSource.close(); updateStatus(连接状态: 流已完成并关闭, status-closed); console.log(Stream finished and closed by server.); return; } // 实时追加数据到 UI // 使用 innerHTML 追加可以实现更丰富的样式 outputElement.innerHTML span classchunk${chunk}/spanspan classdivider | /span; // 确保滚动到底部以查看最新内容 outputElement.scrollTop outputElement.scrollHeight; }; // 4. 监听错误事件处理断线等 eventSource.onerror (error) { console.error(EventSource encountered an error:, error); if (eventSource.readyState EventSource.CLOSED) { updateStatus(连接状态: ❌ 已关闭或断开, status-closed); } else { // EventSource 默认会尝试自动重连 updateStatus(连接状态: ⚠️ 发生错误正在尝试重连..., status-connecting); } }; } // 5. 客户端主动关闭流的函数 function closeStream() { if (eventSource eventSource.readyState ! EventSource.CLOSED) { eventSource.close(); updateStatus(连接状态: 用户手动关闭, status-closed); console.log(User manually closed the stream.); } } // 页面加载完成后立即启动 SSE window.onload initializeSSE; /script /body /html阶段一连接初始化与建立此阶段旨在建立一个持久化的 HTTP 连接并开始监听。步骤动作描述核心代码关键点1. 创建EventSource实例使用EventSource构造函数传入 SSE 接口 URL向服务器发起连接请求。const es new EventSource(url);浏览器自动处理底层 HTTP GET 请求。2. 监听连接打开监听onopen事件确认与服务器的连接已成功建立。es.onopen () { ... }此时流式数据传输通道已打开。3. 更新 UI 状态在onopen中更新界面状态提示用户数据流已开始。setStatus(已连接);提升用户体验。阶段二数据接收与处理核心此阶段是持续接收服务器推送的数据并实时更新 UI。步骤动作描述核心代码关键点4. 监听消息事件监听onmessage事件。每当服务器推送一个完整的data:...\n\n事件块时此回调函数就会触发。es.onmessage (event) { ... }event.data包含了服务器推送的实际内容。5. 实时数据追加将接收到的数据 (event.data)追加到当前展示的文本末尾而不是替换。currentMsg event.data;这是实现“打字机”效果的关键。6. 处理自定义事件如果后端使用了event: [name]字段前端可以通过es.addEventListener(name, callback)针对性地处理不同类型的事件。es.addEventListener(update, ...)适用于需要区分不同业务类型数据的场景。阶段三连接维护、关闭与错误处理此阶段处理流的正常结束、网络错误和重试逻辑。步骤动作描述核心代码关键点7. 监听流结束标记根据与后端约定的结束标记如[DONE]判断流是否完成。if (event.data [DONE]) { ... }流的正常结束避免无限等待。8. 关闭连接当流结束或用户主动点击“停止”按钮时主动调用close()方法。es.close();释放客户端资源。9. 监听错误与重连监听onerror事件。当连接出错时浏览器会根据服务器设置的retry:时间间隔自动尝试重连。es.onerror (error) { ... }自动重试是 SSE 相比于手动fetch的巨大优势。B.Fetch API实现当需要更强大的控制力、自定义请求头或处理非标准 SSE 格式的流时fetch是最佳选择。利用fetch返回的response.body它是一个ReadableStream对象。开发者需要获取这个流的Reader进入一个循环手动读取数据块、解码并根据\n\n规则手动解析SSE 格式。优势控制力强可以发送 POST、PUT 等请求并设置自定义 Header用于鉴权。通用性可以处理任何基于 HTTP 的流式数据不限于 SSE 格式。新标准符合现代 Web 标准浏览器支持良好。挑战实现复杂需要手动编写数据解析和错误重连的代码。性能考量频繁的循环读取和字符串拼接/解码操作需要谨慎优化。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleFetch API SSE 客户端示例/title style body { font-family: Segoe UI, Tahoma, Geneva, Verdana, sans-serif; padding: 20px; } #status { font-weight: bold; margin-bottom: 15px; color: orange; } #output-area { border: 1px solid #ddd; padding: 15px; min-height: 150px; white-space: pre-wrap; background-color: #fcfcfc; overflow-y: auto; } /style /head body h1 Fetch API 模拟 SSE 接收端/h1 p idstatus连接状态: 待启动/p h3接收到的消息流:/h3 div idoutput-area/div button onclickstartStream()▶️ 启动流/button button onclickstopStream() 停止接收流/button script const STREAM_URL http://localhost:3000/api/stream-sse; // 确保 URL 正确 const statusElement document.getElementById(status); const outputElement document.getElementById(output-area); let controller null; // 用于控制请求中止 (AbortController) function updateStatus(text, color) { statusElement.textContent text; statusElement.style.color color; } async function startStream() { if (controller) stopStream(); // 确保前一个流已停止 controller new AbortController(); const signal controller.signal; let textDecoder new TextDecoder(utf-8); let buffer ; // 缓冲区用于存储不完整的事件块 outputElement.innerHTML ; updateStatus(连接状态: 正在连接..., orange); try { // 1. 发起 Fetch 请求设置 signal 用于中止 const response await fetch(STREAM_URL, { method: GET, headers: { Accept: text/event-stream }, signal: signal }); if (!response.body) { throw new Error(响应体不是一个可读流。); } // 2. 获取 ReadableStream 的 Reader const reader response.body.getReader(); // 3. 循环读取流数据 while (true) { const { done, value } await reader.read(); if (done) { updateStatus(连接状态: 流已完成, green); console.log(Stream finished.); break; } // 4. 解码字节数据并追加到缓冲区 // { stream: true } 允许在流继续时进行解码 buffer textDecoder.decode(value, { stream: true }); // 5. 手动解析 SSE 消息块 // SSE 消息以 \n\n 结束分割缓冲区 let messages buffer.split(\n\n); buffer messages.pop(); // 最后一个不完整的块留在缓冲区 messages.forEach(message { if (message) { // 6. 提取 data: 字段内容 const dataMatch message.match(/data: (.*)/); if (dataMatch dataMatch[1]) { const data dataMatch[1].trim(); if (data [DONE]) { controller.abort(); // 接收到结束标记主动中止请求 } else { // 7. 处理接收到的数据 outputElement.innerHTML span classchunk${data}/spanspan classdivider | /span; outputElement.scrollTop outputElement.scrollHeight; } } } }); } } catch (error) { if (error.name AbortError) { updateStatus(连接状态: 流被用户/DONE标记中止, gray); console.log(Fetch request aborted successfully.); } else { updateStatus(连接状态: ❌ 错误: ${error.message}, red); console.error(Fetch SSE 错误:, error); } } } function stopStream() { if (controller) { controller.abort(); controller null; } } // 页面加载完成后设置初始状态 window.onload () updateStatus(连接状态: 待启动, black); /script /body /htmlFetch API 实现流式输出是高度可定制但相对复杂的技术方案它要求开发者手动处理数据的接收、解码和 SSE 格式的解析。整个流程可以清晰地划分为三个主要阶段。阶段一初始化请求与准备工作此阶段的目标是发起请求并设置流读取所需的环境。步骤动作描述关键技术点1. 准备控制器创建AbortController实例。用于在流结束或需要中断时能够中止fetch请求。new AbortController()2. 发起 Fetch 请求发起fetch请求将controller.signal附加到请求配置中。可在headers中设置鉴权信息。fetch(url, { signal: controller.signal, headers: {...} })3. 获取流读取器检查响应是否成功 (response.ok)然后通过response.body.getReader()获取reader对象。response.body.getReader()4. 准备解码器与缓冲区实例化TextDecoder用于将字节解码为字符串并初始化一个buffer变量用于暂存不完整的数据片段。new TextDecoder(utf-8),let buffer 阶段二循环读取、解码与解析核心此阶段是整个流程最复杂的部分需要持续从流中拉取数据并手动解析 SSE 格式。步骤动作描述关键技术点5. 启动读取循环使用while (true)或类似的循环结构开始持续读取数据。while (true)6. 读取数据块调用await reader.read()。它会暂停执行等待服务器推送新的数据块 (value: Uint8Array)。const { done, value } await reader.read()7. 判断流结束检查done属性。如果为true跳出循环进入清理阶段。if (done) break;8. 解码与缓冲使用TextDecoder将原始字节value解码为字符串并将其追加到缓冲区 (buffer) 中。buffer decoder.decode(value, { stream: true })9. 手动解析 SSE使用buffer.split(\n\n)将缓冲区内容分割成潜在的 SSE 消息数组。然后将数组中最后一个不完整的块重新放回缓冲区。buffer.split(\n\n),buffer messages.pop()10. 提取数据遍历其余完整的消息块使用正则表达式如/data: (.*)/手动提取data:字段后的实际内容。message.match(/data: (.*)/)11. 实时处理将提取到的数据追加到 UI 界面并执行所需的业务逻辑。UI 更新、滚动到底部阶段三连接清理与错误处理此阶段确保流的正确终止和资源释放。步骤动作描述关键技术点12. 处理结束标记在数据处理逻辑中如果检测到服务器推送的结束标记如[DONE]则立即调用controller.abort()终止流。controller.abort()13. 错误捕获将整个fetch和流读取循环包裹在try...catch块中。捕获网络错误和因controller.abort()产生的AbortError。try...catch (error)14. 释放资源无论是正常结束还是出错确保所有相关资源如reader得到释放。AbortController自动处理了请求的终止。通过这种手动控制的方式Fetch API 提供了对流的最高级别控制但代价是代码实现更为复杂且需要开发者手动处理自动重连等健壮性机制。C.Axios API实现使用axios实现流式输出主要依赖其配置项但在浏览器环境下的表现不如fetch和EventSource。在axios的配置中需要显式设置responseType: stream。在Node.js环境中这会返回一个 Node.js 的流处理起来比较方便。但在浏览器环境中axios对流的封装不如fetch原生。优势Node.js 友好在 Node.js 环境中处理流式数据很方便。代码统一如果应用大量依赖axios可以避免引入fetch。挑战浏览器兼容性/稳定性浏览器端对axios流的稳定支持不如fetch和EventSource。解析依然需要手动像fetch一样您仍然需要获取流并手动进行 SSE 格式的解析和解码。潜在的内存问题如果流处理不当可能会导致内存占用过高。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleAxios 客户端模拟 SSE 示例/title script srchttps://cdn.jsdelivr.net/npm/axios/dist/axios.min.js/script style body { font-family: Segoe UI, Tahoma, Geneva, Verdana, sans-serif; padding: 20px; } #status { font-weight: bold; margin-bottom: 15px; color: orange; } #output-area { border: 1px solid #ddd; padding: 15px; min-height: 150px; white-space: pre-wrap; background-color: #fcfcfc; overflow-y: auto; } /style /head body h1️ Axios 模拟 SSE 接收端/h1 p idstatus连接状态: 待启动/p h3接收到的消息流:/h3 div idoutput-area/div button onclickstartStream()▶️ 启动流/button button onclickstopStream() 停止接收流/button script const STREAM_URL http://localhost:3000/api/stream-sse; // 确保 URL 正确 const statusElement document.getElementById(status); const outputElement document.getElementById(output-area); let cancelTokenSource null; // 用于取消请求的 Axios Token let lastProcessedIndex 0; // 追踪上次处理到的数据索引 let buffer ; // 用于存放不完整的 SSE 事件块 function updateStatus(text, color) { statusElement.textContent text; statusElement.style.color color; } async function startStream() { if (cancelTokenSource) stopStream(); cancelTokenSource axios.CancelToken.source(); lastProcessedIndex 0; buffer ; outputElement.innerHTML ; updateStatus(连接状态: 正在连接..., orange); try { // 1. 发起 Axios GET 请求 await axios.get(STREAM_URL, { // 2. 设置 responseType 为 stream 或让其默认 (text) // 在浏览器中这依赖于 XHR 的 progress 事件 responseType: text, // 3. 设置取消 Token cancelToken: cancelTokenSource.token, // 4. 核心监听下载进度事件 onDownloadProgress: (progressEvent) { // progressEvent.currentTarget 是底层的 XMLHttpRequest 对象 const responseText progressEvent.currentTarget.responseText; // 5. 提取新增的数据块 const newChunk responseText.substring(lastProcessedIndex); lastProcessedIndex responseText.length; // 6. 追加到缓冲区 buffer newChunk; // 7. 手动解析 SSE 消息块 let messages buffer.split(\n\n); // 最后一个片段可能是未完成的事件保留在缓冲区 buffer messages.pop(); messages.forEach(message { if (message.trim()) { // 8. 提取 data: 字段内容 const dataMatch message.match(/data: (.*)/); if (dataMatch dataMatch[1]) { const data dataMatch[1].trim(); // 9. 检查结束标记与后端定义的 [DONE] 匹配 if (data [DONE]) { stopStream(true); // 接收到结束标记主动中止请求 } else { // 10. 实时追加数据到 UI outputElement.innerHTML span classchunk${data}/spanspan classdivider | /span; outputElement.scrollTop outputElement.scrollHeight; } } } }); } }); // 请求正常结束后端在发送 [DONE] 后关闭连接但 Axios 仍可能等待超时 // 推荐依靠 [DONE] 来调用 stopStream if (!cancelTokenSource) { updateStatus(连接状态: 流已完成 (Axios OK), green); } } catch (error) { if (axios.isCancel(error)) { // 被用户或 [DONE] 标记取消 updateStatus(连接状态: 流已中止, gray); console.log(Axios request manually aborted.); } else { // 其他网络错误或请求失败 updateStatus(连接状态: ❌ 错误: ${error.message}, red); console.error(Axios SSE 错误:, error); } } } // 客户端主动关闭流的函数 function stopStream(isDone false) { if (cancelTokenSource) { cancelTokenSource.cancel(Stream terminated.); cancelTokenSource null; if (isDone) { updateStatus(连接状态: 流已完成并关闭, green); } } } window.onload () updateStatus(连接状态: 待启动, black); /script /body /htmlAxios 并非为 SSE 设计的因为它基于传统的 XMLHttpRequest (XHR) 模型或者在 Node.js 中基于http模块。在浏览器环境中Axios 没有原生EventSource或ReadableStream的支持。因此使用 Axios 实现流式输出的标准工作流程是利用 XHR 的onprogress事件监听机制在 Axios 中表现为onDownloadProgress回调手动获取每次新增的数据块并手动进行 SSE 格式的解析。没有自动解析必须手动解析data:字段和\n\n分隔符。手动追踪数据每次onDownloadProgress触发时返回的是当前已接收的全部数据而不是新增的数据。您需要一个指针来追踪上次处理到了哪里并提取出新的数据块。不支持自动重连必须手动实现错误检测和重连逻辑本示例不包含重连但需注意。流式输出技术的选择1. EventSource API这是实现 SSE 的标准和首选方案。实现复杂度最低。浏览器原生支持代码量最少。核心优势内置健壮性。自动处理 SSE 数据格式解析、消息事件 (onmessage) 触发并内置了断线重连机制支持服务器设置retry:间隔。主要局限只能 GET 请求。不支持自定义 Header 鉴权。只能通过 URL 参数传递 Token安全性相对较低。选择原则强烈推荐用于大多数场景尤其是在安全性要求不涉及请求 Header例如使用 Session Cookie 或 URL Token 鉴权且只需要单向数据推送时。2. Fetch API ReadableStream这是在需要高控制度时使用的现代标准方案。实现复杂度中高。需要手动获取流阅读器 (getReader())并进入循环手动进行 SSE 数据解析解码字节、查找\n\n、提取data:。核心优势控制力最强。支持自定义 HTTP Header适用于传递 JWT 等鉴权信息。支持 POST 等其他 HTTP 方法。可以处理非标准的流格式。主要局限必须手动编写复杂的 SSE 格式解析逻辑和错误重连机制。选择原则当您必须在请求 Header 中传递鉴权信息例如 JWT或者需要处理非标准的流格式时应选择此方案。3. Axios onDownloadProgress这是基于旧的 XHR 机制的模拟方案通常不推荐用于现代 Web 应用的 SSE。实现复杂度高。依赖onDownloadProgress事件每次事件触发返回的是全部已接收数据因此需要额外的逻辑来追踪上次处理的索引以提取新增的数据块。核心优势如果您的项目大量依赖 Axios可以保持技术栈的统一。主要局限实现流程复杂且易错。性能不如原生 API且不支持原生的自动重连。在浏览器环境中缺乏对流的标准化支持。选择原则仅在需要兼容旧环境或因特殊限制无法使用EventSource或fetch的情况下考虑。2.基于WebSocket的双向流式输出实现WebSocket 协议与 SSE单向流和传统的 HTTP 请求一次性传输有着本质的区别它在客户端和服务器之间建立了一个持久的、全双工双向的通信通道非常适合需要高频率、低延迟交互的场景。1. 协议升级WebSocket 连接的建立始于一个标准的 HTTP 请求这个请求包含了特殊的 Header用于请求将连接从 HTTP升级Upgrade到 WebSocket 协议。一旦升级成功连接就不再受限于传统的 HTTP 请求-响应模型。2. 全双工通信一旦连接建立服务器和客户端可以独立、同时地互相发送数据帧。客户端 →→ 服务器客户端可以随时发送输入、控制命令或心跳包。服务器 →→ 客户端服务器可以随时发送流式数据、状态更新或响应。这种双向性使得 WebSocket 不仅可以用于流式输出服务器推数据还可以用于实时接收用户输入完美支持实时聊天或协作编辑等场景。3. 基于帧的传输WebSocket 的数据传输是基于帧的而不是基于文本流或 HTTP 请求。这使得传输效率更高延迟更低。帧可以包含文本数据Text Frame或二进制数据Binary Frame。基于 WebSocket 的双向流式输出需要前端和后端分别使用专门的库和 API 进行实现。1. 后端实现 (Node.js/ws 或 Socket.IO)后端需要一个专门的 WebSocket 服务器来管理连接和发送数据帧。步骤动作描述关键技术点1. 握手与连接监听 HTTP 升级请求接受连接并分配一个唯一的 WebSocket 连接对象。ws.on(connection, (socket) { ... })2. 接收客户端输入监听客户端发送过来的数据帧用于触发 AI 任务或控制流。socket.on(message, (input) { // Process input })3. 流式生成与推送在 AI 模型生成数据的过程中将数据块封装成 WebSocket 帧并实时发送。socket.send(data_chunk)4. 维护连接维护连接状态处理心跳包Ping/Pong并在客户端断开时清理资源。socket.on(close, ...)const WebSocket require(ws); // 创建 WebSocket 服务器监听 8080 端口 const wss new WebSocket.Server({ port: 8080 }); console.log(WebSocket Server running on ws://localhost:8080); // 监听客户端连接事件 wss.on(connection, function connection(ws) { console.log(--- Client Connected ---); // 1. 监听客户端发来的消息双向输入 ws.on(message, function incoming(message) { const clientMessage message.toString(); console.log(Received message from client: ${clientMessage}); try { const data JSON.parse(clientMessage); if (data.command START) { // 客户端请求开始流式输出 startStreaming(ws, data.prompt); } else if (data.command STOP) { // 客户端请求停止流式输出 (实时控制) stopStreaming(ws); ws.send(JSON.stringify({ status: info, message: Streaming stopped by client. })); } } catch (e) { console.error(Error parsing client message:, e); ws.send(JSON.stringify({ status: error, message: Invalid JSON format. })); } }); // 监听连接关闭事件 ws.on(close, function close() { console.log(--- Client Disconnected ---); // 清理资源停止该连接上的所有流 stopStreaming(ws); }); // 初始问候 ws.send(JSON.stringify({ status: ready, message: Welcome! Send {command: START, prompt: ...} to begin streaming. })); }); // 存储当前正在流式输出的连接和定时器 const activeStreams new Map(); /** * 模拟 AI 模型流式输出的函数 * param {WebSocket} ws - 当前连接的 WebSocket 实例 * param {string} prompt - 客户端提供的输入提示 */ function startStreaming(ws, prompt) { if (activeStreams.has(ws)) { ws.send(JSON.stringify({ status: info, message: Stream is already active. })); return; } let counter 0; const intervalId setInterval(() { counter; const chunk [Chunk ${counter}] Output for ${prompt.substring(0, 15)}...: Data block ${counter}.; // 2. 实时将数据块作为 WebSocket 帧发送给客户端 // 使用 JSON 格式封装数据方便客户端解析 const dataToSend JSON.stringify({ type: data, content: chunk, count: counter }); // 检查连接是否仍然打开 if (ws.readyState WebSocket.OPEN) { ws.send(dataToSend); console.log(Pushed data chunk ${counter}); } else { // 如果连接关闭停止流 clearInterval(intervalId); activeStreams.delete(ws); } if (counter 15) { // 3. 流结束发送结束标记 clearInterval(intervalId); activeStreams.delete(ws); if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: done, message: Streaming complete. })); } console.log(--- Streaming finished ---); } }, 500); // 每 500 毫秒发送一个数据块 activeStreams.set(ws, intervalId); } /** * 停止指定连接的流式输出 */ function stopStreaming(ws) { if (activeStreams.has(ws)) { clearInterval(activeStreams.get(ws)); activeStreams.delete(ws); } }2. 前端实现 (Browser WebSocket API)前端使用浏览器原生的WebSocketAPI 来建立连接和处理双向通信。步骤动作描述关键技术点1. 建立连接使用new WebSocket(ws://url)建立连接。注意协议是ws://或wss://。new WebSocket(ws://localhost:8080)2. 监听连接状态监听onopen连接成功、onerror和onclose事件。ws.onopen () {...}3. 接收流式输出监听onmessage事件接收服务器推送的流式数据帧并实时追加到 UI。ws.onmessage (event) { // Append event.data }4. 发送客户端输入当用户有新输入或需要发送控制命令时通过send()方法发送数据帧到服务器。ws.send(JSON.stringify({command: stop}))!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleWebSocket 双向流式输出/title style body { font-family: Arial, sans-serif; padding: 20px; } #status { font-weight: bold; margin-bottom: 15px; } #output-area { border: 1px solid #ccc; padding: 15px; min-height: 150px; background-color: #f9f9f9; white-space: pre-wrap; overflow-y: scroll; } .control-panel { margin-top: 20px; } /style /head body h1 WebSocket 双向流式交互/h1 p idstatus stylecolor: gray;连接状态: 待连接.../p h3服务器输出流:/h3 div idoutput-area/div div classcontrol-panel label forprompt输入提示:/label input typetext idprompt value请给我写一篇关于AI流式输出的文章 stylewidth: 300px; button onclickconnectWebSocket()建立连接/button button onclickstartStream()▶️ 启动流式输出/button button onclickstopStream() 实时停止流/button button onclickcloseWebSocket()断开连接/button /div script const WS_URL ws://localhost:8080; let ws null; const statusElement document.getElementById(status); const outputElement document.getElementById(output-area); const promptInput document.getElementById(prompt); function log(message) { outputElement.innerHTML p stylemargin: 0; padding: 2px 0;${message}/p; outputElement.scrollTop outputElement.scrollHeight; } function updateStatus(text, color) { statusElement.textContent text; statusElement.style.color color; } // 1. 建立 WebSocket 连接 function connectWebSocket() { if (ws ws.readyState WebSocket.OPEN) { updateStatus(连接已存在, blue); return; } ws new WebSocket(WS_URL); updateStatus(正在连接..., orange); outputElement.innerHTML ; // 监听连接成功事件 ws.onopen () { updateStatus(连接状态: ✅ 已建立, green); log(--- WebSocket Connection Opened ---); }; // 2. 监听接收到的数据帧 (核心) ws.onmessage (event) { try { const data JSON.parse(event.data); if (data.type data) { // 实时追加流式输出的内容 log([Server Stream]: ${data.content}); } else if (data.type done) { log([INFO]: ${data.message}); updateStatus(流已完成, blue); } else if (data.status info) { log([INFO]: ${data.message}); } } catch (e) { // 处理非 JSON 格式的消息 log([RAW MESSAGE]: ${event.data}); } }; // 监听错误事件 ws.onerror (error) { updateStatus(连接状态: ❌ 发生错误, red); console.error(WebSocket Error:, error); }; // 监听连接关闭事件 (连接断开) ws.onclose () { updateStatus(连接状态: 已断开, red); log(--- WebSocket Connection Closed ---); ws null; }; } // 3. 客户端发送命令启动流式输出 function startStream() { if (ws ws.readyState WebSocket.OPEN) { const prompt promptInput.value || Default prompt; const command { command: START, prompt: prompt }; // 发送 JSON 格式的启动命令 ws.send(JSON.stringify(command)); updateStatus(流式输出已启动..., purple); log([Client Command]: Starting stream for prompt: ${prompt}); } else { updateStatus(请先建立连接, red); } } // 4. 客户端发送命令实时停止流 function stopStream() { if (ws ws.readyState WebSocket.OPEN) { const command { command: STOP }; // 发送 JSON 格式的停止命令 ws.send(JSON.stringify(command)); updateStatus(发送停止命令, purple); log([Client Command]: Sending STOP command.); } else { updateStatus(请先建立连接, red); } } function closeWebSocket() { if (ws ws.readyState WebSocket.OPEN) { ws.close(); } } // 页面加载后自动尝试连接 window.onload connectWebSocket; /script /body /html特性WebSocket 双向流SSE 单向流通信方向双向全双工单向服务器 →→ 客户端协议基础独立于 HTTP 的 TCP 协议基于 HTTP 协议适用场景实时聊天、协作编辑、AI 实时交互控制纯粹的 AI 流式输出、新闻推送实现复杂度较高需独立服务器支持较低基于标准 HTTP自动重连无原生支持需要手动实现浏览器原生支持如果你的 AI 应用只需要将 LLM 的结果流式输出给用户SSE 是更简单、更健壮的选择。但如果您的应用需要用户在流式输出过程中实时中断、修改输入、或者实现多人协作WebSocket 是唯一的选择。