带重连与解析的 SSE 客户端

功能说明

服务器发送事件使用长连接持续推送文本帧，每帧由若干字段行组成，空行表示帧结束。data 字段可以出现多次，最终用换行拼接；event 字段给出事件类型，id 字段用于断线续传。本例把字节流解析与网络读取分开：parse_sse 只处理已经解码的行，便于单元测试；iter_sse 负责发起请求、按行解码并在连接中断时重试。

重试次数有上限，退避时间随尝试次数线性增长，避免服务端故障时客户端疯狂重连。读取超时设为三十秒，防止连接永久挂起。

关键代码

parse_sse 维护事件字典和 data 列表，空行时若有数据就合并并产出事件，然后重置状态。以冒号开头的行是注释心跳，直接跳过。字段名与值在第一个冒号处分割，值去掉一个前导空格，符合事件流规范。流结束时如果仍有未以空行结尾的数据，也会补发一帧。iter_sse 用 urllib 请求并声明接受事件流，逐行解码后交给解析器，异常时增加计数并在超过上限后重新抛出。

使用注意

该实现没有发送 Last-Event-ID，断线重连可能重复收到事件，消费端应按事件 id 或业务键幂等处理。重连只覆盖读取阶段的网络异常，HTTP 错误状态码需要单独判断，否则认证失败也会被当成可重试故障。长连接必须设置读超时，并定期发送注释心跳以穿过代理的空闲断开。缓冲区没有上限，异常服务端发送超长行时可能耗尽内存。生产环境还应处理 TLS 校验、代理和压缩。

验证方式

运行脚本，解析结果包含一个事件字典，event 为 build，id 为 7，data 为两行拼接。随后打印 build 与解析后的 ok 对象，说明多行 data 被正确合并。可把 FIXTURE 改成没有结尾空行的两帧，确认最后一帧仍会输出。

代码示例

import json
import time
import urllib.error
import urllib.request
from collections.abc import Iterable, Iterator


def parse_sse(lines: Iterable[str]) -> Iterator[dict[str, str]]:
    event: dict[str, str] = {}
    data: list[str] = []
    for raw in lines:
        line = raw.rstrip("\r\n")
        if not line:
            if data:
                event["data"] = "\n".join(data)
                yield event
            event, data = {}, []
            continue
        if line.startswith(":"):
            continue
        field, _, value = line.partition(":")
        value = value.lstrip(" ")
        if field == "data":
            data.append(value)
        elif field:
            event[field] = value
    if data:
        event["data"] = "\n".join(data)
        yield event


def iter_sse(
    url: str, retries: int = 3, backoff: float = 0.5
) -> Iterator[dict[str, str]]:
    attempt = 0
    while True:
        try:
            request = urllib.request.Request(
                url, headers={"Accept": "text/event-stream"}
            )
            with urllib.request.urlopen(request, timeout=30) as response:
                yield from parse_sse(line.decode("utf-8") for line in response)
            return
        except (OSError, urllib.error.URLError):
            attempt += 1
            if attempt > retries:
                raise
            time.sleep(backoff * attempt)


FIXTURE = """event: build
id: 7
data: {"status": "ok"}
data: second line
"""


if __name__ == "__main__":
    events = list(parse_sse(FIXTURE.splitlines()))
    print(events)
    for event in events:
        first_line = event["data"].splitlines()[0]
        print(event.get("event", "message"), json.loads(first_line))