TypeError：无法腌制 _thread.lock 对象答案

【问题标题】：TypeError: can't pickle _thread.lock objectsTypeError：无法腌制 _thread.lock 对象
【发布时间】：2017-10-23 23:36:26
【问题描述】：

尝试使用共享队列同时运行两个不同的函数并出现错误...如何使用共享队列同时运行两个函数？这是 Windows 7 上的 Python 3.6 版。

from multiprocessing import Process
from queue import Queue
import logging

def main():
    x = DataGenerator()
    try:
        x.run()
    except Exception as e:
        logging.exception("message")


class DataGenerator:

    def __init__(self):
        logging.basicConfig(filename='testing.log', level=logging.INFO)

    def run(self):
        logging.info("Running Generator")
        queue = Queue()
        Process(target=self.package, args=(queue,)).start()
        logging.info("Process started to generate data")
        Process(target=self.send, args=(queue,)).start()
        logging.info("Process started to send data.")

    def package(self, queue): 
        while True:
            for i in range(16):
                datagram = bytearray()
                datagram.append(i)
                queue.put(datagram)

    def send(self, queue):
        byte_array = bytearray()
        while True:
            size_of__queue = queue.qsize()
            logging.info(" queue size %s", size_of_queue)
            if size_of_queue > 7:
                for i in range(1, 8):
                    packet = queue.get()
                    byte_array.append(packet)
                logging.info("Sending datagram ")
                print(str(datagram))
                byte_array(0)

if __name__ == "__main__":
    main()

日志显示错误，我尝试以管理员身份运行控制台，但得到相同的消息...

INFO:root:Running Generator
ERROR:root:message
Traceback (most recent call last):
  File "test.py", line 8, in main
    x.run()
  File "test.py", line 20, in run
    Process(target=self.package, args=(queue,)).start()
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\context.py", line 223, in _Popen
    return _default_context.get_context().Process._Popen(process_obj)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\context.py", line 322, in _Popen
    return Popen(process_obj)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\popen_spawn_win32.py", line 65, in __init__
    reduction.dump(process_obj, to_child)
  File "C:\ProgramData\Miniconda3\lib\multiprocessing\reduction.py", line 60, in dump
    ForkingPickler(file, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

【问题讨论】：

queue.Queue 用于线程间通信。 multiprocessing.Queue 用于在进程之间发送东西。
@user2357112 我对multiprocessing.Queue 进行了更改，这解决了问题。谢谢。

标签： python-3.x

【解决方案1】：

我在 Python 3.6.3 中遇到了与 Pool() 相同的问题。

收到错误：TypeError: can't pickle _thread.RLock objects

假设我们想将一些数字num_to_add 并行添加到某个列表num_list 的每个元素中。代码示意图如下：

class DataGenerator:
    def __init__(self, num_list, num_to_add)
        self.num_list = num_list # e.g. [4,2,5,7]
        self.num_to_add = num_to_add # e.g. 1 

        self.run()

    def run(self):
        new_num_list = Manager().list()

        pool = Pool(processes=50)
        results = [pool.apply_async(run_parallel, (num, new_num_list)) 
                      for num in num_list]
        roots = [r.get() for r in results]
        pool.close()
        pool.terminate()
        pool.join()

    def run_parallel(self, num, shared_new_num_list):
        new_num = num + self.num_to_add # uses class parameter
        shared_new_num_list.append(new_num)

这里的问题是函数run_parallel() 中的self 不能被腌制，因为它是一个类实例。将这个并行化函数run_parallel() 移出课堂会有所帮助。但这不是最好的解决方案，因为此函数可能需要使用 self.num_to_add 之类的类参数，然后您必须将其作为参数传递。

解决方案：

def run_parallel(num, shared_new_num_list, to_add): # to_add is passed as an argument
    new_num = num + to_add
    shared_new_num_list.append(new_num)

class DataGenerator:
    def __init__(self, num_list, num_to_add)
        self.num_list = num_list # e.g. [4,2,5,7]
        self.num_to_add = num_to_add # e.g. 1

        self.run()

    def run(self):
        new_num_list = Manager().list()

        pool = Pool(processes=50)
        results = [pool.apply_async(run_parallel, (num, new_num_list, self.num_to_add)) # num_to_add is passed as an argument
                      for num in num_list]
        roots = [r.get() for r in results]
        pool.close()
        pool.terminate()
        pool.join()

上面的其他建议对我没有帮助。

【讨论】：

要在类函数之外访问 self 只需创建一个 fakeSelf。见最后的例子。
我在类中启动进程时遇到了同样的错误。只是使用“multiprocessing.Process”。根据您的建议，我将功能移到了课堂之外，它对我有用！

【解决方案2】：

您需要将from queue import Queue 更改为from multiprocessing import Queue。

根本原因是前者Queue是为线程模块Queue设计的，而后者是为multiprocessing.Process模块设计的。

详情可以看源码或者联系我！

【讨论】：

【解决方案3】：

multiprocessing.Pool - PicklingError: Can't pickle <type 'thread.lock'>: attribute lookup thread.lock failed

将队列移动到 self 而不是作为函数 package 和 send 的参数

【讨论】：

移动队列是什么意思？
您不能将其作为参数传递，将其设置为类的属性

【解决方案4】：

补充 Marina 回答这里的东西可以访问整个班级。它还可以根据我今天的需要愚弄 Pool.map。

fakeSelf = None

def run_parallel(num, shared_new_num_list, to_add): # to_add is passed as an argument
    new_num = num + fakeSelf.num_to_add
    shared_new_num_list.append(new_num)

class DataGenerator:
    def __init__(self, num_list, num_to_add)
        globals()['fakeSelf'] = self
        self.num_list = num_list # e.g. [4,2,5,7]
        self.num_to_add = num_to_add # e.g. 1

        self.run()

    def run(self):
        new_num_list = Manager().list()

【讨论】：

【解决方案5】：

Roque 在此线程末尾发布的 fakeSelf 方法仅在您拥有该对象的一个实例时才有效。否则 fakeSelf 将被重新分配。

【讨论】：