Python 基于 SSE 实现流式模式
引言
在现代 Web 应用中，流式模式通过 Server-Sent Events (SSE) 技术实现，能够持续向客户端发送数据，提升用户体验。本文将介绍如何使用 Python 实现基于 SSE 的流式模式。
什么是流式模式？
流式模式通过流的方式持续发送数据，而不是一次性返回所有数据。与传统 HTTP 请求不同，流式模式允许服务器在连接打开后持续向客户端发送数据，适用于实时数据更新场景。
SSE：流式模式的核心技术
SSE 是一种服务器向客户端推送实时更新的技术，基于 HTTP 协议，通过简单的 GET 请求开启持久连接。服务器使用 Content-Type: text/event-stream 标记返回的数据流，客户端通过 EventSource API 接收数据。
SSE 的基本事件块格式
event: 事件名称
data: 数据内容
每条事件以 event 和 data 两个字段组成，data 是传输内容，event 用于标记事件类型。
SSE 的优势
- 实现简单
- 数据流控制稳定
- 自动重连机制
- 兼容性好
实现基于 SSE 的流式模式
服务端实现（使用 FastAPI）
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time
app = FastAPI()
def generate_stream():
for i in range(1, 11):
yield f"data: Message {i}\n\n"
time.sleep(1)
@app.get("/stream")
async def stream():
return StreamingResponse(generate_stream(), media_type="text/event-stream")
generate_stream函数生成数据流，使用yield逐步发送消息。StreamingResponse类将数据流作为响应发送给客户端，media_type="text/event-stream"标识为 SSE 流。
客户端实现（使用 JavaScript）
<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="UTF-8" />
<title>SSE 流式模式示例</title>
</head>
<body>
<h1>实时消息流</h1>
<div id="messages"></div>
<script>
const messagesDiv = document.getElementById("messages");
const eventSource = new EventSource("/stream");
eventSource.onmessage = (event) => {
const newMessage = document.createElement("p");
newMessage.textContent = event.data;
messagesDiv.appendChild(newMessage);
};
eventSource.onerror = () => {
console.log("连接出错，正在重新连接...");
};
</script>
</body>
</html>
EventSource对象连接服务端的 SSE 流。onmessage事件处理器将接收到的消息实时添加到页面中。
应用场景
- 实时数据监控
- 聊天应用
- 新闻推送
- 流式问答系统
流式模式与 WebSocket 的对比
- SSE：单向通信，适合服务器向客户端推送数据。
- WebSocket：双向通信，适合需要双向互动的场景。
未来展望
随着 HTTP/3 协议的普及，流式模式将在更多场景中得到应用，SSE 的简单实现和稳定数据传输方式将使其在未来更加流行。
\