Web端实时通信技术SSE在携程机票业务中的实践应用

Posted on 2025-06-13 15:32 Jack Jiang 阅读(52) 评论(0) 编辑收藏

本文由携程前端开发专家Chris Xia分享，关注新技术革新和研发效率提升。

1、引言

本文介绍了携程机票前端基于Server-Sent Events（SSE）实现服务端推送的企业级全链路通用技术解决方案。文章深入探讨了 SSE 技术在应用过程中包括方案对比、技术选型、链路层优化以及实际效果等多维度的技术细节，为类似使用场景提供普适性参考和借鉴。该方案设计目标是实现通用性，适用于各种网络架构和业务场景。

技术交流：

- 移动端IM开发入门文章：《新手入门一篇就够：从零开发移动端IM》

- 开源IM框架源码：https://github.com/JackJiang2011/MobileIMSDK（备用地址点此）

（本文已同步发布于：http://www.52im.net/thread-4832-1-1.html）

2、技术背景

在如今互联网应用中，实时数据推送已成为很多业务场景的关键技术解决方案。携程机票业务作为在线旅游行业的核心场景，面临着航班数据实时性要求高、信息维度复杂等挑战。

Server-Sent Events（SSE）技术作为一种基于 HTTP 长连接的服务器推送方案，非常适用于机票业务"服务端主动推送、客户端实时展示"的需求特点。

相较于 WebSocket 等双向通信协议，SSE 在实现简单性、协议轻量级和浏览器兼容性等方面具有显著优势，适合机票列表页这类以服务端数据为主导的业务场景。

3、认识SSE

3.1 SSE 是什么？

Server-Sent Events（SSE）服务器发送事件，是一种基于 HTTP 长连接，允许服务器单向实时推送数据到客户端的技术。

SSE 的工作原理非常简单直观。客户端通过与服务器建立一条持久化的 HTTP 连接，然后服务器使用该连接将数据以事件流（event stream）的形式发送给客户端。这些事件流由多个事件（event）组成，每个事件包含一个标识符、类型和数据字段。客户端通过监听事件流来获取最新的数据，并在接收到事件后进行处理。

关于SSE技术的详细介绍可以阅读《SSE技术详解：一种全新的HTML5服务器推送事件技术》。

3.2 SSE 的使用场景

SSE 使用场景非常广泛，大家熟知的 Chatgpt 对话的交互形式使用的就是 SSE 技术。

SSE 在服务器单向实时推送数据的场景非常适用：

1）实时数据流：如股票市场更新、新闻推送、体育比分更新等；
2）实时通知：如社交媒体消息提醒、新订单通知等；
3）仪表盘更新：如系统监控、实时数据统计等。

关于SSE在如今热门的AI大模型技术的中的应用可以阅读：《全民AI时代，大模型客户端和服务端的实时通信到底用什么协议？》、《大模型时代多模型AI网关的架构设计与实现》。

4、先说效果

机票前端首次在核心业务中（机票航班列表）使用 SSE 技术，机票列表页由原先客户端串行请求获取多批次航班数据变为一次请求由服务持续推送数据给客户端。

在调研了公司内外各种实现方案，最终联合携程框架、SRE、机票前后端团队共同实现了全公司通用的SSE技术解决方案（详情见下文中的全链路支持部分）。

1）使用 SSE 前（如下图）：

1）客户端需要发起两次请求获取完整航班数据；
2）服务端采用预取优化：在响应第一次请求时，提前获取第二批数据并缓存至 Redis（降低客户端第二次请求响应的耗时）；
3）客户端发起第二次请求时，可直接获取缓存数据。

这样的流程和技术方案无疑会提升前后端的代码复杂度，服务端需要额外增加一层缓存来提升响应时间，客户端无法感知服务到底有多少批次数据，需要不断问询。

2）使用 SSE 后（如下图）：

客户端发送一次 SSE 请求，服务端实时推送数据到客户端，服务间上下游同样采用流式传输，实现客户端到服务端全链路流式通信。

3）SSE 为前后端带来的价值：

1）减少请求传输耗时：无需请求多次，减少了多次请求的传输耗时；
2）前后端代码结构优化：代码更简洁且易于理解，减少串行请求的回调监听/嵌套；
3）服务逻辑优化：列表数据移除了 redis 的发布订阅流程，简化了代码架构；
4）资源利用率提升：减少冗余请求（只有一批数据时，客户端不用再次请求问询服务）。

4）SSE 对性能有提升吗？

通过分析请求流程（建立链接 -> 发送请求 -> 响应数据传输）和其原理，发现 HTTP 1.1 和 2 支持链路复用，因此链接建立的次数本质上没有变化。在传输通道和数据压缩方式保持不变的情况下，响应数据传输的耗时也不会有明显变化。

SSE 的核心性能优势在于减少了请求发送的次数，其性能增益取决于具体的使用场景：

a）当服务端响应耗时大于网络传输耗时，性能提升有限。

使用 SSE 与传统串行请求的性能实验数据对比：

b）当网络传输耗时大于服务端处理耗时，减少请求次数可以显著降低整体延迟。

5、方案选型

目前市面上很多服务端推送的技术解决方案：SSE、轮询/串行、Websocket 等，我们从易用性，资源开销，使用场景等多维度对比了几个使用较多的主流方案，最终选择了 SSE。

5.1 服务端推送

简单几行代码实现服务端SSE推送。

SSE 的数据传输规范中有 4 个关键字段 event、data、id 和 retry，用于定义和传输事件数据。

具体是：

1）even：定义消息的事件类型，客户端可以根据事件类型触发不同的处理逻辑；
2）data：消息的主体内容；
3）id：为消息设置一个唯一的 ID，用于客户端断线重连时标识最后接收的消息；
4）retry：服务端指定客户端在连接断开后重新连接的时间间隔（单位为毫秒）。

这些字段共同构成了 SSE 消息的基本格式，每条消息以两个换行符 \n\n 结束，确保客户端能够正确解析和处理事件数据。

前端使用样例：

// 创建 EventSource
const evtSource = new EventSource("接口地址");

// 监听服务端推送的数据
evtSource.onmessage = function (event) {
  console.log("接收到的消息：", event);
};

// 监听连接建立
evtSource.onopen = function () {
  console.log("连接已建立");
};

// 监听报错
evtSource.onerror = function (err) {
  console.error("发生异常:", err);
};

服务使用样例（以 Nodejs 为例）：

const http = require("http");

http
.createServer((req, res) => {
    // 设置Response Header
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    });
    // 不断推送数据给客户端
    const pushData = setInterval(() => {
      res.write(data);
    }, 1000);

    req.on("close", () => clearInterval(pushData));
  })
  .listen(3000);

5.2 内部SSE实践方案

调研发现公司内部有两套实践方案：

1）自定义响应式网关：实现网关轮询服务批量获取数据，从而实现流式传输。绕开公司链路层，没有通用性；
2）前端轮询下沉BFF(服务)：前端与BFF建立SSE通道，BFF不断轮询向上游批量获取数据。轮询位置发生变化，并未实现全链路的流式通信。

在携程企业级网络生态架构下，从通用性和完整度分析对比了两套方案，并没有真正意义上从前到后打通整条链路。

仅仅只是简单接入SSE是远远不够的，离不开全链路（SSE技术选型，多层网络架构的适配，服务间的流式通信等等）的支持，所以最终决定联合携程框架、SRE、机票前后端团队共同来实现对SSE全链路的适配，真正意义上实现全公司通用的普适方案。

5.3 SSE技术选型

确定好整体技术方案后，我们在实际测试过程中发现了 2 个 Web 原生 SSE 的局限性问题。

具体是：

1）仅支持 Get 请求：对需要传递一些复杂请求体的场景不友好；
2）不支持自定义 http header：无法支持自定义 header 透传，鉴权等场景，目前市面大部分解决方案是使用 Cookie 来携带自定义参数。

针对上述问题，调研发现微软开源的 SSE 网络库 @microsoft/fetch-event-source（以下简称 fes）能够很好的解决。fes 是基于 Fetch 和 ReadableStream 来实现的 SSE 功能，旨在提供更加灵活便利的调用方式。

原生 SSE 和 fes 的对比：

fetch-event-source 详解：fes 的核心原理是通过 Fetch 发送请求，ReadableStream 读取响应流，在 JS 侧实现字节流数据的解析。通过对比原生 SSE（chromium 内核中 EventSource）和 fes 的代码，发现整体流程与实现方案大致相同，关键区别在于流的解析，原生 SSE 在浏览器内核由 C++实现，fes 在 JS 侧实现。

fes 的流解析：

1）核心方法：getBytes、getLines 和 getMessages；
2）getBytes：通过 ReadableStream 读取响应字节流，获取每个字节块；
3）getLines：将 getBytes 获取到的字节块解析为 EventSource 行缓冲区，处理这些字节块并解析为行，然后调用 onLine 回调函数处理每一行；
4）getMessages：创建 EventSourceMessage 对象，将行缓冲区数据解析并进行组装，处理完成后回调给调用方。

export async function getBytes(stream: ReadableStream<Uint8Array>, onChunk: (arr: Uint8Array) => void) {
    const reader = stream.getReader();
    let result: ReadableStreamDefaultReadResult<Uint8Array>;
    while (!(result = await reader.read()).done) {
        onChunk(result.value);
    }
}

export function getMessages(
    onId: (id: string) => void,
    onRetry: (retry: number) => void,
    onMessage?: (msg: EventSourceMessage) => void
) {
    let message = newMessage();
    const decoder = new TextDecoder();

    // return a function that can process each incoming line buffer:
    return function onLine(line: Uint8Array, fieldLength: number) {
        if (line.length === 0) {
            // empty line denotes end of message. Trigger the callback and start a new message:
            onMessage?.(message);
            message = newMessage();
        } else if (fieldLength > 0) { // exclude comments and lines with no values
            // line is of format "<field>:<value>" or "<field>: <value>"
            // [url=https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation]https://html.spec.whatwg.org/mul ... ream-interpretation[/url]
            const field = decoder.decode(line.subarray(0, fieldLength));
            const valueOffset = fieldLength + (line[fieldLength + 1] === ControlChars.Space ? 2 : 1);
            const value = decoder.decode(line.subarray(valueOffset));

            switch (field) {
                case 'data':
                    // if this message already has data, append the new value to the old.
                    // otherwise, just set to the new value:
                    message.data = message.data
                        ? message.data + '\n' + value
                        : value;
                    break;
                case 'event':
                    message.event = value;
                    break;
                case 'id':
                    onId(message.id = value);
                    break;
                case 'retry':
                    const retry = parseInt(value, 10);
                    if (!isNaN(retry)) {
                        onRetry(message.retry = retry);
                    }
                    break;
            }
        }
    }
}

export function getLines(onLine: (line: Uint8Array, fieldLength: number) => void) {
    let buffer: Uint8Array | undefined;
    let position: number; // current read position
    let fieldLength: number; // length of the `field` portion of the line
    let discardTrailingNewline = false;

    return function onChunk(arr: Uint8Array) {
        if (buffer === undefined) {
            buffer = arr;
            position = 0;
            fieldLength = -1;
        } else {
            buffer = concat(buffer, arr);
        }

        const bufLength = buffer.length;
        let lineStart = 0; // index where the current line starts
        while (position < bufLength) {
            if (discardTrailingNewline) {
                if (buffer[position] === ControlChars.NewLine) {
                    lineStart = ++position; // skip to next char
                }

                discardTrailingNewline = false;
            }

            let lineEnd = -1; // index of the \r or \n char
            for (; position < bufLength && lineEnd === -1; ++position) {
                switch (buffer[position]) {
                    case ControlChars.Colon:
                        if (fieldLength === -1) { // first colon in line
                            fieldLength = position - lineStart;
                        }
                        break;
                    case ControlChars.CarriageReturn:
                        discardTrailingNewline = true;
                    case ControlChars.NewLine:
                        lineEnd = position;
                        break;
                }
            }

            if (lineEnd === -1) {
                break;
            }

            onLine(buffer.subarray(lineStart, lineEnd), fieldLength);
            lineStart = position; // we're now on the next line
            fieldLength = -1;
        }

        if (lineStart === bufLength) {
            buffer = undefined; // we've finished reading it
        } else if (lineStart !== 0) {
            buffer = buffer.subarray(lineStart);
            position -= lineStart;
        }
    }
}