前端页面：
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>websocket</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<input type="text" id="nickname" autocomplete="off" placeholder="input your nickanme"/>
<button onclick="connect(event)" id="connect_btn">connect</button>
<button onclick="disconnect(event)">disconnect</button>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = null;
var nickname = document.getElementById('nickname');
function connect() {
if(nickname.value.trim()==''){return;}
ws = new WebSocket(`ws://localhost:9000/ws/` + nickname.value);
ws.onopen = function () {
nickname.readOnly = true;
document.getElementById('connect_btn').disabled = true;
console.log('WS open')
};
ws.onclose = function () {
console.log('WS close');
};
ws.onerror = function (event) {
console.log(event.data);
};
ws.onmessage = function (event) {
var messages = document.getElementById('messages');
var message = document.createElement('li');
var content = document.createTextNode(event.data);
message.appendChild(content);
messages.appendChild(message);
};
}
function sendMessage(event) {
var input = document.getElementById("messageText");
ws.send(input.value);
input.value = '';
event.preventDefault();
}
function disconnect() {
ws.close()
}
</script>
</body>
</html>后端代码：
import uvicorn
from typing import Optional, List
from fastapi import Cookie, Depends, FastAPI, Query, WebSocket, status, WebSocketDisconnect
app = FastAPI()
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_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str, exclude):
for connection in self.active_connections:
if connection is exclude:
continue
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{nickname}")
async def websocket_endpoint(websocket: WebSocket, nickname: str):
await manager.connect(websocket)
try:
await manager.send_personal_message("【你 加入了群聊】", websocket)
await manager.broadcast(f"【{nickname} 加入了群聊】", exclude=websocket)
while True:
data = await websocket.receive_text()
await manager.send_personal_message(f"你: {data}", websocket)
await manager.broadcast(f"{nickname}: {data}", websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"【{nickname} 离开了群聊】", None)
if __name__ == '__main__':
uvicorn.run(app="websocket:app", host='0.0.0.0', port=9000, debug=True, workers=1)
tips：如果后端代码报错：Unsupported upgrade request ，需要先pip uninstall uvicorn再pip install uvicorn[standard]
运行效果图