You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

基于Python WebSocket的在线二十一点游戏:实现玩家独立内容推送

实现二十一点游戏的玩家专属WebSocket推送方案

当然可以实现你提到的REST式WebSocket URL方案,Python的websockets库完全支持解析连接路径来区分玩家,同时也有其他几种灵活的替代方案,下面是具体实现思路:

一、实现REST式WebSocket路径路由

核心思路是让服务器解析客户端连接的URL路径,为每个连接绑定对应的玩家身份,再通过身份定向推送专属内容:

  1. 服务器端逻辑:
    • 解析客户端连接的路径(如/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())
  1. 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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.19 13:32:09