Python 3.6 Websocket客户端监听消息时触发ConnectionClose异常

我经常碰到开发者在封装websockets客户端类时踩这个坑，尤其是当循环逻辑没处理好连接生命周期的时候。结合你描述的ConnectionClose异常问题，咱们先梳理常见的诱因，再给你一个修正后的可运行示例代码。

首先，先假设你的原始代码大概是这样（因为你贴的代码没写完）：

import websockets
import asyncio

class WebSocketClient:
    def __init__(self, url):
        self.url = url
        self.websocket = None

    async def connect(self):
        self.websocket = await websockets.connect(self.url)

    async def receive_message(self):
        while True:
            message = await self.websocket.recv()
            print(f"Received: {message}")

这种写法的核心问题是：当服务器主动断开连接、网络中断时，recv()会直接抛出ConnectionClose异常，而循环里没有任何捕获处理，导致整个监听逻辑直接崩溃。

修正后的完整代码 #

import websockets
import asyncio
from websockets.exceptions import ConnectionClose, ConnectionClosedOK, ConnectionClosedError

class WebSocketClient:
    def __init__(self, url, reconnect_interval=5):
        self.url = url
        self.websocket = None
        self.reconnect_interval = reconnect_interval  # 重连间隔（秒）
        self.running = False  # 控制客户端生命周期的标志

    async def connect(self):
        """包含自动重连逻辑的连接方法"""
        while self.running:
            try:
                self.websocket = await websockets.connect(self.url)
                print("✅ 成功连接到服务器")
                # 连接建立后进入消息接收循环
                await self._receive_loop()
            except ConnectionClosedOK:
                print("ℹ️ 连接被正常关闭")
            except ConnectionClosedError as e:
                print(f"⚠️ 连接异常关闭: {e}")
            except ConnectionClose as e:
                print(f"❌ 连接意外中断: {e}")
            except Exception as e:
                print(f"🆘 发生未知错误: {e}")
            
            # 如果客户端还在运行状态，等待后尝试重连
            if self.running:
                print(f"⏳ {self.reconnect_interval}秒后尝试重连...")
                await asyncio.sleep(self.reconnect_interval)

    async def _receive_loop(self):
        """内部消息接收循环，专门处理消息和连接异常"""
        while self.running and self.websocket.open:
            try:
                message = await self.websocket.recv()
                # 把消息处理逻辑抽成独立方法，方便你扩展业务
                await self.handle_message(message)
            except (ConnectionClosedOK, ConnectionClosedError):
                # 正常/异常关闭时退出循环，触发外层重连逻辑
                break
            except ConnectionClose as e:
                print(f"❌ 接收消息时出错: {e}")
                break

    async def handle_message(self, message):
        """自定义消息处理方法，你可以在这里写业务逻辑"""
        print(f"📥 收到消息: {message}")

    async def send_message(self, message):
        """安全的消息发送方法，先检查连接状态"""
        if self.websocket and self.websocket.open:
            await self.websocket.send(message)
        else:
            print("❌ 无法发送消息：WebSocket未连接")

    async def start(self):
        """启动客户端"""
        self.running = True
        await self.connect()

    async def stop(self):
        """优雅停止客户端"""
        self.running = False
        if self.websocket and self.websocket.open:
            await self.websocket.close()
        print("🛑 客户端已停止")

# 使用示例
async def main():
    client = WebSocketClient("ws://your-server-url")
    # 异步启动客户端，不阻塞主逻辑
    client_task = asyncio.create_task(client.start())
    
    # 示例：连接建立后发送一条测试消息
    await asyncio.sleep(1)
    await client.send_message("Hello from Python Client")
    
    # 运行30秒后停止客户端
    await asyncio.sleep(30)
    await client.stop()
    await client_task

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

关键修复点说明 #

• 生命周期管理：新增running标志，让你能优雅地启动/停止客户端，避免异步逻辑失控
• 异常分层处理：区分了正常关闭、异常关闭、通用连接异常，每种情况做对应的处理，不会一遇到异常就崩溃
• 自动重连机制：连接断开后会自动尝试重连，你可以通过reconnect_interval调整间隔
• 模块化拆分：把消息接收、消息处理、发送逻辑拆成独立方法，后续扩展业务更方便
• 安全发送检查：发送消息前先验证连接状态，避免在断开时发送消息报错

额外注意事项 #

• 永远不要忽略recv()抛出的异常：websockets库会在连接状态变化时主动抛出这些异常，你必须捕获并处理
• 如果需要处理重连后的状态恢复（比如重新发送未确认的消息），可以在connect方法中添加相关逻辑
• 异步代码中尽量避免阻塞操作，所有耗时逻辑都要封装成异步方法

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