Server-Sent Events流式推送规范：让服务器像电台一样对你单向播报

可视化演示

展示Server-Sent Events从服务端到客户端的流式数据推送过程与SSE/WebSocket协议对比，调节数据块大小查看流式效果变化。

SSE 初识：从电台广播到单向推送

想象你每天早上开车上班，都会打开固定的电台频率。播音员不间断地播报路况、新闻和音乐，而你只需要安静地收听，不用做任何事情。这种“电台只管播、收音机只管听”的模式，正是 SSE 的工作方式。在 Web 世界里，服务器就是那个电台，浏览器就是你的收音机，一旦调好频率（建立连接），服务器就能持续不断地把数据推送过来。

在 SSE 出现之前，网页要实现“实时”效果往往采用轮询（Polling）方式：浏览器每隔几秒就向服务器问一次“有新数据吗？”就像你每隔五分钟就给电台打一次电话问路况，不仅浪费电话费，还总赶不上最新消息。SSE 改变了这个模式，让服务器能主动开口说话。

SSE 全称 Server-Sent Events，直译就是“服务器发送的事件”。它基于 HTTP 协议，是一种服务端单向推送技术。最关键的特性有三点：单向性——数据只能从服务器流向客户端；持久性——连接建立后保持打开状态，服务器可以随时发送消息；标准性——作为 W3C 标准，浏览器原生支持。大模型 API 的流式输出正是 SSE 的典型应用：当你向 ChatGPT 提问时，回答不是一次性到达，而是像电台播报一样逐字涌出，这就是 SSE 在底层持续推送“文字片段”事件。

EventSource 基础：旋动收音机旋钮

使用 SSE 的第一步是“打开收音机”——创建 EventSource 对象并连接到服务器的推送端点。在 JavaScript 中只需一行代码：const source = new EventSource('/api/stream')。这会向服务器发起一个 HTTP 请求，服务器接受后返回 Content-Type: text/event-stream，连接就建立了，这个连接会一直保持打开状态。

连接建立后，你主要和三个“开关”打交道：onopen 在连接成功建立时触发，相当于收音机“咔哒”一声锁定了频率；onmessage 在收到默认消息时触发，相当于电台开始播报内容；onerror 在连接出错时触发，相当于信号中断时收音机发出的“沙沙”声。三者共同构成了 SSE 的事件处理基础。

默认情况下，服务器发送的每条消息都可以通过 onmessage 回调接收。消息的格式很简单——以 data: 开头，跟一段文本，最后以空行结束。例如服务器发送 data: 你好\n\n，浏览器就会在 onmessage 中收到内容为“你好”的 MessageEvent。

const source = new EventSource('/api/stream');source.onopen = function() {  console.log('连接已建立');};source.onmessage = function(event) {  console.log('收到消息：', event.data);};source.onerror = function() {  console.log('连接出错，准备重连...');};

自定义事件设计：区分不同电台节目

真实电台不会只播一种内容——新闻、音乐、交通播报各有各的频道或时段。SSE 的自定义事件机制也是如此，它允许服务器发送不同类型的事件，客户端可以分别监听，就像你把收音机调到不同频道收听不同节目一样。

自定义事件的格式很简单，在消息中加入 event: 字段来指定事件名称，后面跟 data: 字段携带数据。例如服务器发送 event: update\ndata: 进度50%\n\n，客户端就可以通过 source.addEventListener('update', handler) 来专门监听这种类型的事件。

这种设计带来的最大好处是代码可读性和可维护性。比如在一个 AI 聊天应用中，服务器可能发送多种事件：start 表示开始生成、token 携带生成的文本片段、error 报告错误、done 表示完成。如果都用 onmessage 接收，你就要自己解析 event 字段来区分，代码会变得混乱。使用自定义事件后，每种消息类型都有明确的处理函数，就像收音机上的每个预设频道按钮一样一目了然。

source.addEventListener('token', function(e) {  // 处理AI生成的文本片段  appendToChat(e.data);});source.addEventListener('error', function(e) {  // 处理错误事件  showErrorTip(e.data);});source.addEventListener('done', function(e) {  // AI回答完毕  stopLoading();});

断线重连机制：信号丢失后的自动调谐

开车进入隧道时，电台信号会暂时消失，但当你驶出隧道，收音机会自动重新捕捉到信号继续播放。SSE 的断线重连机制正是如此——当连接意外中断时，浏览器会自动尝试重新连接服务器，不需要你手动干预。

SSE 的重连机制由浏览器自动触发。当连接因网络问题、服务器重启等原因断开时，onerror 回调被调用，随后浏览器会等待一小段时间（默认约 3 秒），然后自动发起新的连接请求。这个等待时间可以由服务器通过 retry: 字段来指定，单位是毫秒。例如 retry: 5000 告诉浏览器“5 秒后再重试”。

这种机制在真实场景中尤为重要。设想你在使用大模型 API 生成一段长文本，生成到一半时网络闪断，如果没有自动重连，你就什么都拿不到。有了 SSE 的自动重连，浏览器会默默重新建立连接，配合后文要讲的 Last-Event-ID 机制，还能实现“从哪里断开就从哪里继续”的无缝体验。

服务器发送的重连指令示例：retry: 5000: 上面的retry字段告诉客户端5秒后重试 服务器在连接中断时会收到事件通知（具体由实现决定）

Last-Event-ID 续传：标记书签，从断点继续

设想你在听一个很长的评书节目，听到一半信号断了，重新连上后，你希望从刚才断掉的地方继续听，而不是从头开始。Last-Event-ID 就是实现这一体验的关键机制——它允许服务器为每条事件分配一个 ID，客户端会记住最后收到的事件 ID，重连时把这个 ID 告诉服务器，服务器据此从断点之后继续推送。

实现原理分三步：首先，服务器在推送的每条事件中通过 id: 字段指定一个唯一 ID；其次，客户端每次收到带 ID 的事件，都会更新内部记录的 lastEventId；最后，当连接重建时，浏览器会自动在请求头中添加 Last-Event-ID 字段，值为之前记录的最后一个 ID，服务器读取这个值就知道该从哪继续推送。

在 AI 流式应用中，这种机制能有效避免因网络波动导致的响应内容丢失。比如大模型已经生成了 10 个 token，第 11 个 token 推送时连接断了，重连后携带 Last-Event-ID: 10，服务器从第 11 个 token 继续推送，用户就能获得完整的回答，不会出现中间缺字的情况。

服务器发送带ID的事件：id: 3data: 第3段内容 重连时浏览器自动携带：GET /api/streamLast-Event-ID: 3 服务器从id=4开始继续推送

SSE 实战要点：从理论到应用的最后一公里

理论讲完了，但真正把 SSE 用在实际项目中，还需要跨越不少“坑”。其中一个经典问题是代理缓冲：很多服务器或代理（如 Nginx）默认会对响应做缓冲，导致数据不是到达即发送，而是积攒到一定量才一次性发出。这把“流式”变成了“批处理”，打字机效果就消失了。解决方法是显式设置 X-Accel-Buffering: no 响应头，或配置代理禁用缓冲。

选型方面，SSE 最常和 WebSocket 一起讨论。两者都是实现实时通信的技术，但定位不同：SSE 是单向的，基于 HTTP，轻量且浏览器原生支持，适合“服务器主动推送”的场景；WebSocket 是双向的，基于 TCP 长连接，需要额外协议握手，适合“双方频繁互发消息”的场景。选型不必纠结，记住一个原则——如果你只需要服务器推送，SSE 往往更简单可靠；如果需要双向交互，WebSocket 才是合适的选择。

在大模型领域，SSE 已成为事实上的标准流式输出方案。主流大模型 API（如 OpenAI、Anthropic、智谱等）都提供了 SSE 格式的流式响应，开发者只需对接 EventSource 或类似机制，就能轻松实现“打字机效果”，显著提升用户体验。

一个SSE服务器端响应头示例：Content-Type: text/event-streamCache-Control: no-cacheConnection: keep-aliveX-Accel-Buffering: no

AI 应用场景

Python 演示

"""模拟SSE服务器端推送的实现示例"""import jsonimport timefrom flask import Flask, Response, request app = Flask(__name__)  def generate_events():    """模拟AI模型逐字生成文本的过程"""    tokens = ["你好", "，", "我", "是", "AI", "助手", "！"]    for index, token in enumerate(tokens):        # 发送token事件，携带当前生成的数据        event_data = json.dumps({            "index": index,            "token": token,            "message": "".join(tokens[: index + 1])        })        # 自定义事件类型：token表示生成的文本片段        yield "event: token\ndata: {}\n\n".format(event_data)        time.sleep(0.5)     # 发送完成事件，告诉客户端生成结束    yield "event: done\ndata: {}\n\n".format(        json.dumps({"message": "生成完成"})    )  @app.route("/api/chat")def chat():    """SSE端点：返回事件流响应"""    # 从请求头获取断点续传的ID，简化代码中仅打印示意    last_event_id = request.headers.get("Last-Event-ID")    if last_event_id is not None:        print("收到断点续传ID: {}", last_event_id)     return Response(        generate_events(),        mimetype="text/event-stream",  # 关键MIME类型，告知浏览器这是事件流        headers={            "Cache-Control": "no-cache",            "X-Accel-Buffering": "no",  # 禁用代理缓冲，确保流式效果        },    )  if __name__ == "__main__":    app.run(host="0.0.0.0", port=5000, debug=False)

1. mimetype="text/event-stream"通知浏览器这是 SSE 事件流，必须设置
2. yield "event: token\ndata: {}\n\n"构造 SSE 消息格式，事件类型为 token
3. request.headers.get("Last-Event-ID")读取客户端重连时携带的断点 ID
4. "X-Accel-Buffering": "no"禁止代理缓冲，保证消息即时推送

常见误区

面试题