树莓派PyGame集成WebSocket服务端 实现HTML客户端控制游戏对象

解决方案1：多线程+消息队列（推荐，易实现） #

这个方案把WebSocket服务和PyGame逻辑拆到两个线程运行，通过线程安全的队列传递数据，完全避免循环阻塞问题，对原有代码改动最小。

import websockets
import asyncio
import threading
import queue
import pygame

# 初始化线程安全的消息队列，用于WebSocket线程向PyGame主线程传消息
msg_queue = queue.Queue(maxsize=10)

# ---------------------- WebSocket服务逻辑（子线程运行） ----------------------
async def echo(websocket, path):
    async for message in websocket:
        print(f"[Received] {message}")
        # 把消息放入队列，满了就丢弃最旧的消息避免堆积
        if msg_queue.full():
            msg_queue.get_nowait()
        msg_queue.put_nowait(message)
        await websocket.send("Success")

async def ws_server():
    async with websockets.serve(echo, "192.168.4.1", 6677):
        await asyncio.Future()  # 永久运行服务

def run_ws_server():
    # 子线程单独启动asyncio事件循环
    asyncio.run(ws_server())

# 启动WebSocket守护线程，主程序退出时会自动销毁
ws_thread = threading.Thread(target=run_ws_server, daemon=True)
ws_thread.start()

# ---------------------- PyGame游戏逻辑（主线程运行） ----------------------
pygame.init()
HEIGHT = 320
WIDTH = 480
screen = pygame.display.set_mode([WIDTH, HEIGHT])
# 圆形初始坐标
circle_x = int(WIDTH/2)
circle_y = int(HEIGHT/2)

running = True
while running:
    # 处理PyGame系统事件
    for event in pygame.event.get():
        if event.type == pygame.QUIT:
            running = False

    # 非阻塞处理WebSocket收到的消息
    while not msg_queue.empty():
        msg = msg_queue.get_nowait()
        # 示例逻辑：根据收到的指令修改圆形位置
        if msg == "left":
            circle_x -= 10
        elif msg == "right":
            circle_x += 10
        elif msg == "up":
            circle_y -= 10
        elif msg == "down":
            circle_y += 10
        # 边界检测避免圆形移出屏幕
        circle_x = max(30, min(circle_x, WIDTH-30))
        circle_y = max(30, min(circle_y, HEIGHT-30))

    # 渲染画面
    screen.fill((255,255,255))
    pygame.draw.circle(screen, (0,0,255), (circle_x, circle_y), 30)
    pygame.display.flip()

pygame.quit()

你可以把前端测试页的按钮替换为上下左右四个按钮，分别发送对应指令字符串，即可直接控制圆形移动。

解决方案2：单线程协程整合 #

如果不想引入多线程，可以把PyGame循环改造为异步任务，每次循环主动让出执行权给asyncio处理WebSocket事件：

import websockets
import asyncio
import pygame

# 存储最新收到的WebSocket消息
latest_msg = ""

async def echo(websocket, path):
    global latest_msg
    async for message in websocket:
        print(f"[Received] {message}")
        latest_msg = message
        await websocket.send("Success")

async def game_loop():
    global latest_msg
    pygame.init()
    HEIGHT = 320
    WIDTH = 480
    screen = pygame.display.set_mode([WIDTH, HEIGHT])
    circle_x = int(WIDTH/2)
    circle_y = int(HEIGHT/2)

    running = True
    while running:
        # 处理PyGame系统事件
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False

        # 处理控制指令
        if latest_msg:
            if latest_msg == "left":
                circle_x -=10
            elif latest_msg == "right":
                circle_x +=10
            elif latest_msg == "up":
                circle_y -=10
            elif latest_msg == "down":
                circle_y +=10
            circle_x = max(30, min(circle_x, WIDTH-30))
            circle_y = max(30, min(circle_y, HEIGHT-30))
            latest_msg = ""

        # 画面渲染
        screen.fill((255,255,255))
        pygame.draw.circle(screen, (0,0,255), (circle_x, circle_y), 30)
        pygame.display.flip()

        # 关键：让出执行权给asyncio事件循环处理WebSocket请求，填1/60可锁60帧
        await asyncio.sleep(0)

    pygame.quit()

async def main():
    # 同时启动WebSocket服务和游戏循环
    async with websockets.serve(echo, "192.168.4.1", 6677):
        await game_loop()

if __name__ == "__main__":
    asyncio.run(main())

该方案没有线程切换开销，但要注意游戏逻辑中不能有长时间阻塞的操作，否则会导致WebSocket消息响应延迟。

内容的提问来源于stack exchange，提问作者nluep