1. 背景

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

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

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

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

2. SSE 简介

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

2.1 主要特点

1. 单向通信：SSE 专为从服务器到客户端的单向通信设计。客户端不能通过 SSE 直接发送数据到服务器，但可以通过其他方法如 AJAX 与服务器进行交互。

2. 基于 HTTP：SSE 基于 HTTP 协议运行，不需要新的协议或端口。这使得它能够轻松地在现有的 Web 应用架构中使用，并且通过标准的 HTTP 代理和中间件进行支持。

3. 自动重连：如果连接断开，浏览器会自动尝试重新连接到服务器。

4. 格式简单：SSE 使用简单的文本格式发送消息，每个消息都以两个连续的换行符分隔。

5. 原生浏览器支持：许多现代浏览器（如 Chrome、Firefox 和 Safari）已原生支持 SSE，但需要注意的是，某些浏览器，如 Internet Explorer 和早期的 Edge 版本，不支持 SSE。

2.2 SSE 与 WebSockets

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

3. 服务端深入解析

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: 123
event: update
data: {"message": "This is a test message"}

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

4. 客户端实践

接入 SSE 并不困难，尤其在客户端这边。主流浏览器提供了EventSourceAPI，使得与 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对象。

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

5. 理论实践

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 客户端

<!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 效果。

6. 业务实践

6.1 存在问题

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

1. 默认请求仅支持GET方法。当前端需要向后端传递参数时，参数只能拼接在请求的 URL 上，对于复杂的业务场景来说实现较为麻烦。

2. 对于服务端返回的数据格式有固定要求，必须按照event、id、retry、data的结构返回。

3. 服务端发送的数据可以在浏览器控制台中查看，这可能会暴露敏感数据，导致数据安全问题。

为了解决以上问题，并使其支持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>;

// 获取数据并返回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);
}

7. 兼容性

发展至今，SSE 已具有广泛的的浏览器兼容性，几乎除 IE 之外的浏览器均已支持。

8. 总结

SSE (Server-Sent Events) 是基于 HTTP 协议的轻量级实时通信技术。其核心特点是由服务器主动推送数据到客户端，而不需要客户端频繁请求。这样的特点使得 SSE 在某些应用场景中成为了理想选择，例如股票行情实时更新、网站活动日志推送、或聊天室中的实时在线人数统计。

然而，尽管 SSE 有很多优势，如断线重连机制、相对简单的实现和轻量性等，但它也存在明显的局限性。首先，SSE 只支持单向通信，即服务器到客户端的数据推送，而无法实现真正的双向交互。其次，由于浏览器对并发连接数有限制，当需要大量的实时通信连接时，SSE 可能会受到限制。

相对而言，WebSockets 提供了一个更加强大的双向通信机制，能够满足高并发、高吞吐量和低延迟的需求。因此，在选择适合的实时通信方案时，开发者需要根据应用的具体需求和场景来做出选择。简而言之，对于需要简单、低频率更新的场景，SSE 是一个非常不错的选择；而对于需要复杂、高频、双向交互的应用，WebSockets 可能更为合适。

最后，无论选择哪种技术，都应对其优缺点有深入了解，以确保在特定场景下可以提供最佳的用户体验。

作者：京东科技 卞荣成

来源：京东云开发者社区 转载请注明来源