10.3 双向实时通信实现
FastAPI 双向实时通信实现教程 - 新手快速入门
本教程详细讲解如何使用FastAPI和WebSocket实现双向实时通信。涵盖基础概念、安装步骤、代码示例及解释，适合开发者快速上手，构建聊天应用和实时通知系统。
推荐工具
FastAPI 实现双向实时通信教程
介绍
双向实时通信允许服务器和客户端之间同时发送和接收数据，无需等待响应。在FastAPI中，WebSocket是实现此功能的理想选择，特别适用于聊天应用、实时监控和游戏开发等场景。本教程将引导您从零开始，使用FastAPI的WebSocket支持轻松实现双向通信。
前提条件
- Python 3.7+：确保已安装Python。
- 安装FastAPI和Uvicorn：通过pip安装必需库：
pip install fastapi uvicorn - 基础异步编程知识：FastAPI使用
async/await进行异步处理，简单了解即可。
实现步骤
1. 创建FastAPI应用
首先，导入必要的模块并初始化一个FastAPI应用。
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
2. 定义WebSocket端点
使用@app.websocket装饰器创建一个WebSocket端点，处理连接和消息。
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept() # 接受WebSocket连接
while True:
data = await websocket.receive_text() # 接收客户端消息
await websocket.send_text(f"Echo: {data}") # 发送回显消息
代码解释：
WebSocket类：处理WebSocket连接的实例。accept()：建立连接，必须首先调用。receive_text()：异步接收文本消息。send_text()：异步发送文本消息。- 无限循环保持连接活跃，实现实时交互。
3. 完整示例：简单聊天服务器
为了直观展示，这里提供一个包含HTML客户端的完整示例。
服务器代码 (main.py)：
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
# HTML客户端页面
html = """
<!DOCTYPE html>
<html>
<head>
<title>FastAPI WebSocket Chat</title>
</head>
<body>
<h1>WebSocket 聊天示例</h1>
<form onsubmit="sendMessage(event)">
<input type="text" id="messageText" placeholder="输入消息..." autocomplete="off"/>
<button>发送</button>
</form>
<ul id="messages"></ul>
<script>
// 连接到WebSocket服务器
var ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
var messages = document.getElementById('messages');
var message = document.createElement('li');
message.textContent = event.data;
messages.appendChild(message);
};
function sendMessage(event) {
var input = document.getElementById("messageText");
ws.send(input.value);
input.value = '';
event.preventDefault(); // 阻止表单默认提交
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"服务器回复: {data}")
except Exception as e:
print(f"连接错误: {e}") # 简单错误处理
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
4. 运行和测试
- 保存代码为
main.py。 - 在终端运行服务器：
uvicorn main:app --reload - 打开浏览器，访问
http://localhost:8000，在输入框中发送消息，页面会显示服务器回复。
扩展和优化
- 处理多个客户端：使用列表存储所有WebSocket连接，实现广播功能。
connections = [] @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() connections.append(websocket) try: while True: data = await websocket.receive_text() for connection in connections: await connection.send_text(f"广播: {data}") except: connections.remove(websocket) - 错误处理：添加更多异常处理，如连接断开或无效数据。
- 添加认证：结合FastAPI的依赖注入，验证WebSocket连接。
总结
通过本教程，您学会了如何在FastAPI中利用WebSocket实现双向实时通信。FastAPI的内置支持让WebSocket开发变得简单高效，适合新手快速构建实时应用。扩展部分提供了进阶思路，鼓励您探索更多功能。
如有问题，欢迎查阅FastAPI官方文档。祝您学习顺利！
开发工具推荐