FastAPI WebSocket
FastAPI WebSocket
后端搭建websocket服务器
from fastapi import WebSocket
@app.websocket('/ws')
async def websocket_dep(websocket:WebSocket):
await websocket.accept()
while True:
data =await websocket.receive_text()
websocket.send_text() # 响应内容前端发送 请求 和接收内容
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Document</title>
<style>
ul {
list-style-type: none;
}
li {
position: relative;
height: 40px;
}
</style>
</head>
<body>
<h1>Websocket</h1>
<form action="">
<input type="text" name="mess" id="msg">
<input type="button" value="submit" id="btn">
</form>
<ul id='rst'>
<li><span style="float: left;">aaaaaa</span></li>
<li><span style="float: right;">bbbbbb</span></li>
</ul>
<script>
var ws = new WebSocket('ws://localhost:8000/ws');
ws.onmessage = function(event){
li = document.createElement('li');
span = document.createElement('span');
span.innerHTML = event.data; // 接收消息
span.style.float = 'left';
li.appendChild(span);
rst = document.getElementById('rst');
rst.appendChild(li);
}
var msg = document.getElementById('msg');
var btn = document.getElementById('btn');
btn.onclick = function(event){
ws.send(msg.value) // 发送消息
li = document.createElement('li');
span = document.createElement('span');
span.innerHTML = msg.value;
span.style.float = 'right';
li.appendChild(span);
msg.value = '';
rst = document.getElementById('rst');
rst.appendChild(li);
// event.preventDefault(); // 不跳转
return false // 不跳转
}
</script>
</body>
</html>托管
from fastapi import WebSocket
# 用于管理websocket的连接
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def send_json(self,data,websocket: WebSocket):
await websocket.send_json(data)
# 给所有连接发送文本
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")Last updated