Python WebSocket客户端Spring Boot服务重启后重连异常求助

核心解决思路 #

• 每次重连创建全新的WebSocket客户端实例，避免旧实例的无效状态污染新连接
• 给on_open等回调函数添加异常捕获，防止单个回调的异常导致重连逻辑崩溃
• 采用指数退避算法控制重连间隔，平衡重连及时性和服务端压力

具体实现（基于websocket-client库） #

以下是封装好的可重连客户端，包含异常防护和重连调度逻辑：

import websocket
import time
from typing import Callable

class ReconnectingWebSocket:
    def __init__(self, url: str, on_message: Callable, on_error: Callable = None, on_close: Callable = None):
        self.url = url
        self.on_message = on_message
        self.on_error = on_error or self._default_on_error
        self.on_close = on_close or self._default_on_close
        self._ws = None
        self._reconnect_delay = 1  # 初始重连间隔（秒）
        self._max_reconnect_delay = 30  # 最大重连间隔上限
        self._running = True

    def _default_on_error(self, ws, error):
        print(f"WebSocket错误: {str(error)}")

    def _default_on_close(self, ws, close_status_code, close_msg):
        print(f"WebSocket关闭 | 状态码: {close_status_code}, 消息: {close_msg}")
        self._schedule_reconnect()

    def _on_open(self, ws):
        try:
            # 替换为你实际的on_open逻辑，比如发送初始化指令
            ws.send("CLIENT_INIT")
            print("连接成功，已发送初始化消息")
            self._reconnect_delay = 1  # 连接成功后重置重连间隔
        except Exception as e:
            print(f"on_open执行异常: {str(e)}")
            # 主动关闭连接，触发重连流程
            ws.close()

    def _connect(self):
        if not self._running:
            return
        try:
            self._ws = websocket.WebSocketApp(
                self.url,
                on_open=self._on_open,
                on_message=self.on_message,
                on_error=self.on_error,
                on_close=self.on_close
            )
            self._ws.run_forever()
        except Exception as e:
            print(f"连接尝试失败: {str(e)}")
            self._schedule_reconnect()

    def _schedule_reconnect(self):
        if not self._running:
            return
        print(f"{self._reconnect_delay}秒后尝试重连...")
        time.sleep(self._reconnect_delay)
        # 指数退避：间隔翻倍，不超过最大值
        self._reconnect_delay = min(self._reconnect_delay * 2, self._max_reconnect_delay)
        self._connect()

    def stop(self):
        """优雅停止客户端，终止重连调度"""
        self._running = False
        if self._ws:
            self._ws.close()

# 使用示例
def handle_server_message(ws, message):
    print(f"收到服务端消息: {message}")

if __name__ == "__main__":
    ws_client = ReconnectingWebSocket(
        url="ws://localhost:8080/your-websocket-endpoint",
        on_message=handle_server_message
    )
    ws_client._connect()

关键细节说明 #

1. 全新实例重连：每次重连都会创建新的WebSocketApp对象，彻底避免旧实例的连接状态、句柄等残留问题，这是解决重连时on_open异常的核心。
2. on_open异常防护：在_on_open方法内捕获所有异常，一旦出现错误主动关闭连接，触发重连流程，不会让客户端卡在异常状态。
3. 指数退避重连：初始重连间隔1秒，每次失败后翻倍，直到达到30秒上限，既保证服务端恢复后能快速重连，又避免频繁重试给服务端或网络造成压力。
4. 运行状态控制：通过_running标记实现优雅停止，调用stop()方法后会终止后续的重连调度。

场景适配验证 #

• 服务端停止/恢复：服务端停止时客户端触发on_close，进入重连调度；服务端恢复后，客户端会在下次重连尝试时成功建立连接，on_open正常执行。
• 网络异常：网络中断时WebSocket底层会检测到连接失效，触发on_error或on_close，自动进入重连流程，网络恢复后自动连接。
• on_open重连异常：如果重连时on_open内的操作（如发送初始化消息）失败，会被捕获并主动关闭连接，重新进入重连调度，不会导致客户端崩溃或停止重试。

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