Python WebSocket服务器握手后如何向客户端发送动态消息？

这是一个实现Python服务器向Javascript客户端页面发送硬编码消息的小型WebSocket客户端与服务器POC，当前需求是在服务器运行后，向客户端发送后续的动态消息（而非硬编码内容）。

客户端HTML代码 #

<!DOCTYPE html> 
<html lang="en">
<body> See console for messages </body>
<script>
// 创建WebSocket连接
const socket = new WebSocket('ws://localhost:8000');

// 监听连接建立事件
socket.addEventListener('open', function (event) {
    socket.send('Connection Established');
});

// 接收服务器消息并打印到控制台
socket.addEventListener('message', function (event) { 
    console.log(event.data);
});
 
</script>
</html>

原始Python服务器代码 #

import asyncio 
import websockets

# 处理每个客户端连接
async def handler(websocket, path):
    await websocket.send("message from websockets server")

# 启动WebSocket服务器
start_server = websockets.serve(handler, "localhost", 8000)

# 运行异步事件循环
asyncio.get_event_loop().run_until_complete(start_server) 
asyncio.get_event_loop().run_forever()

现有代码能成功发送初始硬编码消息，但无法触发后续动态消息。直接将socket存入全局变量的方案不可行——因为服务器处于持续异步循环中，无法直接在外部同步调用send方法。

解决方案 #

方法1：维护活跃连接集合

通过全局集合保存所有在线的WebSocket连接，需要发送动态消息时遍历集合批量发送，同时处理连接关闭的清理逻辑：

import asyncio
import websockets

# 存储所有活跃的WebSocket连接
active_connections = set()

async def handler(websocket, path):
    # 将新连接加入集合
    active_connections.add(websocket)
    try:
        # 发送初始消息
        await websocket.send("message from websockets server")
        # 保持连接存活，直到客户端主动断开
        await websocket.wait_closed()
    finally:
        # 连接关闭后从集合移除，避免无效连接占用资源
        active_connections.remove(websocket)

async def send_dynamic_message(message):
    # 给所有活跃连接并发发送消息
    if active_connections:
        await asyncio.gather(*[conn.send(message) for conn in active_connections])

# 示例：模拟业务逻辑定时触发动态消息
async def periodic_sender():
    while True:
        await asyncio.sleep(5)
        await send_dynamic_message(f"动态消息：{asyncio.get_event_loop().time()}")

async def main():
    # 启动WebSocket服务器
    await websockets.serve(handler, "localhost", 8000)
    # 同时运行消息发送任务和服务器循环
    await asyncio.gather(periodic_sender())

asyncio.run(main())

如果你的主应用需要触发消息，只需在异步上下文内调用send_dynamic_message(message)即可。若主应用是同步线程，可使用asyncio.run_coroutine_threadsafe(send_dynamic_message(msg), loop)来跨线程调用。

方法2：使用消息队列解耦

通过异步队列作为中间层，主应用只需往队列中放入消息，服务器的每个连接会自动监听队列并发送消息，适合主应用与WebSocket服务器不在同一异步上下文的场景：

import asyncio
import websockets

# 全局消息队列，用于接收待发送的动态消息
message_queue = asyncio.Queue()

async def handler(websocket, path):
    await websocket.send("message from websockets server")
    
    # 启动子任务监听队列，给当前连接发送消息
    async def queue_listener():
        while True:
            msg = await message_queue.get()
            await websocket.send(msg)
    
    # 同时等待连接关闭和队列监听任务，任一完成则退出
    await asyncio.gather(websocket.wait_closed(), queue_listener())

async def main():
    await websockets.serve(handler, "localhost", 8000)
    
    # 示例：模拟外部业务触发消息
    async def simulate_trigger():
        while True:
            await asyncio.sleep(3)
            await message_queue.put(f"队列推送的动态消息：{asyncio.get_event_loop().time()}")
    
    await asyncio.gather(simulate_trigger())

asyncio.run(main())

额外提示

• 若需要定向发送给特定客户端，可以将集合改为字典，用客户端标识（如用户ID）作为键，WebSocket连接作为值，发送时根据标识找到对应连接即可。
• 务必处理连接异常关闭的情况，避免无效连接占用资源或导致发送失败。

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