当前位置：首页 > news >正文

Server-Sent Events (SSE) 技术详解

news 2025/7/1 7:52:23

Server-Sent Events (SSE) 技术详解

一、基础概念

1. 核心特点

单向通信：仅服务器→客户端方向
基于HTTP：使用标准HTTP/1.1协议
文本协议：UTF-8编码的纯文本格式
自动重连：内置断线重连机制
轻量级：API设计简洁

2. 协议对比

特性	SSE	WebSocket	Polling
协议	HTTP	ws/wss	HTTP
延迟	低	极低	高
数据方向	单向	双向	半双工
浏览器支持	除IE	全部	全部

二、技术实现细节

1. HTTP 响应头

服务器必须设置以下头部：

Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

2. 事件流格式

基本结构：

event: messageType\n
data: messageContent\n\n

示例：

event: status
data: {"time": "2023-05-20", "value": 42}data: This is a simple message: comment line (ignored)
id: 12345
retry: 5000

3. 字段说明

data: 消息内容（可多行，最终会合并）
event: 事件类型（默认message）
id: 事件ID（用于断线重连时Last-Event-ID头）
retry: 重连延迟（毫秒）
:: 注释行（会被客户端忽略）

三、实践

以下是用Python实现SSE(Server-Sent Events)的完整示例，包含服务端和客户端实现：

1、服务端

from flask import Flask, Response, request
import time
import jsonapp = Flask(__name__)@app.route('/stream')
def stream():def event_stream():count = 0while True:time.sleep(1)  # 模拟延迟count += 1# 构建SSE格式数据data = {"time": time.strftime("%Y-%m-%d %H:%M:%S"),"count": count}# SSE格式要求event = f"data: {json.dumps(data)}\n\n"yield eventif count >= 10:  # 限制发送10条后结束yield "event: end\ndata: Stream completed\n\n"breakreturn Response(event_stream(),mimetype="text/event-stream",headers={'Cache-Control': 'no-cache','Connection': 'keep-alive','X-Accel-Buffering': 'no'  # 禁用Nginx缓冲})if __name__ == '__main__':app.run(threaded=True, port=5000)

2、客户端

import requests
import jsonurl = 'http://localhost:5000/stream'def sse_client():with requests.get(url, stream=True) as response:content_type = response.headers.get('content-type', '')if 'text/event-stream' in content_type:buffer = ""for chunk in response.iter_content(chunk_size=1024):if chunk:buffer += chunk.decode('utf-8')while '\n\n' in buffer:event, buffer = buffer.split('\n\n', 1)process_event(event)else:print(f"Server does not support SSE, content-type: {content_type}")def process_event(event):lines = event.split('\n')event_data = {}event_type = 'message'  # 默认事件类型for line in lines:if line.startswith('event:'):event_type = line.split(':', 1)[1].strip()elif line.startswith('data:'):data = line.split(':', 1)[1].strip()try:event_data = json.loads(data)except json.JSONDecodeError:event_data = dataif event_type == 'end':print("Stream ended")else:print(f"Received {event_type}: {event_data}")if __name__ == '__main__':sse_client()