在类方法中用multiprocessing结合WebSocket时出现SSLSocket无法序列化错误的原因

我在创建一个建立WebSocket连接的Python对象时，希望在接收到消息后用multiprocessing在子进程中运行类方法，但遇到了cannot pickle 'SSLSocket' object错误，复现代码如下：

import websocket
import json
from multiprocessing import Process


def on_close(ws, close_status_code, close_msg):
    print('closed')


def on_error(ws, error):
   print(error)


class ObClient():
    def __init__(self):
        self.ws = None

    def on_open(self, ws):
        subscribe_message = {"method": "SUBSCRIBE",
                             "params": 'btcusdt@depth5@1000ms',
                             "id": 1}
        ws.send(json.dumps(subscribe_message))

    def func(self):
        print('hello')


    def on_message(self, ws, message):
        process = Process(target=self.func)
        process.start()
        process.join()

    def connect(self):
        self.ws = websocket.WebSocketApp('wss://stream.binance.com:443/stream', on_close=on_close, on_error=on_error,
                                         on_open=self.on_open, on_message=self.on_message)
        self.ws.run_forever()

if __name__ == '__main__':
    client = ObClient()
    client.connect()

错误原因 #

• Python的multiprocessing创建子进程时，默认会通过pickle序列化父进程中的对象并传递给子进程。
• 你的ObClient实例持有self.ws（WebSocketApp对象），而WebSocketApp内部包含了SSLSocket连接。SSLSocket是操作系统级别的网络资源，属于不可序列化的对象，无法被pickle处理。
• 当你执行Process(target=self.func)时，self（整个ObClient实例）会被作为参数隐式传递给子进程，这就需要序列化整个实例，包括其中的self.ws，因此触发了无法pickle SSLSocket的错误。

解决方法 #

方法1：改用线程替代进程 #

如果你的任务不需要利用多CPU核心，仅需要异步执行，用threading.Thread代替multiprocessing.Process更合适。线程共享父进程内存空间，无需序列化整个类实例：

import websocket
import json
from threading import Thread  # 替换成Thread

# 保留原on_close、on_error函数不变

class ObClient():
    def __init__(self):
        self.ws = None

    def on_open(self, ws):
        subscribe_message = {"method": "SUBSCRIBE",
                             "params": 'btcusdt@depth5@1000ms',
                             "id": 1}
        ws.send(json.dumps(subscribe_message))

    def func(self):
        print('hello')

    def on_message(self, ws, message):
        thread = Thread(target=self.func)  # 改用Thread
        thread.start()
        thread.join()

    def connect(self):
        self.ws = websocket.WebSocketApp('wss://stream.binance.com:443/stream', on_close=on_close, on_error=on_error,
                                         on_open=self.on_open, on_message=self.on_message)
        self.ws.run_forever()

if __name__ == '__main__':
    client = ObClient()
    client.connect()

方法2：将方法改为独立函数，避免传递类实例 #

把func从类中剥离，作为独立函数，这样创建进程时不需要传递整个ObClient实例：

import websocket
import json
from multiprocessing import Process

def on_close(ws, close_status_code, close_msg):
    print('closed')

def on_error(ws, error):
   print(error)

# 独立函数
def func():
    print('hello')

class ObClient():
    def __init__(self):
        self.ws = None

    def on_open(self, ws):
        subscribe_message = {"method": "SUBSCRIBE",
                             "params": 'btcusdt@depth5@1000ms',
                             "id": 1}
        ws.send(json.dumps(subscribe_message))

    def on_message(self, ws, message):
        process = Process(target=func)  # 直接调用独立函数
        process.start()
        process.join()

    def connect(self):
        self.ws = websocket.WebSocketApp('wss://stream.binance.com:443/stream', on_close=on_close, on_error=on_error,
                                         on_open=self.on_open, on_message=self.on_message)
        self.ws.run_forever()

if __name__ == '__main__':
    client = ObClient()
    client.connect()

方法3：使用multiprocessing的其他序列化方式（不推荐） #

可以尝试使用multiprocessing.set_start_method('spawn')或者改用dill等替代pickle的库，但这种方式可能引入额外复杂度，且对于SSLSocket这类资源仍可能存在问题，不建议优先使用。

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