FastAPI 中的 SSE：构建高性能实时应用
Server-Sent Events (SSE) 是一种强大的技术，允许服务器单向地将数据推送到客户端。与传统的轮询或 WebSockets 相比，SSE 提供了一种更简单、更轻量级的解决方案，尤其适用于实时数据更新、通知和流式传输等场景。FastAPI，作为一个高性能的 Python Web 框架，完美地支持 SSE，使其成为构建实时应用的理想选择。本文将深入探讨如何在 FastAPI 中使用 SSE，涵盖从基础概念到高级应用的各个方面。
SSE 的工作原理
SSE 基于 HTTP 协议，利用持久的 HTTP 连接，服务器可以持续地向客户端发送数据。客户端通过 EventSource API 建立连接，并监听服务器发送的事件。服务器发送的数据采用简单的文本格式，每条消息以 data: 开头，并以两个换行符 \n\n 结尾。
相比于 WebSockets 的双向通信，SSE 的单向通信模式更加简洁高效，减少了协议的复杂性和开销。对于只需要服务器推送数据的场景，SSE 往往是更优的选择。
FastAPI 中的 SSE 实现
FastAPI 提供了简洁易用的 API 来实现 SSE。核心在于 StreamingResponse 对象，它允许我们以流的形式返回数据给客户端。结合异步生成器，我们可以轻松地实现实时数据推送。
“`python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
app = FastAPI()
@app.get(“/sse”)
async def sse_stream(request: Request):
async def event_generator():
while True:
# 检查客户端连接是否断开
if await request.is_disconnected():
break
# 模拟实时数据
data = {"message": "Hello from SSE!", "timestamp": asyncio.get_event_loop().time()}
# 使用 Server-Sent Events 格式发送数据
yield f"data: {data}\n\n"
# 控制发送频率
await asyncio.sleep(1)
return StreamingResponse(event_generator(), media_type="text/event-stream")
“`
代码解析:
event_generator是一个异步生成器函数，负责生成要发送给客户端的 SSE 事件。request.is_disconnected()用于检测客户端是否断开连接，避免资源浪费。yield f"data: {data}\n\n"将数据格式化为 SSE 事件并发送。asyncio.sleep(1)控制事件发送的频率。StreamingResponse将异步生成器包装成流式响应，并设置media_type为text/event-stream。
客户端 JavaScript 代码:
“`javascript
const eventSource = new EventSource(‘/sse’);
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log(data);
};
eventSource.onerror = (error) => {
console.error(‘SSE connection error:’, error);
};
“`
处理客户端断开连接
在上面的例子中，我们使用了 request.is_disconnected() 来检测客户端断开连接。这对于避免服务器资源浪费至关重要。当客户端断开连接时，request.is_disconnected() 会返回 True，从而终止事件生成器。
更复杂的 SSE 应用
除了简单的定时推送，FastAPI 的 SSE 还可以实现更复杂的应用场景：
- 实时数据更新: 将 SSE 与数据库或其他数据源集成，实现实时数据更新推送。例如，监控系统、股票行情等。
- 通知系统: 使用 SSE 实现实时通知功能，例如消息推送、系统告警等。
- 进度条: 将长时间运行的任务的进度通过 SSE 实时推送给客户端，提供友好的用户体验。
- 流式数据处理: 处理大量的流式数据，例如日志文件、传感器数据等，并通过 SSE 将处理结果实时推送给客户端。
高级技巧：自定义事件类型
SSE 支持自定义事件类型，可以通过 event: 字段指定。这允许客户端根据不同的事件类型进行不同的处理。
python
yield f"event: custom_event\ndata: {data}\n\n"
客户端 JavaScript 代码：
javascript
eventSource.addEventListener('custom_event', (event) => {
console.log('Custom event received:', event.data);
});
错误处理和重连
在实际应用中，网络连接可能会中断。客户端 EventSource API 提供了 onerror 事件，可以用于处理连接错误并实现自动重连。
javascript
eventSource.onerror = (error) => {
console.error('SSE connection error:', error);
// 尝试重新连接
setTimeout(() => {
new EventSource('/sse');
}, 5000);
};
与 WebSockets 的比较
SSE 和 WebSockets 都是实现实时通信的技术，但它们各有优缺点。SSE 更轻量级，更易于实现，适用于服务器推送数据的场景。WebSockets 支持双向通信，更灵活，但协议更复杂，开销更大。选择哪种技术取决于具体的应用需求。
总结
FastAPI 提供了简洁而强大的 SSE 支持，使其成为构建高性能实时应用的理想选择。通过异步生成器和 StreamingResponse，我们可以轻松地实现实时数据推送。结合客户端 EventSource API，可以构建丰富的实时应用，例如实时数据更新、通知系统、进度条等。 理解 SSE 的工作原理和 FastAPI 的实现方式，可以帮助开发者更好地利用这项技术，构建更优秀的实时应用。 选择 SSE 还是 WebSockets 取决于具体的应用场景，对于单向数据推送的场景，SSE 通常是更优的选择。 通过本文的介绍，相信读者已经对 FastAPI 中的 SSE 有了更深入的理解，并能够将其应用到实际项目中。