无法使用 websocket：websocket._exceptions.WebSocketConnectionClosedException：套接字已关闭答案

【问题标题】：Failed to use websocket : websocket._exceptions.WebSocketConnectionClosedException: socket is already closed无法使用 websocket：websocket._exceptions.WebSocketConnectionClosedException：套接字已关闭
【发布时间】：2022-12-21 10:37:06
【问题描述】：

感谢您入住！我有一个关于 websocket 连接的问题。

这是我用来订阅交易所的 websocket 的 websocket 类，其中连接是在 self.connectWS_public() 函数中建立的。

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

        return

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

    def on_error(self, ws, error):
        print(f'webscoket error: {error}')
    
    def on_close(self, ws):
        print("Closing websocket connection...")

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

    def on_open(self, ws):
        print('Websocket opened')

    def on_ping(self, message):
        dt_string = dt.now().strftime("%d/%m/%Y %H:%M:%S")
        print(message)
        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_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()

但是，当我在另一个名为test.py的文件中测试连接时，总是会遇到如下错误：

  File "/Users/rickycheng/Desktop/pair-trading-bot/venv/lib/python3.7/site-packages/websocket/_socket.py", line 143, in send
    raise WebSocketConnectionClosedException("socket is already closed.")
websocket._exceptions.WebSocketConnectionClosedException: socket is already closed.

下面是我用来测试 websocket 连接的 test.py：

from Bybit_api.Bybit_ws import Bybit_WS_test
import json

if __name__ == '__main__':
    x = Bybit_WS_test()
    x.connectWS_public()

    while (x.ws_public.sock):
        print(True)
        topic = "orderBookL2_25.BTCUSD"
        x.ws_public.send(json.dumps({"op": "subscribe",'args':[topic]}))

您可以通过以下方式查看交易所 API 文档 https://bybit-exchange.github.io/docs/linear/#t-heartbeat

【问题讨论】：

标签： python api websocket bybit python-bybit

【解决方案1】：

伙计们，我有解决方案：

一般来说，如果你和我一样遇到过类似的错误，有可能是你建立连接后没有让程序休眠。

尝试在调用ws.run_forever()后添加：time.sleep(5)

这允许在向它发送任何请求之前成功连接 websocket 连接。

【讨论】：

【解决方案2】：

你能粘贴最终的工作代码吗？

【讨论】：