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

【FastAPI】使用FastAPI和Redis实现实时通知(SSE)

news 2026/1/4 2:45:49

在当今快速发展的Web应用程序中，实时通知已成为用户体验的重要组成部分。无论是社交媒体更新、消息通知，还是系统状态提醒，实时数据推送可以极大地提升用户互动性。本文将详细介绍如何使用FastAPI和Redis实现Server-Sent Events (SSE) 来推送实时通知。

什么是Server-Sent Events (SSE)？

Server-Sent Events（SSE）是一种通过HTTP连接从服务器向客户端发送实时更新的技术。与WebSocket相比，SSE的实现更加简单，适用于单向数据流场景。服务器可以持续向客户端推送数据，而无需客户端不断发起请求。

为什么选择FastAPI和Redis？

FastAPI：作为一个现代的Web框架，FastAPI以其高性能和易用性而闻名。它支持异步编程，使得构建高并发应用变得更加简单。
Redis：作为一个高性能的键值存储数据库，Redis适合用作缓存和消息代理。在我们的场景中，Redis可以存储心跳信息，以决定何时停止推送通知。

环境准备

在开始之前，确保你的开发环境中安装了fastapi、uvicorn和redis库。可以通过以下命令进行安装：

pip install fastapi uvicorn redis

实现步骤

1. 创建FastAPI应用

首先，我们需要设置一个基本的FastAPI应用。下面是实现SSE的基本代码：

from fastapi import FastAPI
from starlette.responses import EventSourceResponse
import asyncio
import redisapp = FastAPI()
redis_client = redis.Redis()async def event_stream(key: str):while True:if redis_client.ttl(key) == -1:yield f"data: 心跳已过期\n\n"breakawait asyncio.sleep(1)  # 模拟等待yield f"data: 这里是实时通知\n\n"@app.get("/notifications/{key}")
async def notifications(key: str):return EventSourceResponse(event_stream(key))@app.post("/heartbeat/{key}/{seconds}")
async def set_heartbeat(key: str, seconds: int):redis_client.setex(key, seconds, "active")return {"message": "Heartbeat set"}

2. 代码解析

event_stream 函数：这是一个异步生成器。它将不断检查Redis中指定键的TTL（生存时间）。如果TTL过期，生成器将停止发送消息。
/notifications/{key}：这个端点用于建立SSE连接，允许客户端接收实时通知。
/heartbeat/{key}/{seconds}：这个端点用于设置Redis中的心跳时间，确保在指定时间内保持连接。

3. 前端实现

在前端，我们可以使用JavaScript来发送心跳请求并接收来自服务器的通知。以下是一个简单的示例：

const key = 'your_key';
const heartbeatDuration = 10; // 设置心跳时长为10秒// 发送心跳请求
fetch(`/heartbeat/${key}/${heartbeatDuration}`, { method: 'POST' });// 接收SSE通知
const eventSource = new EventSource(`/notifications/${key}`);
eventSource.onmessage = function(event) {console.log(event.data); // 在控制台显示通知
};

4. 测试与调试

在开发过程中，你可以使用uvicorn来运行你的FastAPI应用：

uvicorn your_file_name:app --reload

打开浏览器，确保前端代码能成功连接到后端，并能正确接收通知。如果遇到问题，检查浏览器的开发者工具，查看网络请求和控制台输出，以便进行调试。

应用场景

这种SSE实现可以广泛应用于以下场景：

社交网络：实时推送新消息或评论。
在线游戏：通知玩家状态变化。
监控系统：实时显示系统性能或状态更新。

总结

通过以上步骤，我们实现了一个使用FastAPI和Redis的实时通知系统。该系统能够根据心跳状态，持续推送通知，直到心跳到期为止。你可以根据具体需求进一步扩展这个示例，例如添加用户身份验证、处理不同类型的通知等。

希望这篇文章对你有所帮助！如果你有任何问题或建议，欢迎在评论区留言。