基于Python WebSocket的在线二十一点游戏:实现玩家独立内容推送
实现二十一点游戏的玩家专属WebSocket推送方案
当然可以实现你提到的REST式WebSocket URL方案,Python的websockets库完全支持解析连接路径来区分玩家,同时也有其他几种灵活的替代方案,下面是具体实现思路:
一、实现REST式WebSocket路径路由
核心思路是让服务器解析客户端连接的URL路径,为每个连接绑定对应的玩家身份,再通过身份定向推送专属内容:
- 服务器端逻辑:
- 解析客户端连接的路径(如
/player_one),验证合法身份 - 维护字典存储玩家身份与WebSocket连接的映射
- 推送消息时,根据玩家身份发送专属牌面信息(比如给玩家1推送自己的全牌+玩家2的明牌)
- 解析客户端连接的路径(如
import asyncio import websockets from websockets.server import WebSocketServerProtocol # 存储玩家连接:key为玩家标识,value为WebSocket连接对象 player_connections = {} async def handle_connection(websocket: WebSocketServerProtocol): # 解析连接路径获取玩家身份 player_id = websocket.path.strip("/") if player_id not in ["player_one", "player_two"]: await websocket.close(code=1008, reason="无效的玩家路径") return # 注册玩家连接 player_connections[player_id] = websocket print(f"玩家 {player_id} 已连接") try: # 模拟游戏状态推送逻辑 while True: if player_id == "player_one": # 玩家1专属内容:自己的全牌 + 玩家2的明牌 game_data = { "own_cards": ["♠A", "♥5", "?"], # 包含盖牌 "opponent_cards": ["♣Q", "?"], # 仅显示明牌 "available_actions": ["hit", "stand"] } else: # 玩家2专属内容:自己的全牌 + 玩家1的明牌 game_data = { "own_cards": ["♦10", "♣7", "?"], "opponent_cards": ["♠A", "?"], "available_actions": ["hit", "stand"] } await websocket.send(str(game_data)) await asyncio.sleep(2) except websockets.exceptions.ConnectionClosed: print(f"玩家 {player_id} 已断开连接") del player_connections[player_id] async def main(): async with websockets.serve(handle_connection, "localhost", 8001): await asyncio.Future() # 保持服务器持续运行 if __name__ == "__main__": asyncio.run(main())
- Textual客户端示例(玩家1):
import asyncio from textual.app import App, ComposeResult from textual.widgets import Static import websockets class BlackjackPlayerOne(App): async def on_mount(self) -> None: try: # 连接到专属路径 self.websocket = await websockets.connect("ws://localhost:8001/player_one") asyncio.create_task(self.receive_game_data()) except Exception as e: self.query_one(Static).update(f"连接失败:{e}") async def receive_game_data(self): while True: data = await self.websocket.recv() # 更新终端UI展示牌面 self.query_one(Static).update(f"玩家1视角:\n{data}") def compose(self) -> ComposeResult: yield Static("等待游戏数据...") if __name__ == "__main__": BlackjackPlayerOne().run()
二、其他可行替代方案
如果不想用路径区分,还有几种更灵活的方式:
- 查询参数标识:客户端连接时使用
ws://localhost:8001?player=one,服务器通过websocket.query_params.get("player")获取身份,逻辑与路径方案一致。 - 连接后身份认证:所有客户端先连接到统一地址
ws://localhost:8001,连接成功后发送身份消息(如{"type": "auth", "player": "one"}),服务器验证后绑定身份,后续推送对应内容,适合动态分配玩家身份的场景。 - 房间模式:将每个玩家视为独立房间的成员,服务器维护房间与玩家的映射关系,推送时按房间定向发送内容。
核心逻辑都是服务器能够识别每个连接对应的玩家身份,从而推送专属的牌面信息,你可以根据自己的开发习惯选择合适的方案。
内容的提问来源于stack exchange,提问作者wavesinaroom
相关产品推荐
相关产品推荐

