为什么将WebSocket封装到自定义函数后无法运行？该如何解决

直接全局实例化WebSocket时运行正常，代码如下：

ws = websocket.WebSocketApp(f"wss://fstream.binance.com/ws/{symbol}@aggTrade",on_message=on_message,on_close=on_close)
ws.run_forever()

封装到自定义函数后调用无法正常运行，封装代码如下：

def func(symbol):
    ws = websocket.WebSocketApp(f"wss://fstream.binance.com/ws/{symbol}@aggTrade",on_message=on_message,on_close=on_close)
    ws.run_forever()
func("btcusdt")

• 局部变量被垃圾回收：函数内部创建的ws实例属于局部变量，websocket-client库的底层回调逻辑没有强持有实例引用，函数执行过程中如果Python触发垃圾回收，实例被销毁就会导致连接异常中断
• 回调函数作用域不匹配：如果on_message、on_close回调函数是定义在其他作用域，或者封装后函数内部无法访问到对应的回调方法，会触发执行异常直接终止连接
• 参数校验缺失：币安U本位合约websocket接口要求交易对参数必须为全小写，若调用时传入大小写混合的交易对参数，会直接返回连接错误，属于易忽略的隐性问题

你可以通过持有实例引用、补充错误回调、增加参数校验的方式修复问题，示例代码如下：

import websocket

# 回调函数定义在全局作用域，或者保证func可以正常访问
def on_message(ws, message):
    print(message)

def on_close(ws, close_status_code, close_msg):
    print("连接关闭", close_status_code, close_msg)

def on_error(ws, error):
    print("连接报错", error)

def func(symbol):
    # 统一转小写，避免参数格式错误
    symbol = symbol.lower()
    ws = websocket.WebSocketApp(
        f"wss://fstream.binance.com/ws/{symbol}@aggTrade",
        on_message=on_message,
        on_close=on_close,
        on_error=on_error
    )
    # 将实例返回给外部持有，防止被垃圾回收
    return ws

# 外部持有实例引用
ws_instance = func("btcusdt")
ws_instance.run_forever()

如果需要在多线程场景下调用，也可以将ws实例存到全局列表、类属性等不会被轻易回收的位置即可。

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