前面我们发送的大多数都是text类型的消息,对于text消息来说,后端处理出来要麻烦的多,那么我们可以不可以传递json格式的数据,对于前后端来说都比较友好,答案是肯定的,我们需要做下处理。
首先,我们在websocket管理器中增加处理json格式消息的方法:
from typing import List, Dict from starlette.websockets import WebSocket class ConnectionManager: def __init__(self): """存放链接""" self.active_connections: List[Dict[str, WebSocket]] = [] async def connect(self, user: str, ws: WebSocket): """链接""" await ws.accept() self.active_connections.append({"user": user, "ws": ws}) async def disconnect(self, user: str, ws: WebSocket): """断开链接,移除""" self.active_connections.remove({"user": user, "ws": ws}) @staticmethod async def send_personal_message(message: str, ws: WebSocket): """发送所有人消息""" await ws.send_text(message) async def send_other_message(self, message: dict, user: str): """发送个人消息""" for coon in self.active_connections: if coon["user"] == user: await coon["ws"].send_json(message) async def broadcast(self, data: str): """广播""" for conn in self.active_connections: await conn["ws"].send_text(data) async def broadcast_json(self, data: dict): """广播json数据""" for conn in self.active_connections: await conn["ws"].send_json(data)
接口中如何接收json数据呢,新增接口如下:
@app.websocket("/ws/json/{user}") async def websocket_json_data( websocket: WebSocket, user: str, cookie_or_token: str = Depends(get_cookie_or_token) ): """发送接收json数据(前面一对一接口其实就是发送的json数据)""" await ws_manager.connect(user, websocket) try: while True: data = await websocket.receive_json() send_user = data["username"] if send_user: await ws_manager.send_other_message(data, send_user) else: await ws_manager.broadcast_json(data) except WebSocketDisconnect as e: await ws_manager.disconnect(user, websocket)
前端也要相对应的修改:
Chat WebSocket 聊天
部署看效果,因为发送给后端的数据格式是前端拼的json,所以页面上只需要填文本即可: