python

《Python编程实战》10.2 SSE 与流式响应

用一条只读 HTTP 流做服务端推送：真跑 text/event-stream 的 data/event/id 字段、StreamingResponse 流式输出与 Last-Event-ID 断线续传，并给出 SSE 与 WebSocket 的选型对照。

Leeting Yan 6 分钟阅读 3004 字

本节目标：掌握 SSE 的 text/event-stream 报文格式与 FastAPI StreamingResponse 用法，真跑通流式输出与 Last-Event-ID 断线续传，并在 SSE 与 WebSocket 之间做出有理有据的选型。
适用版本：Python 3.12+（实测 3.14.6）；fastapi 0.143.0、starlette 1.7.0、httpx 0.28.1

10.2 SSE 与流式响应

10.1 的 WebSocket 是「双向通道」。但很多场景其实只需要服务端单向推：AI 逐字吐答案、日志实时滚动、任务进度条、股票报价。这类场景开一条全双工 WebSocket 是杀鸡用牛刀——SSE（Server-Sent Events）用一条普通的、只读的 HTTP 响应流就够，而且天生带自动重连和断点续传。

10.2.1 SSE 是什么

SSE 本质就是：服务端返回一个不结束的 HTTP 响应，Content-Type 是 text/event-stream，然后一条一条往里写文本，浏览器用 EventSource 自动解析。

它和「普通流式下载」的区别在于有格式约定：每个事件是若干 字段: 值 行，用一个空行结束。浏览器收到空行才认为「这个事件完整了」，触发一次 message/onmessage 回调。

关键在于它完全建立在 HTTP 之上：没有协议升级、没有新帧格式、浏览器原生支持 EventSource（自带重连）、可以穿过绝大多数只认 HTTP 的代理和 CDN。代价是只能服务端→客户端单向，客户端要发消息得另开一个 HTTP 请求。

10.2.2 text/event-stream 的格式

SSE 的字段只有四个，全是行首关键字：

字段含义备注
data:事件数据多条 data: 会用 \n 拼接；只有它才算「有内容」
event:事件类型名客户端用 addEventListener("log", ...) 按名监听
id:事件 ID浏览器记住它，重连时放进 Last-Event-ID 请求头
retry:重连等待毫秒数服务端告诉客户端「断了我该等多久再试」

用代码拼一个事件块（实测输出）：

import json

def sse_pack(event: str, data: dict, eid: int | None = None, retry: int | None = None) -> str:
    lines = []
    if eid is not None:
        lines.append(f"id: {eid}")
    lines.append(f"event: {event}")
    if retry is not None:
        lines.append(f"retry: {retry}")
    lines.append(f"data: {json.dumps(data, ensure_ascii=False)}")
    return "\n".join(lines) + "\n\n"     # 末尾空行 = 事件结束

原始字节（repr）:
'id: 1\nevent: log\nretry: 3000\ndata: {"seq": 1, "text": "log-1"}\n\n'

两个极易踩的坑：分隔符必须是 \n\n（空行），不是单个 \n——少了空行浏览器会一直攒着不触发；以及 data 里的换行必须拆成多条 data:，直接塞 \n 会截断事件。JSON 序列化后没有裸换行，所以安全。

10.2.3 StreamingResponse：流式输出

FastAPI 用 StreamingResponse 接收一个异步生成器，边产出边发送，而不是等整个响应体拼完。这正是「逐字输出」的实现方式：

import asyncio
from collections.abc import AsyncIterator
from fastapi import FastAPI, Header, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
EVENTS = [f"log-{i}" for i in range(1, 6)]

@app.get("/events")
async def events(request: Request, last_event_id: str | None = Header(default=None)):
    start = int(last_event_id) if last_event_id else 0

    async def gen() -> AsyncIterator[str]:
        for i in range(start, len(EVENTS)):
            if await request.is_disconnected():   # 客户端走了就停
                break
            yield sse_pack("log", {"seq": i + 1, "text": EVENTS[i]}, eid=i + 1, retry=3000)
            await asyncio.sleep(0.05)
        yield sse_pack("done", {"total": len(EVENTS)}, eid=len(EVENTS))

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

用 httpx 的 ASGITransport 直接把请求打给 ASGI 应用（不需要起真实服务器），逐行读事件流，真跑（实测）：

import httpx

transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
    async with client.stream("GET", "/events") as resp:
        print("Content-Type:", resp.headers["content-type"])
        async for line in resp.aiter_lines():
            print(repr(line))

=== 首次请求（无 Last-Event-ID）===
Content-Type: text/event-stream; charset=utf-8
'id: 1'
'event: log'
'retry: 3000'
'data: {"seq": 1, "text": "log-1"}'
''
'id: 2'
'event: log'
'retry: 3000'
'data: {"seq": 2, "text": "log-2"}'
''
... (略) ...
'id: 5'
'event: done'
'data: {"total": 5}'
''

注意每块之间的那个 ''（空行）——它就是事件分隔符在 aiter_lines() 里的可见形态。三个工程要点：

  • request.is_disconnected() 必须查。用户关掉页面后，生成器若不退出会一直空转，白占一个连接。
  • 两个响应头：Cache-Control: no-cache 防中间层缓存流；X-Accel-Buffering: no 关掉 Nginx 的响应缓冲（否则事件会被攒着一起发，流式效果消失）。
  • 生成器里别做阻塞调用。await 之间若夹了同步 time.sleep 或重查询，整个事件循环会被卡住。

10.2.4 断线续传：Last-Event-ID

SSE 最值钱的能力在这里。浏览器 EventSource 断线后自动重连，并且会把最后收到的 id 放进 Last-Event-ID 请求头带回来。服务端只要按这个 ID 续发，用户就感觉不到断过线：

last_event_id: str | None = Header(default=None)
start = int(last_event_id) if last_event_id else 0

真跑（客户端带 Last-Event-ID: 3，实测输出）：

=== 断线续传（Last-Event-ID: 3）===
'id: 4'
'event: log'
'retry: 3000'
'data: {"seq": 4, "text": "log-4"}'
''
'id: 5'
'event: log'
'retry: 3000'
'data: {"seq": 5, "text": "log-5"}'
''
'id: 5'
'event: done'
'data: {"total": 5}'
''

从 id: 4 开始——前三个事件被跳过，因为客户端已经收过了。这套机制要真正好用，服务端必须把 id 设成「可定位的序号」（如数据库自增 ID、日志 offset），而不是随便一个 UUID：拿到 ID 就能从那个位置继续读。若 ID 不可定位，续传就无从谈起。

用 Python 客户端（httpx）自己实现同样的续传逻辑——边读边记住最后的 id，断线后用 Last-Event-ID 头重连：

async def consume(client: httpx.AsyncClient, last_id: str | None) -> str | None:
    headers = {"Last-Event-ID": last_id} if last_id else {}
    async with client.stream("GET", "/events", headers=headers) as resp:
        eid = last_id
        async for line in resp.aiter_lines():
            if line.startswith("id: "):
                eid = line[4:]                      # 记住最后收到的事件 ID
            elif line.startswith("data: "):
                print(f"    [last_id={last_id}] data -> {line[6:]}")
        return eid

真跑（消费到 id=2 主动断开，再带 Last-Event-ID=2 重连，实测）：

=== 第一段：消费到 id=2 后主动断开 ===
    [client] 收到 id=2，模拟断线
=== 第二段：带 Last-Event-ID=2 重连 ===
    [last_id=2] data -> {"seq": 3, "text": "log-3"}
    [last_id=2] data -> {"seq": 4, "text": "log-4"}
    [last_id=2] data -> {"seq": 5, "text": "log-5"}
    [last_id=2] data -> {"total": 5}

重连后从 seq: 3 无缝接上，一条不重、一条不漏。这就是为什么 SSE 特别适合「日志滚动」「任务进度」这类可续传的追加流：客户端只要记住一个整数，就能在任何断点原地复活。

注意：Last-Event-ID 是 EventSource 自动带上的。如果你用自己的客户端（如 httpx）消费，得手动把上次的 id 记下来、下次请求时放进头里——上面这段代码做的就是这件事。

10.2.5 SSE vs WebSocket：怎么选

两者都能推消息，选型看通信方向和生态约束：

维度SSEWebSocket
方向服务端 → 客户端（单向）双向
协议普通 HTTP，无升级HTTP 升级到 ws://
断线重连EventSource 自动 + Last-Event-ID 续传要自己实现（见 10.3）
二进制不支持，只能文本支持（opcode 0x2）
浏览器 APIEventSource（原生）WebSocket（原生）
代理/CDN 兼容好（就是 HTTP）一般（部分代理掐长连接）
并发连接限制HTTP/1.1 下每域名 6 条无此限制
典型场景通知、进度、AI 流式输出、行情只读聊天、协同编辑、游戏

一句话决策：只需要服务端推 → SSE；需要客户端也频繁主动发 → WebSocket。AI 对话的「逐字吐答案」是 SSE 的经典用法——客户端发一次提问（普通 POST），服务端用 SSE 流式回答案，全程单向。

10.2.6 工程注意点

  • HTTP/1.1 的 6 连接上限：浏览器对同一域名最多 6 条 HTTP/1.1 连接，SSE 会长期占用其中一条。多标签页场景要留意，生产建议上 HTTP/2（多路复用，不再有每域名 6 条的限制）。
  • 心跳注释保活：中间代理常把长时间无数据的连接掐掉。SSE 没有协议级 ping，惯例是定期发一个注释行 : keep-alive\n\n（以冒号开头的行被浏览器忽略），既保活又不触发事件。
  • 连接是有成本的：每个 SSE 连接 = 一个长期占用的协程 + socket。万级并发要配合反向代理的超时设置与文件描述符上限一起调。
  • 错误也要发事件：流中途出错，别直接断——发一个 event: error 的事件再正常结束，客户端才能区分「服务端说完了」和「连接被掐了」。

延伸阅读

小结

  • SSE 是建立在普通 HTTP 上的一条只读流，Content-Type: text/event-stream，浏览器 EventSource 原生支持。
  • 报文只有四个字段：data（内容）、event（类型）、id（续传锚点）、retry（重连间隔），事件以空行分隔。
  • FastAPI 用 StreamingResponse + 异步生成器实现流式输出，别忘了查 request.is_disconnected() 和设 X-Accel-Buffering: no。
  • Last-Event-ID 是 SSE 的杀手锏：浏览器自动重连并带回最后的事件 ID，服务端按 ID 续发即可无缝续传——前提是 ID 可定位。
  • 选型口诀：只需服务端推 → SSE；需要双向 → WebSocket。AI 流式输出、进度、只读行情用 SSE 更省事。
  • 生产要上 HTTP/2（绕开 6 连接上限），并用注释行 : keep-alive 保活。

SSE 用 HTTP 的「自动重连 + 续传」省了事，但 WebSocket 没有这套免费午餐——断线重连、心跳保活、广播扇出全得自己写。这正是下一节的主题。

阅读导航：上一节：WebSocket 与消息协议 · 下一节：心跳、重连与广播 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时