来源：京东云开发者

一、背景

随着现代技术的快速发展，即时交互变得越来越重要。用户不仅希望获取信息，而且希望以更直观和实时的方式体验它。这在聊天应用程序和其他实时通信工具中尤为明显，用户习惯看到对方正在输入的提示。

ChatGPT，作为 OpenAI 的代表性产品之一，不仅为用户提供了强大的自然语言处理能力，而且关注用户的整体交互体验。在使用 ChatGPT 进行交互时，用户可能已经注意到了一个细节：当它产生回复时，回复会像人类逐字输入的方式逐渐出现，而不是一次性显示完整答案。

这种打字效果给人一种仿佛与真人对话的感觉，进一步增强了其自然语言处理的真实感。一开始，许多开发者可能会误以为这是通过 WebSockets 实现的，这是因为 WebSockets 是一种常用于实时通信的技术。然而，仔细研究后，我们发现 ChatGPT 使用了一种不同的技术：基于 EventStream 的方法。更具体地说，它似乎是通过 SSE (Server-Sent Events) 来实现逐个字地推送答案的。

此外，考虑到 ChatGPT 的复杂性和其涉及的大量计算，响应时间可能会长于其他基于数据库的简单查询。因此，采用 SSE 逐步推送结果的方式可以帮助减少用户感到的等待时间，从而增强用户体验。

二、SSE 简介

Server-Sent Events（通常简称为 SSE）是一种允许服务器向Web页面发送实时更新的技术。与WebSocket技术相比，SSE专门设计用于从服务器到客户端的单向通信。这种单向性使其在某些场景中更为简单和直观。

2.1 主要特点

单向通信：SSE 专为从服务器到客户端的单向通信设计。客户端不能通过SSE直接发送数据到服务器，但可以通过其他方法如AJAX与服务器进行交互。
基于HTTP：SSE 基于 HTTP 协议运行，不需要新的协议或端口。这使得它能够轻松地在现有的Web应用架构中使用，并且通过标准的HTTP代理和中间件进行支持。
自动重连：如果连接断开，浏览器会自动尝试重新连接到服务器。
格式简单：SSE 使用简单的文本格式发送消息，每个消息都以两个连续的换行符分隔。
原生浏览器支持：许多现代浏览器（如 Chrome、Firefox 和 Safari）已原生支持SSE，但需要注意的是，某些浏览器，如Internet Explorer和早期的Edge版本，不支持SSE。

2.2 SSE 与 WebSockets

虽然 SSE 与 WebSockets 在某种程度上有些相似，但它们之间还存在一些关键差异，如下所示：

对比项	Server-Sent Events (SSE)	WebSockets
基于协议	基于 HTTP，简化了连接和交互的过程	通常基于 WS/WSS（基于TCP），更为灵活
通信能力	单向通信：仅服务器向客户端发送消息	双向通信能力
配置	配置简单，易于理解和使用	需要更复杂的配置和理解
断线与消息追踪	自带的断线重连和消息跟踪功能	通常需要手动处理或使用额外库
数据格式	通常为文本，但可以发送经过编码/压缩的二进制消息	支持文本和原始二进制消息
事件处理	支持多种自定义事件	基本消息机制，不能像SSE那样自定义事件类型
连接并发性	连接数可能受到 HTTP 版本的限制，尤其是在HTTP/1.1中	WebSocket被设计为支持更高的连接并发性
安全性	仅支持HTTP和HTTPS的安全机制	支持WS和WSS，可以在WSS上实现更强大的加密
浏览器兼容性	大部分现代浏览器支持，但不是所有浏览器	几乎所有现代浏览器都支持
开销	由于基于HTTP，每次消息可能有较大的头部开销	握手后，消息头部开销相对较小

三、服务端深入解析

3.1 SSE 的协议机制

Server-Sent Events（SSE）是一个基于 HTTP 的协议，允许服务器单向地向浏览器推送信息。为了成功地使用 SSE，服务器和客户端都必须遵循一定的规范和流程。

当客户端（例如浏览器）发出请求订阅 SSE 服务时，服务器需要通过设置特定的响应头部信息来确认该请求。这些头部信息包括：

Content-Type: text/event-stream: 这表示返回的内容为事件流。
Cache-Control: no-cache: 这确保服务器推送的消息不会被缓存，以保障消息的实时性。
Connection: keep-alive: 这指示连接应始终保持开放，以便服务器可以随时发送消息。

3.2 消息的格式和结构

SSE 使用简单的文本格式来组织和发送消息。基本的消息结构是由一系列行组成，每一行由字段名、一个冒号和字段值组成。

以下是消息中可以使用的一些字段及其用途：

event: 定义了事件的类型。这可以帮助客户端确定如何处理接收到的消息。
id: 提供事件的唯一标识符。如果连接中断，客户端可以使用最后收到的事件 ID 来请求服务器从某个点重新发送消息。
retry: 指定了当连接断开时，客户端应等待多少毫秒再尝试重新连接。这为连接中断和重连提供了一种机制。
data: 这是消息的主体内容。它可以是任何 UTF-8 编码的文本，而且可以跨多行。每行数据都会在客户端解析时连接起来，中间使用换行符分隔。

为了确保消息的正确和完整传输，服务器通常在消息的末尾添加一个空行，表示消息的结束。

示例:

id: 123event: updatedata: {"message": "This is a test message"}

此外，SSE 也支持多条连续消息的发送。只要每条消息之间使用两个换行符隔开即可。

四、客户端实践

接入 SSE 并不困难，尤其在客户端这边。主流浏览器提供了 EventSource API，使得与 SSE 服务端建立和维护连接变得异常简单。

4.1 如何建立连接

首先，需要创建一个 EventSource 对象，它将代表与服务器的持久连接。初始化时，可以为它提供一些选项，以满足特定需求。







const options = {  withCredentials: true  // 允许跨域请求携带凭证};
// 创建一个 EventSource 对象以开始监听const eventSource = new EventSource('your_server_url', options);

在上面的代码中，withCredentials 参数用于指示是否应该在请求中发送凭证（例如 cookies）。这在跨域场景中可能会非常有用。

4.2 如何处理收到的事件

一旦与服务器建立了连接，就可以开始监听从服务器发送过来的事件。

通用事件处理:

默认情况下，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);});

4.3 关闭连接

如果不再需要从服务器接收事件，可以使用 close 方法关闭连接。

eventSource.close();

关闭连接后，将不再接收任何事件，除非再次初始化 EventSource 对象。

总结：使用 EventSource API，客户端可以方便地与 SSE 服务器交互，从而实时接收数据更新。这为创建响应迅速的 web 应用提供了极大的便利，同时避免了传统的轮询方式带来的资源浪费。

五、理论实践

5.1 服务端


































































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');

5.2 客户端


























































<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 Demotitle>head><body>  <h1>SSE Demoh1>  <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}
`;    }script>body>html>

将上面的两份代码保存为 server.js 和 index.html，并在命令行中执行 node server.js 启动服务端，然后在浏览器中打开 http://localhost:3000 即可看到 SSE 效果。

六、业务实践

6.1 存在问题

在业务真实使用场景中，基于SSE的方法存在一些问题和限制：

默认请求仅支持 GET 方法。当前端需要向后端传递参数时，参数只能拼接在请求的 URL 上，对于复杂的业务场景来说实现较为麻烦。
对于服务端返回的数据格式有固定要求，必须按照 event、id、retry、data 的结构返回。
服务端发送的数据可以在浏览器控制台中查看，这可能会暴露敏感数据，导致数据安全问题。

为了解决以上问题，并使其支持 POST 请求以及自定义的返回数据格式，我们可以使用以下技巧

6.2 优化技巧

利用 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."); // 当连接关闭时的回调    }});

6.3 封装插件

我们定义一个名为eventStreamHandler.ts的文件

















































































































// 定义请求主体的接口，需要根据具体的应用场景定义具体的属性interface RequestBody {    // 示例属性，具体属性需要根据实际需求定义    key?: string;}
// 错误响应的结构interface ErrorResponse {    error: string;    detail: string;}
// 返回值类型定义type TextStream = ReadableStreamDefaultReader<Uint8Array>;
// 获取数据并返回TextStreamasync 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-

ITPub博客

大模型问答助手前端实现打字机效果