赞
踩
一、背景
随着现代技术的快速发展,即时交互变得越来越重要。用户不仅希望获取信息,而且希望以更直观和实时的方式体验它。这在聊天应用程序和其他实时通信工具中尤为明显,用户习惯看到对方正在输入的提示。
ChatGPT,作为 OpenAI 的代表性产品之一,不仅为用户提供了强大的自然语言处理能力,而且关注用户的整体交互体验。在使用 ChatGPT 进行交互时,用户可能已经注意到了一个细节:当它产生回复时,回复会像人类逐字输入的方式逐渐出现,而不是一次性显示完整答案。
这种打字效果给人一种仿佛与真人对话的感觉,进一步增强了其自然语言处理的真实感。一开始,许多开发者可能会误以为这是通过 WebSockets 实现的,这是因为 WebSockets 是一种常用于实时通信的技术。然而,仔细研究后,我们发现 ChatGPT 使用了一种不同的技术:基于 EventStream 的方法。更具体地说,它似乎是通过 SSE (Server-Sent Events) 来实现逐个字地推送答案的。
此外,考虑到 ChatGPT 的复杂性和其涉及的大量计算,响应时间可能会长于其他基于数据库的简单查询。因此,采用 SSE 逐步推送结果的方式可以帮助减少用户感到的等待时间,从而增强用户体验。
二、SSE 简介
Server-Sent Events(通常简称为 SSE)是一种允许服务器向Web页面发送实时更新的技术。与WebSocket技术相比,SSE专门设计用于从服务器到客户端的单向通信。这种单向性使其在某些场景中更为简单和直观。
单向通信:SSE 专为从服务器到客户端的单向通信设计。客户端不能通过SSE直接发送数据到服务器,但可以通过其他方法如AJAX与服务器进行交互。
基于HTTP:SSE 基于 HTTP 协议运行,不需要新的协议或端口。这使得它能够轻松地在现有的Web应用架构中使用,并且通过标准的HTTP代理和中间件进行支持。
自动重连:如果连接断开,浏览器会自动尝试重新连接到服务器。
格式简单:SSE 使用简单的文本格式发送消息,每个消息都以两个连续的换行符分隔。
原生浏览器支持:许多现代浏览器(如 Chrome、Firefox 和 Safari)已原生支持SSE,但需要注意的是,某些浏览器,如Internet Explorer和早期的Edge版本,不支持SSE。
虽然 SSE 与 WebSockets 在某种程度上有些相似,但它们之间还存在一些关键差异,如下所示:
对比项 | Server-Sent Events (SSE) | WebSockets |
基于协议 | 基于 HTTP,简化了连接和交互的过程 | 通常基于 WS/WSS(基于TCP),更为灵活 |
通信能力 | 单向通信:仅服务器向客户端发送消息 | 双向通信能力 |
配置 | 配置简单,易于理解和使用 | 需要更复杂的配置和理解 |
断线与消息追踪 | 自带的断线重连和消息跟踪功能 | 通常需要手动处理或使用额外库 |
数据格式 | 通常为文本,但可以发送经过编码/压缩的二进制消息 | 支持文本和原始二进制消息 |
事件处理 | 支持多种自定义事件 | 基本消息机制,不能像SSE那样自定义事件类型 |
连接并发性 | 连接数可能受到 HTTP 版本的限制,尤其是在HTTP/1.1中 | WebSocket被设计为支持更高的连接并发性 |
安全性 | 仅支持HTTP和HTTPS的安全机制 | 支持WS和WSS,可以在WSS上实现更强大的加密 |
浏览器兼容性 | 大部分现代浏览器支持,但不是所有浏览器 | 几乎所有现代浏览器都支持 |
开销 | 由于基于HTTP,每次消息可能有较大的头部开销 | 握手后,消息头部开销相对较小 |
三、服务端深入解析
Server-Sent Events(SSE)是一个基于 HTTP 的协议,允许服务器单向地向浏览器推送信息。为了成功地使用 SSE,服务器和客户端都必须遵循一定的规范和流程。
当客户端(例如浏览器)发出请求订阅 SSE 服务时,服务器需要通过设置特定的响应头部信息来确认该请求。这些头部信息包括:
Content-Type: text/event-stream: 这表示返回的内容为事件流。
Cache-Control: no-cache: 这确保服务器推送的消息不会被缓存,以保障消息的实时性。
Connection: keep-alive: 这指示连接应始终保持开放,以便服务器可以随时发送消息。
SSE 使用简单的文本格式来组织和发送消息。基本的消息结构是由一系列行组成,每一行由字段名、一个冒号和字段值组成。
以下是消息中可以使用的一些字段及其用途:
event: 定义了事件的类型。这可以帮助客户端确定如何处理接收到的消息。
id: 提供事件的唯一标识符。如果连接中断,客户端可以使用最后收到的事件 ID 来请求服务器从某个点重新发送消息。
retry: 指定了当连接断开时,客户端应等待多少毫秒再尝试重新连接。这为连接中断和重连提供了一种机制。
data: 这是消息的主体内容。它可以是任何 UTF-8 编码的文本,而且可以跨多行。每行数据都会在客户端解析时连接起来,中间使用换行符分隔。
为了确保消息的正确和完整传输,服务器通常在消息的末尾添加一个空行,表示消息的结束。
示例:
- id: 123
- event: update
- data: {"message": "This is a test message"}
此外,SSE 也支持多条连续消息的发送。只要每条消息之间使用两个换行符隔开即可。
四、客户端实践
接入 SSE 并不困难,尤其在客户端这边。主流浏览器提供了 EventSource API,使得与 SSE 服务端建立和维护连接变得异常简单。
首先,需要创建一个 EventSource 对象,它将代表与服务器的持久连接。初始化时,可以为它提供一些选项,以满足特定需求。
- const options = {
- withCredentials: true // 允许跨域请求携带凭证
- };
-
-
- // 创建一个 EventSource 对象以开始监听
- const eventSource = new EventSource('your_server_url', options);
在上面的代码中,withCredentials 参数用于指示是否应该在请求中发送凭证(例如 cookies)。这在跨域场景中可能会非常有用。
一旦与服务器建立了连接,就可以开始监听从服务器发送过来的事件。
通用事件处理:
默认情况下,EventSource 对象会对三种基本的事件类型进行响应:open、message 和 error。可以设置对应的处理函数来对它们进行响应。
- // 监听连接打开事件
- eventSource.onopen = function(event) {
- console.log('Connection to SSE server established!');
- };
-
-
- // 监听标准消息事件
- eventSource.onmessage = function(event) {
- console.log('Received data from server: ', event.data);
- };
-
-
- // 监听错误事件
- eventSource.onerror = function(event) {
- console.error('An error occurred while receiving data:', event);
- };
自定义事件处理:
除了上述的基本事件外,服务器还可能发送自定义的事件类型。为了处理这些事件,需要使用 addEventListener() 方法。
- // 监听一个名为 "update" 的自定义事件
- eventSource.addEventListener('update', function(event) {
- console.log('Received update event:', event.data);
- });
如果不再需要从服务器接收事件,可以使用 close 方法关闭连接。
eventSource.close();
关闭连接后,将不再接收任何事件,除非再次初始化 EventSource 对象。
总结:使用 EventSource API,客户端可以方便地与 SSE 服务器交互,从而实时接收数据更新。这为创建响应迅速的 web 应用提供了极大的便利,同时避免了传统的轮询方式带来的资源浪费。
五、理论实践
- const http = require('http');
- const fs = require('fs');
-
-
- // 初始化 HTTP 服务器
- http.createServer((req, res) => {
-
-
- // 为了简洁,将响应方法抽离成函数
- function serveFile(filePath, contentType) {
- fs.readFile(filePath, (err, data) => {
- if (err) {
- res.writeHead(500);
- res.end('Error loading the file');
- } else {
- res.writeHead(200, {'Content-Type': contentType});
- res.end(data);
- }
- });
- }
-
-
- function handleSSEConnection() {
- res.writeHead(200, {
- 'Content-Type': 'text/event-stream',
- 'Cache-Control': 'no-cache',
- 'Connection': 'keep-alive'
- });
-
-
- let id = 0;
- const intervalId = setInterval(() => {
- const message = {
- event: 'customEvent',
- id: id++,
- retry: 30000,
- data: { id, time: new Date().toISOString() }
- };
- for (let key in message) {
- if (key !== 'data') {
- res.write(`${key}: ${message[key]}\n`);
- } else {
- res.write(`data: ${JSON.stringify(message.data)}\n\n`);
- }
- }
- }, 1000);
-
-
- req.on('close', () => {
- clearInterval(intervalId);
- res.end();
- });
- }
-
-
- switch (req.url) {
- case '/':
- serveFile('index.html', 'text/html');
- break;
- case '/events':
- handleSSEConnection();
- break;
- default:
- res.writeHead(404);
- res.end();
- break;
- }
-
-
- }).listen(3000);
-
-
- console.log('Server listening on port 3000');
- <!DOCTYPE html>
- <html lang="en">
- <head>
- <meta charset="UTF-8">
- <meta http-equiv="X-UA-Compatible" content="IE=edge">
- <meta name="viewport" content="width=device-width, initial-scale=1.0">
- <title>SSE Demo</title>
- </head>
- <body>
- <h1>SSE Demo</h1>
- <button onclick="connectSSE()">建立 SSE 连接</button>
- <button onclick="closeSSE()">断开 SSE 连接</button>
- <br /><br />
- <div id="message"></div>
-
-
- <script>
- const messageElement = document.getElementById('message');
- let eventSource;
-
-
- // 连接 SSE
- function connectSSE() {
- eventSource = new EventSource('/events');
-
-
- eventSource.addEventListener('customEvent', handleReceivedMessage);
- eventSource.onopen = handleConnectionOpen;
- eventSource.onerror = handleConnectionError;
- }
-
-
- // 断开 SSE 连接
- function closeSSE() {
- eventSource.close();
- appendMessage(`SSE 连接关闭,状态${eventSource.readyState}`);
- }
-
-
- // 处理从服务端收到的消息
- function handleReceivedMessage(event) {
- const data = JSON.parse(event.data);
- appendMessage(`${data.id} --- ${data.time}`);
- }
-
-
- // 连接建立成功的处理函数
- function handleConnectionOpen() {
- appendMessage(`SSE 连接成功,状态${eventSource.readyState}`);
- }
-
-
- // 连接发生错误的处理函数
- function handleConnectionError() {
- appendMessage(`SSE 连接错误,状态${eventSource.readyState}`);
- }
-
-
- // 将消息添加到页面上
- function appendMessage(message) {
- messageElement.innerHTML += `${message}<br />`;
- }
- </script>
- </body>
- </html>
将上面的两份代码保存为 server.js 和 index.html,并在命令行中执行 node server.js 启动服务端,然后在浏览器中打开 http://localhost:3000 即可看到 SSE 效果。
六、业务实践
在业务真实使用场景中,基于SSE的方法存在一些问题和限制:
默认请求仅支持 GET 方法。当前端需要向后端传递参数时,参数只能拼接在请求的 URL 上,对于复杂的业务场景来说实现较为麻烦。
对于服务端返回的数据格式有固定要求,必须按照 event、id、retry、data 的结构返回。
服务端发送的数据可以在浏览器控制台中查看,这可能会暴露敏感数据,导致数据安全问题。
为了解决以上问题,并使其支持 POST 请求以及自定义的返回数据格式,我们可以使用以下技巧
利用 Fetch API 的流处理能力,我们可以实现对 SSE 的扩展:
- /**
- * Utf8ArrayToStr: 将Uint8Array的数据转为字符串
- * @param {Uint8Array} array - Uint8Array数据
- * @return {string} - 转换后的字符串
- */
- function Utf8ArrayToStr(array) {
- const decoder = new TextDecoder();
- return decoder.decode(array);
- }
-
-
- /**
- * fetchStream: 建立一个SSE连接,并支持多种HTTP请求方式
- * @param {string} url - 请求的URL地址
- * @param {object} params - 请求的参数,包括HTTP方法、头部、主体内容等
- * @return {Promise} - 返回一个Promise对象
- */
- const fetchStream = (url, params) => {
- const { onmessage, onclose, ...otherParams } = params;
-
-
- return fetch(url, otherParams)
- .then(response => {
- let reader = response.body?.getReader();
-
-
- return new ReadableStream({
- start(controller) {
- function push() {
- reader?.read().then(({ done, value }) => {
- if (done) {
- controller.close();
- onclose?.();
- return;
- }
- const decodedData = Utf8ArrayToStr(value);
- console.log(decodedData);
-
-
- onmessage?.(decodedData);
-
-
- controller.enqueue(value);
-
-
- push();
- });
- }
- push();
- }
- });
- })
- .then(stream => {
- return new Response(stream, {
- headers: { "Content-Type": "text/html" }
- }).text();
- });
- };
-
-
- // 示例:调用fetchStream函数
- fetchStream("/events", {
- method: "POST", // 使用POST方法
- headers: {
- "content-type": "application/json"
- },
- credentials: "include",
- body: JSON.stringify({
- // 这里列出了一些示例数据,实际业务场景请替换为你的数据
- boxId: "exampleBoxId",
- sessionId: "exampleSessionId",
- queryContent: "exampleQueryContent"
- }),
- onmessage: res => {
- console.log(res); // 当接收到消息时的回调
- },
- onclose: () => {
- console.log("Connection closed."); // 当连接关闭时的回调
- }
- });
我们定义一个名为eventStreamHandler.ts的文件
- // 定义请求主体的接口,需要根据具体的应用场景定义具体的属性
- interface RequestBody {
- // 示例属性,具体属性需要根据实际需求定义
- key?: string;
- }
-
-
- // 错误响应的结构
- interface ErrorResponse {
- error: string;
- detail: string;
- }
-
-
- // 返回值类型定义
- type TextStream = ReadableStreamDefaultReader<Uint8Array>;
-
-
- // 获取数据并返回TextStream
- async function fetchData(
- url: string,
- body: RequestBody,
- accessToken: string,
- onError: (message: string) => void
- ): Promise<TextStream | undefined> {
- try {
- // 尝试发起请求
- const response = await fetch(url, {
- method: "POST",
- cache: "no-cache",
- keepalive: true,
- headers: {
- "Content-Type": "application/json",
- Accept: "text/event-stream",
- Authorization: `Bearer ${accessToken}`,
- },
- body: JSON.stringify(body),
- });
-
-
- // 检查是否有冲突,例如重复请求
- if (response.status === 409) {
- const error: ErrorResponse = await response.json();
- onError(error.detail);
- return undefined;
- }
-
-
- return response.body?.getReader();
- } catch (error) {
- onError(`Failed to fetch: ${error.message}`);
- return undefined;
- }
- }
-
-
- // 读取流数据
- async function readStream(reader: TextStream): Promise<string | null> {
- const result = await reader.read();
- return result.done ? null : new TextDecoder().decode(result.value);
- }
-
-
- // 处理文本流数据
- async function processStream(
- reader: TextStream,
- onStart: () => void,
- onText: (text: string) => void,
- onError: (error: string) => void,
- shouldClose: () => boolean
- ): Promise<void> {
- try {
- // 开始处理数据
- onStart();
-
- while (true) {
- if (shouldClose()) {
- await reader.cancel();
- return;
- }
- const text = await readStream(reader);
- if (text === null) break;
-
-
- onText(text);
- }
- } catch (error) {
- onError(`Processing stream failed: ${error.message}`);
- }
- }
-
-
- /**
- * 主要的导出函数,用于处理流式文本数据。
- *
- * @param url 请求的URL。
- * @param body 请求主体内容。
- * @param accessToken 访问令牌。
- * @param onStart 开始处理数据时的回调。
- * @param onText 接收到数据时的回调。
- * @param onError 错误处理回调。
- * @param shouldClose 判断是否需要关闭流的函数。
- */
- export async function streamText(
- url: string,
- body: RequestBody,
- accessToken: string,
- onStart: () => void,
- onText: (text: string) => void,
- onError: (error: string) => void,
- shouldClose: () => boolean
- ): Promise<void> {
- const reader = await fetchData(url, body, accessToken, onError);
-
- if (!reader) {
- console.error("Reader is undefined!");
- return;
- }
-
-
- await processStream(reader, onStart, onText, onError, shouldClose);
- }
七、兼容性
发展至今,SSE 已具有广泛的的浏览器兼容性,几乎除 IE 之外的浏览器均已支持。
八、总结
SSE (Server-Sent Events) 是基于 HTTP 协议的轻量级实时通信技术。其核心特点是由服务器主动推送数据到客户端,而不需要客户端频繁请求。这样的特点使得 SSE 在某些应用场景中成为了理想选择,例如股票行情实时更新、网站活动日志推送、或聊天室中的实时在线人数统计。
然而,尽管 SSE 有很多优势,如断线重连机制、相对简单的实现和轻量性等,但它也存在明显的局限性。首先,SSE 只支持单向通信,即服务器到客户端的数据推送,而无法实现真正的双向交互。其次,由于浏览器对并发连接数有限制,当需要大量的实时通信连接时,SSE 可能会受到限制。
相对而言,WebSockets 提供了一个更加强大的双向通信机制,能够满足高并发、高吞吐量和低延迟的需求。因此,在选择适合的实时通信方案时,开发者需要根据应用的具体需求和场景来做出选择。简而言之,对于需要简单、低频率更新的场景,SSE 是一个非常不错的选择;而对于需要复杂、高频、双向交互的应用,WebSockets 可能更为合适。
最后,无论选择哪种技术,都应对其优缺点有深入了解,以确保在特定场景下可以提供最佳的用户体验。
-end-
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。