Python对接Bybit WebSocket报socket is already closed异常求助

• 连接建立是异步过程：调用connectWS_public()只是启动了后台线程执行WebSocket连接逻辑，主线程不会等待TCP握手、WebSocket协议升级完成就继续往下执行。此时x.ws_public.sock可能刚完成初始化但实际未连通，甚至连接过程中出错直接被关闭，此时调用send()方法必然触发socket is already closed异常。
• 轮询发送逻辑错误：编写的while (x.ws_public.sock)循环会在sock对象存在时无限重复发送相同的订阅请求，既可能在连接未就绪时触发写错误，也会因为高频重复无效请求被交易所主动断开连接。
• 回调绑定缺失：初始化WebSocketApp时没有绑定on_close回调，连接被关闭时无法第一时间拿到断开状态、错误码，没法准确判断连接可用性。
• 回调参数写法错误：原代码中on_ping、on_close的入参和websocket库要求的参数列表不匹配，回调触发时会抛出参数错误，直接导致连接断开。

• 所有订阅、鉴权等需要向服务端发送数据的操作，必须放在on_open回调内执行，只有on_open触发时才代表WebSocket连接完全建立、处于可读写状态，禁止在主线程轮询sock状态提前发数据。
• 补全on_close回调绑定，修正所有回调的入参写法，方便排查连接断开原因，如需断线重连可在on_close逻辑中实现。
• 去掉主线程无脑循环发送订阅的逻辑，同一连接下相同订阅只需要发送一次即可，重复发送无意义还会触发服务端拦截。
• 保留当前20秒ping间隔、10秒pong超时的配置，符合交易所心跳规则，不需要额外自定义心跳逻辑和内置心跳冲突。

修正后的类实现代码 #

from datetime import datetime as dt
import threading
import websocket
import json
import time

class Bybit_WS_test():

    def __init__ (self):
        self.api_url_public = 'wss://stream.bybit.com/realtime_public'
        self.api_url_private = 'wss://stream-testnet.bybit.com/realtime_private'
        self.api_key = ''
        self.api_secret = ''

        self.ping_interval = 20
        self.ping_timeout = 10

        self.ws_public = None
        self.ws_private = None
        # 统一维护公共频道订阅列表
        self.public_sub_topics = ["orderBookL2_25.BTCUSD"]

    def on_message(self, ws, message):
        data = json.loads(message)
        print('Received message:')
        print(data)

    def on_error(self, ws, error):
        print(f'websocket error: {error}')
    
    def on_close(self, ws, close_status_code, close_msg):
        print(f"Closing websocket connection, status code: {close_status_code}, msg: {close_msg}")
        # 需要断线重连可在此处加延迟重连逻辑

    def on_pong(self, ws, message):
        print('Received pong')

    def on_open(self, ws):
        print('Websocket opened')
        # 连接建立成功后再发送订阅请求
        sub_req = json.dumps({"op": "subscribe", "args": self.public_sub_topics})
        ws.send(sub_req)

    def on_ping(self, ws, message):
        dt_string = dt.now().strftime("%d/%m/%Y %H:%M:%S")
        print(f'Received ping @ {dt_string}')

    def connectWS_public(self):
        self.ws_public = websocket.WebSocketApp(
            url = self.api_url_public,
            on_message = self.on_message,
            on_error = self.on_error,
            on_close = self.on_close,
            on_ping= self.on_ping,
            on_pong= self.on_pong,
            on_open= self.on_open
        )
        self.wst_public = threading.Thread(target=lambda: self.ws_public.run_forever(
            ping_interval=self.ping_interval, 
            ping_timeout=self.ping_timeout
        ))
        self.wst_public.daemon = True
        self.wst_public.start()

修正后的测试代码 #

from Bybit_api.Bybit_ws import Bybit_WS_test
import time

if __name__ == '__main__':
    x = Bybit_WS_test()
    x.connectWS_public()
    # 主线程阻塞保活即可，不需要轮询发送请求
    while True:
        time.sleep(1)

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