FastAPI实现SSE：实时数据流的最佳实践

2025-3-19

FastAPI 实现 SSE：实时数据流的最佳实践

引言

在现代 Web 应用开发中，实时数据流的需求越来越普遍。无论是聊天应用、股票行情、实时位置跟踪，还是系统监控、物联网(IoT)数据，都需要将服务器端的数据变化实时推送给客户端，而无需客户端频繁发起请求。传统的轮询（Polling）或长轮询（Long Polling）技术存在效率低、资源消耗大、延迟高等问题。Server-Sent Events (SSE) 作为一种更优的解决方案，正受到越来越多的关注。

SSE 是一种基于 HTTP 的单向通信协议，允许服务器主动向客户端推送数据。它利用一个持久的 HTTP 连接，服务器可以持续地向客户端发送文本数据流。与 WebSocket 相比，SSE 更轻量级、简单，更易于实现，并且在某些场景下更具优势。

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，基于 Python 3.6+ 的类型提示。它以其简洁的语法、自动化的文档生成、以及出色的性能而著称。FastAPI 天然支持异步编程，这使得它非常适合处理 SSE 这种需要长时间保持连接的场景。

本文将深入探讨如何使用 FastAPI 实现 SSE，并分享构建高效、可靠的实时数据流服务的最佳实践。

1. 理解 Server-Sent Events (SSE)

1.1 SSE 的工作原理

SSE 基于一个简单的 HTTP 长连接。客户端发起一个普通的 HTTP 请求到服务器，请求头中包含 Accept: text/event-stream，表明客户端希望接收事件流。服务器收到请求后，保持连接打开，并通过该连接持续向客户端发送文本数据。

服务器发送的数据遵循特定的格式：

  • 数据块 (Event Stream): 整个数据流由多个数据块组成。
  • 事件 (Event): 每个数据块代表一个事件。
  • 字段 (Field): 每个事件可以包含多个字段，常见的字段有：
    • data: 事件的数据内容。可以是任意文本，通常是 JSON 格式。
    • event: 事件的类型。客户端可以根据事件类型来处理不同的事件。
    • id: 事件的唯一标识符。用于断线重连时，客户端可以告知服务器从哪个事件开始继续接收。
    • retry: 重新连接的时间间隔（毫秒）。如果连接断开，客户端会等待指定时间后尝试重新连接。

每个字段占一行，以字段名和冒号开头，后面跟着字段值。字段之间用换行符 (\n) 分隔。一个空行 (\n\n) 表示一个事件的结束。

示例：

```
event: message
data: {"user": "John", "text": "Hello!"}
id: 123

event: update
data: {"count": 5}
id: 124

```

1.2 SSE 与 WebSocket 的对比

特性 Server-Sent Events (SSE) WebSocket
通信方向 单向（服务器 -> 客户端） 双向
协议 基于 HTTP 独立的协议 (ws:// 或 wss://)
实现复杂度 简单 较复杂
适用场景 服务器向客户端推送数据 客户端与服务器双向实时通信
连接数限制 受浏览器限制（通常每个域名 6 个） 理论上无限制
断线重连 自动重连 (EventSource) 需要手动实现
兼容性 较好（除 IE 外） 良好

1.3 SSE 的优势与局限性

优势：

  • 简单易用： 基于 HTTP，无需额外的协议或库。
  • 轻量级： 相比 WebSocket，开销更小。
  • 自动重连： 浏览器内置的 EventSource 对象支持自动重连。
  • 单向通信： 适用于服务器推送数据的场景，避免了不必要的双向通信开销。
  • 易于调试： 数据以文本形式传输，可以使用浏览器开发者工具直接查看。

局限性：

  • 单向通信： 不适合需要双向通信的场景。
  • 连接数限制： 浏览器通常限制同一域名下的 SSE 连接数。
  • 文本数据： 主要用于传输文本数据，传输二进制数据需要进行编码。
  • 兼容性： 不支持 IE 浏览器（可以使用 Polyfill）。

2. FastAPI 中实现 SSE

FastAPI 提供了对 SSE 的出色支持，可以轻松地构建 SSE 服务端。

2.1 安装 FastAPI 和 Uvicorn

bash
pip install fastapi uvicorn

2.2 基本 SSE 实现

```python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import time

app = FastAPI()

async def event_generator():
"""模拟数据流"""
i = 0
while True:
yield f"data: {i}\n\n"
i += 1
await asyncio.sleep(1)

@app.get("/stream")
async def stream_events(request: Request):
"""SSE 端点"""
return StreamingResponse(event_generator(), media_type="text/event-stream")

``
这段代码实现了一个简单的SSE服务器，该服务器无限生成从0开始递增的数字流。
客户端访问/stream` 路由将开始接收数据。

解释：

  • event_generator()：这是一个异步生成器函数，它模拟了一个持续的数据流。每次循环产生一个符合 SSE 格式的字符串 (data: ...\n\n)，并使用 await asyncio.sleep(1) 模拟数据产生的间隔。
  • stream_events()：这是 SSE 的端点函数。它使用 StreamingResponse 将 event_generator() 的输出作为事件流返回给客户端。media_type="text/event-stream" 指定响应的内容类型为 SSE。
  • async/await 确保了异步执行。

2.3 使用 EventSource 客户端测试

在浏览器中，可以使用 JavaScript 的 EventSource 对象来连接 SSE 端点：

```html




SSE Example

SSE Data:



```

这段代码使用 EventSource 对象连接到 /stream 端点。onmessage 事件处理程序在每次接收到新数据时更新页面内容。onerror 事件处理程序处理连接错误。

2.4 添加事件类型和 ID

```python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import time
import json

app = FastAPI()

async def event_generator():
"""模拟不同类型的事件流"""
i = 0
while True:
if i % 2 == 0:
event_data = {
"type": "message",
"data": {"user": "System", "text": f"Message {i}"}
}
else:
event_data = {
"type": "update",
"data": {"count": i}
}

    yield f"event: {event_data['type']}\ndata: {json.dumps(event_data['data'])}\nid: {i}\n\n"
    i += 1
    await asyncio.sleep(1)

@app.get("/stream")
async def stream_events(request: Request):
"""SSE 端点"""
return StreamingResponse(event_generator(), media_type="text/event-stream")

```

客户端 JavaScript 代码：

```javascript
const eventSource = new EventSource("/stream");

eventSource.addEventListener("message", (event) => {
const data = JSON.parse(event.data);
console.log("Received message:", data);
// ... 处理 message 类型的事件 ...
});

eventSource.addEventListener("update", (event) => {
const data = JSON.parse(event.data);
console.log("Received update:", data);
// ... 处理 update 类型的事件 ...
});

eventSource.onopen = () => {
console.log("Connection to server opened.");
}

eventSource.onerror = (error) => {
console.error("EventSource failed:", error);
eventSource.close();
};

```

在这个例子中：

  • 服务器根据 i 的奇偶性发送不同类型的事件（message 或 update）。
  • 使用 json.dumps() 将 Python 字典转换为 JSON 字符串。
  • 客户端使用 addEventListener() 监听特定类型的事件，并分别处理。
  • onopen 事件处理程序在连接建立时触发。

3. 最佳实践

3.1 错误处理和重连

  • 服务器端错误处理： 在生成器函数中捕获异常，并发送错误信息给客户端。可以定义一个特殊的事件类型（如 error）来表示错误。

    python
    try:
    # ... 产生数据的代码 ...
    except Exception as e:
    yield f"event: error\ndata: {str(e)}\n\n"

  • 客户端错误处理： 使用 EventSource.onerror 监听连接错误，并在必要时关闭连接或尝试重新连接。

  • 设置 retry 字段： 在服务器端设置 retry 字段，告知客户端在断开连接后多久尝试重新连接。

    python
    yield f"retry: 5000\ndata: ...\n\n" # 5 秒后重连
    * 利用id字段： 如果连接中断，客户端可以从最后一个接收到的事件ID重新开始。
    ```python
    async def event_generator(last_event_id: int = 0):
    """支持从指定 ID 开始的事件流"""
    i = last_event_id
    while True:
    # ...
    yield f"event: {event_data['type']}\ndata: {json.dumps(event_data['data'])}\nid: {i}\n\n"
    # ...

    @app.get("/stream")
    async def stream_events(request: Request, lastEventId: int = Header(None, alias="Last-Event-ID")):

    return StreamingResponse(event_generator(lastEventId), media_type="text/event-stream")

    ``
    客户端在请求头里添加    Last-Event-ID`，服务器将返回从该ID开始的数据流。

3.2 数据格式化

  • JSON： 使用 JSON 格式传输数据是最常见的做法，因为它易于解析和处理。
  • 自定义格式： 如果需要传输更复杂的数据结构，可以考虑使用 Protocol Buffers 或 Avro 等序列化格式。

3.3 性能优化

  • 异步生成器： 使用异步生成器函数可以避免阻塞主线程，提高服务器的并发处理能力。
  • 批量发送： 如果数据产生的速度很快，可以考虑将多个事件合并成一个数据块发送，减少网络开销。
  • 数据压缩： 如果数据量较大，可以考虑使用 Gzip 等压缩算法来减少传输的数据量。
  • 连接管理： 及时关闭不再需要的连接，释放资源。

3.4 安全性

  • 身份验证和授权： 像保护其他 API 端点一样保护 SSE 端点，确保只有授权用户才能访问实时数据。

  • 跨域资源共享 (CORS)： 如果 SSE 服务端和客户端不在同一个域下，需要配置 CORS。
    ```python
    from fastapi.middleware.cors import CORSMiddleware

    app = FastAPI()

    origins = [
    "http://localhost",
    "http://localhost:8080",
    "*"
    ]

    app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=[""],
    allow_headers=["    "],
    )
    ```

  • 防止 DDoS 攻击： 限制每个客户端的连接数，并对请求进行速率限制。

3.5 可扩展性

  • 负载均衡： 如果需要处理大量的并发连接，可以使用负载均衡器将请求分发到多个服务器实例。
  • 消息队列： 使用消息队列（如 Redis、RabbitMQ、Kafka）来解耦数据生产者和 SSE 服务端。数据生产者将数据发布到消息队列，SSE 服务端从消息队列订阅数据并推送给客户端。

4. 真实案例：实时股票行情

以下是一个使用 FastAPI 实现实时股票行情推送的简化示例：

```python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import time
import json
import random

app = FastAPI()

async def get_stock_price(symbol: str):
"""模拟获取股票价格"""
while True:
price = round(random.uniform(100, 200), 2) # 模拟价格波动
yield {
"symbol": symbol,
"price": price,
"timestamp": int(time.time())
}
await asyncio.sleep(2)

async def stock_event_generator(symbol: str):
"""股票行情事件生成器"""
async for stock_data in get_stock_price(symbol):
yield f"event: stock_update\ndata: {json.dumps(stock_data)}\n\n"

@app.get("/stock/{symbol}")
async def stream_stock_price(symbol: str, request: Request):
"""SSE 股票行情端点"""
return StreamingResponse(stock_event_generator(symbol), media_type="text/event-stream")
```

客户端 JavaScript 代码（简化）：

```javascript
const eventSource = new EventSource("/stock/AAPL"); // 假设要获取苹果公司的股票行情

eventSource.addEventListener("stock_update", (event) => {
const data = JSON.parse(event.data);
console.log("Stock update:", data);
// ... 更新股票行情显示 ...
});
```

5. 总结

FastAPI 提供了一种简单、高效的方式来实现 SSE 实时数据流。通过结合异步编程、生成器函数、以及 FastAPI 的 StreamingResponse，可以轻松地构建各种实时应用。

在实际开发中，还需要考虑错误处理、重连机制、数据格式化、性能优化、安全性、以及可扩展性等方面。遵循最佳实践，可以构建出稳定、可靠、高性能的实时数据流服务。

希望这篇文章能够帮助你更好地理解和使用 FastAPI 实现 SSE。如果你有任何问题或建议，欢迎提出。

版权声明：
作者：admin
链接：https://hostlocvps.com/2025/03/19/fastapi%e5%ae%9e%e7%8e%b0sse%ef%bc%9a%e5%ae%9e%e6%97%b6%e6%95%b0%e6%8d%ae%e6%b5%81%e7%9a%84%e6%9c%80%e4%bd%b3%e5%ae%9e%e8%b7%b5/
文章版权归作者所有，未经允许请勿转载。
THE END
道琼斯30指数成分股完整列表及分析
<<上一篇
Kubernetes API：核心概念、版本控制与扩展
下一篇>>
相关推荐
如何使用 Git Stash 避免代码丢失？
CentOS安装MySQL：一步步教你搞定
Azure Point-to-Site VPN 与 Site-to-Site VPN：选择哪个？
学习正则表达式：匹配数字、字母、特殊字符
越南电子签证代办：省时省力，专业高效
Debian Docker 镜像详解：优势、用途与最佳实践
Nginx端口转发：配置详解与应用指南
道琼斯指数(Dow Jones)是什么？一文详解
探索 GitHub 上的顶级 Lua 项目
最新B次元资讯、活动与更新