请教下 Python 多进程类方法 pickle 的问题

我用 multiprocessing.Pool 来对类方法进行多进程。然后以下报错: cPickle.PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup builtin.instancemethod failed

之后查了一下，说 multiprocessing 默认只支持不在类里面的函数或者类里面的 @staticmethod，于是用

import copy_reg
import types
def _pickle_method(m):
    if m.im_self is None:
        return getattr, (m.im_class, m.im_func.func_name)
    else:
        return getattr, (m.im_self, m.im_func.func_name)


copy_reg.pickle(types.MethodType, _pickle_method)

这段代码，解决了上面的错误，但是又以下报错: TypeError: can't pickle thread.lock objects

然后又查了貌似是类中有些变量不能序列化导致的，然后想请教下这种情况怎么解决啊。我看了些资料说是可以重写__getstate__和__setstate__魔法方法，但也不知道其参数是怎么传的。也有说将类设成全局变量，然后 pool.apply 的时候就不需要把类传进去，但是这样发现，类会初始化多次，请问要怎么解决呢？

Supplement 1 · Nov 20, 2019

补充下，我的那个类方法里面大致如下:
```
def received_message(self, socket_url):
----ws = create_connection(socket_url) # 有建了一个 websocket 连接
----.....
----rabbitConn = rabbit.rabbitConnect()
----rabbitConn.connect(type=self.rmq_exchange_type) # 有连一个消息队列
----.....
----self.logger.info("开始获取推送数据..") # 有写一个日志
----.....
----self.db..... # 有执行些数据库操作
```
请问我是需要取消一些类成员，而是将其作为变量传进去吗。

Supplement 2 · Nov 20, 2019

再补充一下，其实本身这代码在单进程里是能跑的。后来说改成连多个 websocket，然后我就想到多进程去连。因为那些连 socket，获取数据，分析处理数据都在 received_message 方法里，然后就弄成多进程跑上面的那个 received_message 的类方法，然后就发现上面的问题了，或许我的思路有点问题？

17 replies • 2019-11-22 14:00:55 +08:00

wuwukai007

Nov 20, 2019

pip install dill
dill.dumps(obj)

xiaolinjia

Nov 20, 2019

@wuwukai007 这个 obj 指的是？

wuwukai007

Nov 20, 2019

你要序列化的对象啊，可以是函数，类，内嵌函数，用的时候在 dill.loads(obj)

wuwukai007

Nov 20, 2019

a = dill.dumps(obj)
把 a 当作参数放到进程的 args 里面

xiaolinjia

Nov 20, 2019

@wuwukai007 我进程的 args 里面的参数是可以 pickle 的，可能是进程里的类方法里有些类成员不能 pickle。

wuwukai007

Nov 20, 2019

你开多进程，把这个类序列化，传进去当参数，类里面的方法会报错？

xiaolinjia

Nov 20, 2019

@wuwukai007 我补充了一下问题，我调用的时候是在 main 方法中:
client = Client()
from multiprocessing import Pool
pool = Pool(processes=2)
for url in ['ws://192.168.0.18:9988/websocket/37', 'ws://192.168.0.18:9988/websocket/44']:
pool.apply_async(func=client.received_message, args=(url,))
pool.close()
pool.join()

xiaolinjia

Nov 20, 2019

@xiaolinjia 那个空格被吃了，
client = Client()
from multiprocessing import Pool
pool = Pool(processes=2)
for url in ['ws://192.168.0.18:9988/websocket/37', 'ws://192.168.0.18:9988/websocket/44']:
----pool.apply_async(func=client.received_message, args=(url,))
pool.close()
pool.join()

hustlibraco

Nov 20, 2019

一般不建议这么搞，进程之间共享的数据最好只是整型、字符串，每个进程单独创建一个 client 比较好。强行修改序列化对象一个是不安全，一个是复杂不可重用。

xiaolinjia

Nov 20, 2019

@hustlibraco 其实我也有想过这个，不过就是单独创建一个实例的话，会多次初始化，但是我又想只初始化一次。

aaronhua

Nov 20, 2019

看着自己堆的一坨坨 shi 山，不禁感慨，有时候代码简单点，效率低点也没有关系。

wuwukai007

Nov 20, 2019

socket 对象好像不能被序列化把

xiaolinjia

Nov 20, 2019

@wuwukai007 socket 这里，我刚才单独测试的时候应该是没问题，因为他是每个进程内，才创建的。应该不会参与序列化这块。。

wuwukai007

Nov 20, 2019

copy_reg.pickle(types.MethodType, _pickle_method)
你这一步报错
把这一步换成
dill.dumps(_pickle_method)
试一下，

xiaolinjia

Nov 21, 2019

@wuwukai007 这个连第一步都失败了，现在总结得出，应该是有个数据库的持续化链接的问题，就是 self.db 。其他的 websocket，rabbitmq，因为都是在进程内创建的都没问题，应该就是剩下这个 oracle 的数据库链接不能 pickle 了

wuwukai007

Nov 21, 2019

oracle 数据库连接不能序列化，用连接池 pip install DBUtils

hustlibraco

Nov 22, 2019

@xiaolinjia 不知道你有没有测试过，序列化对象的性能开销和对象重新创建的性能开销，谁更大