如何为异步WebSocket添加无服务器Ping接收时的超时重连机制

要解决长时间无数据时自动断连重连的问题，你需要添加连接活跃度检测逻辑，核心思路是跟踪最后一次接收数据（包括服务器发送的Ping帧）的时间，一旦超过设定阈值就主动断开连接，触发外层的自动重连逻辑。

具体修改步骤如下：

1. 引入asyncio和time模块，用于超时检查和时间戳记录
2. 在连接成功后，启动一个后台任务定期检查连接活跃度
3. 监听WebSocket的所有帧（包括Ping帧），更新最后活动时间
4. 超时触发时主动关闭连接，让外层的async for循环自动重连

修改后的代码示例：

import asyncio
import time
import websockets

async def wsrun(uri, timeout_seconds=300):  # 设置5分钟超时，可根据实际调整
    async for websocket in websockets.connect(uri):
        last_active_time = time.time()
        # 标记是否需要停止检查任务
        stop_check = asyncio.Event()

        async def check_timeout():
            nonlocal last_active_time
            while not stop_check.is_set():
                await asyncio.sleep(10)  # 每10秒检查一次
                if time.time() - last_active_time > timeout_seconds:
                    print("长时间未接收数据，触发断连重连")
                    await websocket.close()
                    stop_check.set()

        # 启动超时检查后台任务
        check_task = asyncio.create_task(check_timeout())

        try:
            # 循环接收所有帧（包括Ping、Pong、Text/Binary数据）
            async for message in websocket:
                # 更新最后活动时间，不管是业务数据还是Ping帧
                last_active_time = time.time()
                # 这里处理你的业务数据逻辑
                # (websocket processing code here)
                print(f"收到数据: {message[:50]}...")  # 示例打印

        except websockets.exceptions.ConnectionClosed:
            print("连接已关闭")
        except Exception as error:
            print(f"发生异常: {error}")
        finally:
            # 停止超时检查任务
            stop_check.set()
            await check_task

关键说明： #

• timeout_seconds：设置超时阈值，比如5分钟（300秒），可根据服务器Ping间隔调整，建议比服务器Ping周期长一点
• 后台任务check_timeout：定期检查最后活动时间，超时则主动关闭连接
• 监听所有message：websockets库会自动处理Ping帧并回复Pong，这里通过接收所有帧来更新活动时间，确保服务器的Ping能被检测到
• 外层的async for websockets.connect(uri)：在连接关闭后会自动尝试重连，符合你原来的重连逻辑

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