python_sse：使用FastAPI实现服务器发送的事件

共11个文件

py：4个

html：3个

gitignore：1个

5星 · 超过95%的资源需积分: 45 74 浏览量 2021-02-16 08:39:27 上传评论收藏 11KB ZIP 举报

在本文中，我们将深入探讨如何使用Python的FastAPI框架实现服务器发送事件（Server-Sent Events，简称SSE）。SSE是一种允许服务器向客户端推送实时更新的技术，常用于实时数据流应用，如股票报价、在线聊天或实时仪表板。FastAPI是一个现代、高性能的Web API框架，基于Python 3.6+，设计简洁且易于使用，与TypeScript紧密集成，提供自动化的API文档。让我们了解SSE的基本原理。SSE通过HTTP连接实现单向数据流，由服务器将更新推送到客户端，而无需客户端发起新的请求。这种模式在长轮询、WebSocket等其他实时通信技术中是不同的，因为它只需要一个开放的HTTP连接即可。SSE协议使用“text/event-stream”MIME类型，并且每个事件都包含事件类型和数据。接下来，我们看看如何在FastAPI中设置SSE。确保你已经安装了FastAPI和Uvicorn，这是FastAPI的推荐ASGI服务器。你可以使用以下命令安装： ```bash pip install fastapi uvicorn ``` 然后创建一个新的Python文件，例如`main.py`，并导入必要的库： ```python from fastapi import FastAPI, Response app = FastAPI() ``` 为了实现SSE，我们需要创建一个返回类型为`Response`的路由，这样可以自定义响应体的生成方式。下面是一个简单的例子，展示了如何发送实时时间戳： ```python @app.get("/sse", response_class=Response) async def sse(): while True: current_time = str(int(time.time() * 1000)) yield f"data: {current_time}\n\n" # 每行以"\n\n"结束表示一个完整的事件 await asyncio.sleep(1) # 每隔1秒发送新事件 ``` 在这个例子中，我们创建了一个名为`sse`的GET路由，它会无限循环地发送当前时间戳。`data:`后面是事件数据，每个事件之间必须以两个换行符分隔。为了让客户端能够接收这些事件，你需要在HTML页面中添加一个事件源（EventSource）元素。由于标签是HTML，我们将在前端部分讨论。创建一个`index.html`文件，放在FastAPI项目的静态文件夹`static`下： ```html <!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>FastAPI SSE示例</title> <script> var source = new EventSource("/sse"); source.onmessage = function(event) { var timestamp = event.data; console.log("接收到的时间戳:", timestamp); // 在这里处理接收到的数据，如更新UI }; </script> </head> <body> <h1>实时时间戳</h1> </body> </html> ``` 这个HTML页面创建了一个EventSource实例，连接到我们的SSE端点，并监听`onmessage`事件来处理接收到的时间戳。现在，你可以使用Uvicorn运行FastAPI服务器： ```bash uvicorn main:app --reload ``` 打开浏览器访问`http://localhost:8000/static/index.html`，你应该能看到实时更新的时间戳。总结一下，我们已经介绍了如何使用FastAPI实现SSE，包括设置SSE路由、发送事件以及在HTML页面中接收和处理这些事件。这种技术对于构建实时更新的应用非常有用，如实时统计、通知系统或任何需要服务器主动推送数据的场景。通过结合FastAPI的易用性和高效性，你可以快速搭建出功能强大的实时Web服务。

资源推荐

资源详情

资源评论