正如摘要所述，本文将深入探讨在使用 websockets 库构建 WebSocket 服务时，如何解决 websockets.broadcast() 在无限循环中可能导致的阻塞问题，并提供替代方案 asyncio.wait()，帮助开发者构建高性能的 WebSocket 应用。
问题背景与分析
在使用 websockets 库构建实时应用时，经常需要将服务器端的数据广播到所有连接的客户端。一种常见的做法是在一个无限循环中读取数据（例如视频帧），进行处理，然后使用 websockets.broadcast() 将结果广播出去。
然而，当客户端数量增多，或者数据处理过程耗时较长时，websockets.broadcast() 可能会阻塞事件循环，导致客户端无法及时接收到数据，甚至出现连接超时等问题。这是因为 websockets.broadcast() 实际上是一个同步操作，它会依次向每个客户端发送数据，直到所有客户端都发送完毕才返回。
解决方案：使用 asyncio.wait() 进行异步广播
为了解决这个问题，我们可以使用 asyncio.wait() 来实现异步广播。asyncio.wait() 允许我们并发地执行多个异步任务，而不会阻塞事件循环。
以下是使用 asyncio.wait() 替代 websockets.broadcast() 的代码示例：
Server.py:
import websockets
import cv2
import asyncio
import time
def predict(image):
# 模拟耗时的数据处理过程
time.sleep(0.1)
return "test"
async def echo(websocket, path):
global vidCap, i
while True:
ret, image = vidCap.read()
if ret:
start = time.time()
result = predict(image)
# 使用 asyncio.wait() 进行异步广播
await asyncio.wait([ws.send(result) for ws in clients])
end = time.time()
print("exec time：%f s" % (end - start))
else:
# 视频读取完毕，退出循环
break
async def handler(websocket, path):
clients.add(websocket)
try:
await echo(websocket, path)
finally:
clients.remove(websocket)
async def serve():
start_server = await websockets.serve(handler, "localhost", 8765)
await start_server.wait_closed()
if __name__ == '__main__':
vidCap = cv2.VideoCapture(0) # 使用摄像头
clients = set()
asyncio.run(serve())Client.py:
系统特点：技术领先：系统基于被广泛使用的Windows平台开发，集百家之所长，技术领先、功能完备； 快速建店：只需简单设置，3分钟即可以建立一个功能完备的网上商城； 操作简便：软件操作界面由专业设计人员设计，采用人性化的布局，界面规范，操作简捷； 安装方便：只需传到您的虚拟空间即可； HTML编辑器：内置优秀的HTML在线编辑器； 可扩展性：软件构架灵活，考虑未来功能扩充之需要，具有较强的可扩展性
import websockets
import asyncio
import time
async def get_result(uri):
async with websockets.connect(uri) as websocket:
while True:
try:
start = time.time()
recv_text = await websocket.recv()
print(recv_text)
end = time.time()
print("exec：%f s" % (end - start))
except websockets.exceptions.ConnectionClosed:
print("Connection closed")
break
except Exception as e:
print(f"Error: {e}")
break
if __name__ == '__main__':
asyncio.run(get_result("ws://127.0.0.1:8765/ws"))代码解释：
asyncio.wait([ws.send(result) for ws in clients]): 这行代码的关键在于使用列表推导式 [ws.send(result) for ws in clients] 创建了一个包含所有客户端发送任务的列表。然后，asyncio.wait() 并发地执行这些任务，这意味着服务器可以同时向所有客户端发送数据，而不会阻塞事件循环。
异常处理: 在Client.py中添加了websockets.exceptions.ConnectionClosed异常的处理，可以更优雅地处理客户端断开连接的情况。
websockets.broadcast() vs. asyncio.wait()
| 特性 | websockets.broadcast() | asyncio.wait() |
|---|---|---|
| 执行方式 | 同步 | 异步 |
| 阻塞性 | 会阻塞事件循环 | 不会阻塞事件循环 |
| 并发性 | 无 | 支持并发执行多个发送任务 |
| 适用场景 | 客户端数量较少，数据处理速度快 | 客户端数量较多，数据处理速度慢，需要高并发性能的场景 |
总结：
- websockets.broadcast() 简单易用，但在高并发场景下性能较差。
- asyncio.wait() 可以实现异步广播，在高并发场景下性能更优。
注意事项
- 错误处理: 在使用 asyncio.wait() 时，需要注意错误处理。如果某个客户端的发送任务失败，asyncio.wait() 会抛出异常。你需要根据实际情况选择合适的错误处理策略，例如忽略错误、重试或关闭连接。
- 性能优化: 虽然 asyncio.wait() 可以提高并发性能，但在客户端数量非常庞大时，仍然可能成为性能瓶颈。可以考虑使用更高级的广播算法，例如基于发布/订阅模式的消息队列。
- 视频流读取: 示例代码中使用 cv2.VideoCapture(0) 从摄像头读取视频流。 你可以根据实际情况修改为读取本地视频文件或 RTSP 流。请确保安装了 OpenCV 库 (pip install opencv-python).
- 资源释放: 在程序结束时，务必释放视频捕获对象 vidCap.release()，以避免资源泄露。
通过使用 asyncio.wait() 替代 websockets.broadcast()，可以显著提高 WebSocket 服务的并发性能和响应速度，从而构建更稳定、更高效的实时应用。 记住，选择合适的广播方式取决于你的具体应用场景和性能需求。
