如何在类中实现WebSocket数据获取并支持外部访问实例变量

问题根因 #

• WebSocket连接无法建立：WebSocketApp初始化时未传入已定义的on_open回调，连接建立后不会自动发送订阅请求，导致无法正常和服务端交互，看起来像连接失败
• 回调无法访问实例变量：on_close、on_error等类实例方法未声明第一个参数为实例自身self，同时回调参数顺序不符合websocket库要求，导致方法内无法正常访问实例属性

修正后的完整代码 #

import websocket
import json
import threading # 用于异步运行WS连接，避免阻塞主线程

class WS:
    def __init__(self):
        self.socket = 'wss://ftx.com/ws/'
        self.close = None # 提前初始化属性，避免未收到消息时调用取值接口报错

    def stream(self):
        self.ws = websocket.WebSocketApp(
            self.socket,
            on_open=self.on_open, # 补传on_open回调
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.ws.run_forever()

    # 新增异步启动方法，避免run_forever阻塞主线程
    def start_stream(self):
        threading.Thread(target=self.stream, daemon=True).start()

    def on_open(self, ws): # 补全回调要求的参数
        print('connected')
        data = {'op': 'subscribe', 'channel': 'ticker', 'market': 'ETH-PERP'}
        ws.send(json.dumps(data))

    def on_close(self, ws, close_status_code, close_msg): # 补全self和回调要求的参数
        print('disconnected')

    def on_message(self, ws, message): # 补全参数
        json_msg = json.loads(message)
        # 过滤非行情推送的消息，避免取值报错
        if 'data' in json_msg and 'last' in json_msg['data']:
            self.close = json_msg['data']['last']

    def get_data_out(self):
        return self.close
   
    def on_error(self, ws, error): # 补全self和回调要求的参数
        print(error)

使用示例 #

if __name__ == "__main__":
    ws_client = WS()
    ws_client.start_stream()
    # 等待几秒保证连接建立并收到第一笔数据
    import time
    time.sleep(3)
    # 类外部访问实例变量获取最新成交价
    print("ETH-PERP最新成交价：", ws_client.get_data_out())

补充说明 #

• 所有类的实例方法第一个参数必须为self，后续按websocket库的回调规则传入ws实例、消息内容、错误信息等参数
• 用子线程运行ws连接逻辑，避免run_forever阻塞主线程，导致无法在外部调用接口获取数据
• 对消息内容做了合法性判断，避免收到服务端的非行情类推送时触发KeyError
• 在__init__中提前初始化self.close，避免未收到数据时调用取值方法抛出属性不存在的异常

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