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

2025/2/23 7:29:03 来源：https://blog.csdn.net/h1773655323/article/details/142528314 浏览: 次关键词：【FastAPI】使用FastAPI和Redis实现实时通知(SSE)

在当今快速发展的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); // 在控制台显示通知
};