SSE(Server Sent Events) 数据如何采集？-北京尧图网络

一、SSE 基础认知SSEServer-Sent Events服务器发送事件是 HTML5 标准定义的一种单向实时通信协议基于标准 HTTP/HTTPS 协议实现允许服务器在客户端发起请求后通过持久化连接持续向客户端推送文本数据。与 WebSocket 的双向全双工通信不同SSE 是服务器→客户端的单向推送具备原生自动重连、断点续传、协议兼容性强等特点广泛应用于实时行情、日志流、AI 流式输出、消息通知等场景。SSE 数据帧格式规范SSE 传输的是纯文本数据流每条消息遵循固定格式以双换行符\n\n作为事件结束标识plaintextid: 1001 # 事件ID用于断点续传可选 event: price-update # 自定义事件类型可选默认为 message retry: 3000 # 断线重连间隔单位毫秒可选 data: {symbol:AAPL,price:189.5} # 消息正文可多行 : keep-alive # 冒号开头为注释常用作心跳保活核心字段说明data必填字段承载实际业务数据支持多行拼接event自定义事件名称客户端可按事件类型分别监听id事件唯一标识断线重连时客户端会通过Last-Event-ID请求头携带实现断点续传retry告知客户端断线后的重连等待时间二、浏览器端原生采集方案浏览器环境下采集 SSE 数据最标准的方式是使用原生EventSourceAPI这是 W3C 规范内置的接口无需引入第三方库。基础采集实现javascript运行// 1. 建立SSE连接 const eventSource new EventSource(https://api.example.com/stream); // 2. 监听连接建立 eventSource.onopen (event) { console.log(SSE 连接已建立, event); }; // 3. 采集默认 message 类型事件核心 eventSource.onmessage (event) { // event.data 为服务端推送的原始数据字符串 const data JSON.parse(event.data); console.log(收到数据:, data); // 此处可执行数据存储、渲染、业务处理等逻辑 }; // 4. 监听自定义事件类型 eventSource.addEventListener(price-update, (event) { const priceData JSON.parse(event.data); console.log(行情更新:, priceData); }); // 5. 错误与重连处理 eventSource.onerror (error) { console.error(SSE 连接异常:, error); // 原生EventSource默认自动重连如需手动控制可关闭后重建 if (eventSource.readyState EventSource.CLOSED) { console.log(连接已关闭); } }; // 6. 主动关闭连接 // eventSource.close();跨域场景处理当 SSE 接口与页面不同源时需在初始化时开启跨域配置javascript运行const eventSource new EventSource(https://cross-domain.com/stream, { withCredentials: true // 携带Cookie等凭证按需开启 });浏览器端采集的局限性仅支持 GET 请求无法自定义请求体、复杂请求头原生 API 无法手动控制重连逻辑、超时时间仅支持 UTF-8 文本无法直接处理二进制数据浏览器会自动管理连接生命周期页面关闭后连接自动断开三、服务端 / 脚本化采集方案在后端服务、数据爬虫、离线分析等非浏览器场景下需要通过 HTTP 客户端模拟 SSE 连接解析流式数据帧完成采集。方案一Python 生态采集Python 是数据采集最常用的语言有成熟的 SSE 客户端库也可基于原生 HTTP 流手动解析。1. sseclient 库最通用sseclient是轻量级 SSE 客户端兼容 Python 2/3自动解析 SSE 数据帧格式。安装bash运行pip install sseclient requests采集代码python运行import sseclient import requests import json def collect_sse_data(url, headersNone): # 必须开启 streamTrue启用流式响应 response requests.get( url, streamTrue, headersheaders or {Accept: text/event-stream}, timeout30 ) response.raise_for_status() # 初始化SSE客户端 client sseclient.SSEClient(response) # 迭代采集所有事件 for event in client.events(): # event.data / event.event / event.id / event.retry try: data json.loads(event.data) print(f事件ID:{event.id}, 类型:{event.event}, 数据:{data}) # 业务处理写入数据库、文件、消息队列等 except json.JSONDecodeError: print(f非JSON数据: {event.data}) if __name__ __main__: collect_sse_data(http://localhost:3000/events)2. httpx-sse 库异步场景基于 httpx 异步 HTTP 客户端的 SSE 实现适合高并发异步采集场景。安装bash运行pip install httpx-sse异步采集示例python运行import httpx from httpx_sse import aconnect_sse async def async_collect_sse(url): async with httpx.AsyncClient() as client: async with aconnect_sse(client, GET, url) as event_source: async for sse in event_source.aiter_sse(): print(f收到数据: {sse.data})3. 原生 HTTP 流手动解析无需第三方依赖直接读取 HTTP 响应流按 SSE 协议规则手动拆分数据帧python运行from http.client import HTTPConnection def raw_sse_collect(host, path, port80): conn HTTPConnection(host, port, timeout60) conn.request(GET, path, headers{Accept: text/event-stream}) response conn.getresponse() buffer while not response.closed: chunk response.readline().decode(utf-8) if not chunk: continue buffer chunk # 双换行符标识事件结束 if buffer.endswith(\n\n): # 解析单条事件 event parse_sse_event(buffer.strip()) print(解析到事件:, event) buffer def parse_sse_event(raw_text): 手动解析SSE事件帧 event {data: , event: message, id: None, retry: None} for line in raw_text.split(\n): if line.startswith(data:): event[data] line[5:].strip() \n elif line.startswith(event:): event[event] line[6:].strip() elif line.startswith(id:): event[id] line[3:].strip() elif line.startswith(retry:): event[retry] int(line[6:].strip()) event[data] event[data].rstrip(\n) return event方案二Node.js 采集Node.js 生态可通过原生http模块或eventsource包实现 SSE 数据采集。eventsource 包浏览器 API 兼容bash运行npm install eventsourcejavascript运行const EventSource require(eventsource); const es new EventSource(http://localhost:3000/events); es.onmessage (event) { console.log(采集到数据:, event.data); }; es.addEventListener(custom-event, (event) { console.log(自定义事件:, event.data); }); es.onerror (err) { console.error(连接错误:, err); };四、爬虫场景的 SSE 数据采集针对第三方网站的 SSE 流式数据抓取需要处理反爬校验、鉴权、断点续传等问题核心采集流程如下1. 接口定位与分析通过浏览器开发者工具定位 SSE 接口打开「网络」面板筛选EventStream类型查看请求 URL、请求头、查询参数确认鉴权方式Cookie、Token、签名参数等观察数据帧格式与事件类型2. 反爬适配要点请求头模拟完整携带User-Agent、Referer、Accept等浏览器特征头Accept需包含text/event-stream鉴权复刻从浏览器复制完整 Cookie、Authorization 等鉴权字段连接保活开启 HTTP Keep-Alive避免频繁断开触发风控重连策略模拟浏览器重连逻辑断线后携带Last-Event-ID续传避免数据重复或丢失3. 完整爬虫采集示例Pythonpython运行import sseclient import requests import time import json class SSEScraper: def __init__(self, url, headersNone): self.url url self.headers headers or {} self.last_event_id None self.running True def _build_headers(self): base_headers { Accept: text/event-stream, Cache-Control: no-cache, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Connection: keep-alive } base_headers.update(self.headers) if self.last_event_id: base_headers[Last-Event-ID] self.last_event_id return base_headers def start(self): retry_count 0 max_retry 5 while self.running and retry_count max_retry: try: response requests.get( self.url, streamTrue, headersself._build_headers(), timeout60 ) response.raise_for_status() retry_count 0 client sseclient.SSEClient(response) for event in client.events(): if not self.running: break # 更新最新事件ID if event.id: self.last_event_id event.id # 数据处理逻辑 self._process_data(event) except Exception as e: retry_count 1 print(f连接异常第{retry_count}次重连: {str(e)}) time.sleep(min(2 ** retry_count, 10)) # 指数退避 def _process_data(self, event): 业务数据处理存储、清洗、分析等 try: data json.loads(event.data) # 示例写入文件 with open(sse_data.log, a, encodingutf-8) as f: f.write(json.dumps({ id: event.id, event: event.event, data: data, timestamp: int(time.time()) }, ensure_asciiFalse) \n) except: print(f原始数据: {event.data}) if __name__ __main__: scraper SSEScraper(https://target-site.com/stream) scraper.start()五、采集关键问题与最佳实践1. 断线重连与数据一致性优先使用id字段实现断点续传重连时自动携带Last-Event-ID请求头服务端需支持按 ID 补发历史数据避免数据丢失客户端做幂等校验防止重复数据写入2. 心跳与保活处理SSE 连接长时间无数据时容易被中间代理、防火墙切断。采集端需处理识别服务端的注释心跳冒号开头的行忽略不做业务处理配置合理的超时时间超过阈值无数据时主动重连长时采集场景建议开启 TCP Keep-Alive3. 内存与性能优化流式处理边接收边解析不要等待全部数据加载到内存数据分片大批量数据按事件分批写入存储避免内存堆积连接复用复用 HTTP 连接池减少握手开销4. 异常处理边界网络波动实现指数退避重连机制避免无限重试压垮服务格式异常兼容不规范的 SSE 数据帧做好容错解析资源释放程序退出时主动关闭连接避免服务端连接泄漏六、适用场景与选型建议表格采集场景推荐方案前端页面实时渲染原生 EventSource APIPython 数据同步 / 离线分析sseclient requests高并发异步采集服务httpx-sse 异步方案第三方网站数据爬取手动 HTTP 流解析反爬适配Node.js 后端服务eventsource 包SSE 作为轻量级实时推送协议其采集核心本质是维护持久 HTTP 连接按协议规则解析流式文本帧。不同场景下的实现复杂度差异较大但只要掌握协议规范和数据帧格式就能在任意技术栈中完成稳定的数据采集。

SSE(Server Sent Events) 数据如何采集？

相关推荐

红酒品鉴入门：从葡萄品种到餐酒搭配

从10W到300W高并发系统架构实战：微服务与性能优化指南

工业自动化中伺服驱动器的选型、安装与维护指南

更多资讯

VAE如何决定Stable Diffusion图像质量：编解码原理与实操指南

SpringBoot Redis配置三大生死线与生产级调优指南

Linux文件读取函数封装：从read系统调用到底层I/O工程实践

DeepSeek政务大模型落地指南：架构、安全与实施避坑

FastMoss推荐码是什么？FastMoss优惠折扣码及FastMoss六大核心功能全面介绍

主动源面波数据处理实战：频散曲线求解程序的原理、操作与避坑指南

今日更新

MATLAB+Yalmip+CPLEX实战：综合能源系统优化调度全流程解析

I3C比I2C快10倍？RK3576实战：速率、DTS配置与混合总线避坑指南

字符串转对象：JSON.parse、new Function与URLSearchParams

本周精选

如何彻底离线使用 Laya-CoreML：Hugging Face 下载、缓存机制与 local_files_only 详解

基于1Panel 部署MySQL MCP 服务：TaoToken 统一 Key 接入与 config.toml 配置实战

崇州园区营销网站建设避坑指南：5大注意事项保备案安全

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？