不能腌制 <class '__main__.JobQueueManager'>

Question

我在这个code 中遇到了一个可腌制问题（也附在下面）。我已经阅读了相关的帖子[1][2]，但我找不到有用的对应。能否请您对此错误的解释或解决方案？

下面是返回错误的部分代码：

pickle.PicklingError: Can't pickle <class '__main__.JobQueueManager'>: it's not found as __main__.JobQueueManager

谢谢！

def make_server_manager(port, authkey):

    job_q = Queue.Queue()
    result_q = Queue.Queue()

    class JobQueueManager(SyncManager):
        pass

    JobQueueManager.register('get_job_q', callable=lambda: job_q)
    JobQueueManager.register('get_result_q', callable=lambda: result_q)

    manager = JobQueueManager(address=('', port), authkey=authkey)
    manager.start()
    print 'Server started at port %s' % port
    return manager

^{PS：Python 2.7.7，Win 7}

Answer 1

试试：

class JobQueueManager(SyncManager):
        pass

def make_server_manager(port, authkey):

    job_q = Queue.Queue()
    result_q = Queue.Queue()

    JobQueueManager.register('get_job_q', callable=lambda: job_q)
    JobQueueManager.register('get_result_q', callable=lambda: result_q)

    manager = JobQueueManager(address=('', port), authkey=authkey)
    manager.start()
    print 'Server started at port %s' % port
    return manager

将类的定义移动到pickle可以找到的位置应该允许酸洗。 Pickle 将在 __main__ 模块中查找该类，但使用您的代码，它无法找到它，因为它在函数内部。 但是，正如 cmets 中所指出的，管理器不需要被腌制，因此必须将另一个对象拖入，例如在其全局变量中包含管理器的函数。

Answer 2

据我所知，要使此模式在 Windows 上工作，您需要创建一个可提取的 queue.Queue。您可以通过创建一个定义__setstate__ and __getstate__ 的Queue 子类来做到这一点，并让它只腌制我们实际需要在进程之间发送的状态片段，而将其他东西（不可腌制的内部锁）排除在外。

我们需要做的其他更改是将自定义Manager 类定义移至顶层，并且不使用lambda 函数作为callable 的参数。相反，我们使用partial 和顶级函数，因为它可以腌制。这是最终代码：

import sys
from multiprocessing.managers import SyncManager
from functools import partial
import multiprocessing
from Queue import Queue as _Queue

class Queue(_Queue):
    """ A picklable queue. """   
    def __getstate__(self):
        # Only pickle the state we care about
        return (self.maxsize, self.queue, self.unfinished_tasks)

    def __setstate__(self, state):
        # Re-initialize the object, then overwrite the default state with
        # our pickled state.
        Queue.__init__(self)
        self.maxsize = state[0]
        self.queue = state[1]
        self.unfinished_tasks = state[2]


def get_q(q):
    return q

class JobQueueManager(SyncManager):
    pass


def make_server_manager(port, authkey):
    job_q = Queue()
    result_q = Queue()

    job_q.put("hey")
    JobQueueManager.register('get_job_q', callable=partial(get_q, job_q))
    JobQueueManager.register('get_result_q', callable=partial(get_q, result_q))

    manager = JobQueueManager(address=('', port), authkey=authkey)
    #manager.start()
    print('Server started at port %s' % port)
    return manager

def make_client_manager(port, authkey):
    JobQueueManager.register('get_job_q')
    JobQueueManager.register('get_result_q')
    manager = JobQueueManager(address=('localhost', port), authkey=authkey)
    manager.connect()
    queue = manager.get_job_q()
    print("got queue {}".format(queue))
    print(queue.get_nowait())

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--client":
        make_client_manager(50000, 'abcdefg')
    else:
        manager = make_server_manager(50000, "abcdefg")
        server = manager.get_server()
        server.serve_forever()

Answer 3

你需要有Queue.Queuepickleable，以及你的lambda函数和你的JobQueueManager。

要做到这一点，我认为你可以很懒惰，你需要做的就是获取dill包和import dill。

我没有在 Windows 上进行测试，但它应该如下所示。 dill 在这里可用：https://github.com/uqfoundation。

>>> import dill
>>> import Queue
>>> from multiprocessing.managers import SyncManager
>>> 
>>> def make_server_manager(port, authkey):
...   job_q = Queue.Queue()
...   result_q = Queue.Queue()
...   class JobQueueManager(SyncManager):
...     pass
...   JobQueueManager.register('get_job_q', callable=lambda: job_q)
...   JobQueueManager.register('get_result_q', callable=lambda: result_q)
...   manager = JobQueueManager(address=('',port), authkey=authkey)
...   manager.start()
...   print "server started at port %s" % port
...   return manager
... 
>>> sm = make_server_manager(12345, 'foofoo')
server started at port 12345

Answer 4

multiprocessing 库为您提供了一个开箱即用的解决方案 - multiprocessing.Queue，它应该可以在任何地方自动选择，即使在 Windows 上也是如此（并且可以追溯到 2.7）。

尝试使Queue.Queue 可腌制对我来说似乎是个坏主意。您不会获得一个可以从两个不同进程中使用的队列——您将在另一个进程中获得该队列的独立副本。

如果您想在另一个进程中拥有队列当前状态的副本，那么将队列中的所有元素提取为免费腌制的普通旧列表（如果所有元素是可腌制的），将列表发送过来，然后在另一侧重新构建一个新的Queue.Queue。

另外，正如我想你现在已经发现的那样，你不能腌制本地 lambdas - 那怎么办？相反，为该命名空间创建一个全局函数，然后将该全局函数与所需数据一起发送过来。