Python websockets+asyncio实现非阻塞周期发送同时接收数据

代码的报错和功能异常来自三个核心问题：

• asyncio.wait_for() 返回的是协程对象，代码中未加await关键字执行，直接触发「协程从未被等待」的运行时警告
• 事件循环调用方式是Python3.10之前的旧规范，直接使用get_event_loop()在新版本中会触发弃用警告
• 收发任务调度逻辑错误：每次循环同时创建收发任务再串行等待执行，会导致接收逻辑被发送逻辑阻塞，无法实现非阻塞接收客户端不定期上报的数据

将收发逻辑拆分为两个独立的常驻并发任务：发送任务按1秒固定周期推送数据，接收任务永久挂起等待客户端消息，无需设置短超时反复重建任务；使用新版标准的asyncio.run()启动服务，替换旧的事件循环调用逻辑。

修复后的完整可运行代码：

import asyncio
import random
import websockets
import json


async def recv_handler(websocket):
    # 常驻接收协程，非阻塞等待客户端消息
    async for message in websocket:
        # 此处编写收到客户端消息后的业务处理逻辑
        print(f"收到客户端消息: {message}")


async def send_handler(websocket):
    # 周期发送协程，每隔1秒向客户端推送数据
    while True:
        data = [
            {
                "name": "Random Int 1",
                "number": random.randint(0, 1000)
            },
            {
                "name": "Random Int 2",
                "number": random.randint(1001, 2000)
            },
            {
                "name": "Random Int 3",
                "number": random.randint(2001, 3000)
            }
        ]
        await websocket.send(json.dumps(data))
        await asyncio.sleep(1)


async def connection_handler(websocket):
    # 连接建立后同时启动收发两个任务，并发执行互不阻塞
    recv_task = asyncio.create_task(recv_handler(websocket))
    send_task = asyncio.create_task(send_handler(websocket))
    # 任意一个任务结束（通常是连接断开），立刻取消剩余任务避免资源泄漏
    done, pending = await asyncio.wait(
        [recv_task, send_task],
        return_when=asyncio.FIRST_COMPLETED
    )
    for task in pending:
        task.cancel()


async def run_server():
    async with websockets.serve(connection_handler, "localhost", 3500):
        await asyncio.Future()  # 永久保持服务运行


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

• 接收逻辑使用websockets自带的异步迭代器实现，无需手动编写recv()加超时的循环逻辑，客户端发送消息时会立刻触发处理，全程非阻塞，完全不会影响发送任务的1秒执行周期
• 收发任务仅在客户端连接建立时创建一次，连接断开时自动回收，不会出现反复创建任务导致的协程泄漏问题
• 采用Python3.7+官方推荐的asyncio.run()方式启动服务，彻底消除旧版事件循环写法带来的弃用警告
• 发送间隔调整为1秒，匹配需求中的周期推送要求

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