我想在 accepted answer 上进行扩展，所以我添加了自己的内容，其中还详细介绍了一种使队列、锁等可挑选并能够通过池发送的方法。
为什么会这样
基本上，并不是队列不能序列化，只是 multiprocessing 只有在知道有关它将被发送到的目标进程（无论是当前进程还是其他进程）的足够信息时才能够序列化这些队列，这就是为什么当您自己生成一个进程（使用Process 类）时它会起作用，但当您只是将它放入队列时（例如使用Pool 时）则不起作用。
查看multiprocessing.queues.Queue（或其他连接对象，如Condition）的源代码。你会发现在他们的__getstate__方法（队列实例被pickle时调用的方法）中，有一个函数multiprocessing.context.assert_spawning的调用。这个“断言”只有在当前线程正在产生一个过程。如果不是这种情况，multiprocessing 会引发您看到的错误并退出。
现在 multiprocessing 甚至懒得去腌制队列以防断言失败的原因是它无法访问线程创建子进程时创建的 Popen 对象（对于 Windows，您可以在 multiprocessing.popen_spawn_win32.Popen 找到它).该对象存储有关目标进程的数据，包括其 pid 和进程句柄。多处理需要此信息，因为 Queue 包含互斥体，并且要成功地 pickle 并稍后再次重建这些，多处理必须使用来自 Popen 对象的信息通过 winapi 调用 DuplicateHandle。如果没有此对象，多处理将不知道该做什么并引发错误。所以这就是我们的问题所在，但如果我们可以教 multiprocessing 一种不同的方法来从目标进程本身内部窃取重复句柄，而无需事先要求它的信息，那么它是可以解决的。
制作可挑选队列
关注班级multiprocessing.synchronize.SemLock。它是所有多处理锁的基类，所以它的对象随后出现在队列、管道等中。它当前被 pickle 的方式就像我上面描述的那样，它需要目标进程的句柄来创建一个重复的句柄。但是，我们可以为 SemLock 定义一个 __reduce__ 方法，我们将使用当前进程的句柄创建一个重复的句柄，然后从目标进程复制先前创建的句柄，该句柄现在在目标进程的上下文中有效.这相当冗长，但实际上也使用类似的方法来腌制 PipeConnection 对象，但它使用 dispatch table 来代替 __reduce__ 方法。
完成后，我们可以继承Queue 并删除对assert_spawning 的调用，因为不再需要它。这样，我们现在就可以成功地 pickle 锁、队列、管道等。下面是带有示例的代码：
import os, pickle
from multiprocessing import Pool, Lock, synchronize, get_context
import multiprocessing.queues
import _winapi
def work(q):
print("Worker: Main says", q.get())
q.put('haha')
class DupSemLockHandle(object):
"""
Picklable wrapper for a handle. Attempts to mirror how PipeConnection objects are pickled using appropriate api
"""
def __init__(self, handle, pid=None):
if pid is None:
# We just duplicate the handle in the current process and
# let the receiving process steal the handle.
pid = os.getpid()
proc = _winapi.OpenProcess(_winapi.PROCESS_DUP_HANDLE, False, pid)
try:
self._handle = _winapi.DuplicateHandle(
_winapi.GetCurrentProcess(),
handle, proc, 0, False, _winapi.DUPLICATE_SAME_ACCESS)
finally:
_winapi.CloseHandle(proc)
self._pid = pid
def detach(self):
"""
Get the handle, typically from another process
"""
# retrieve handle from process which currently owns it
if self._pid == os.getpid():
# The handle has already been duplicated for this process.
return self._handle
# We must steal the handle from the process whose pid is self._pid.
proc = _winapi.OpenProcess(_winapi.PROCESS_DUP_HANDLE, False,
self._pid)
try:
return _winapi.DuplicateHandle(
proc, self._handle, _winapi.GetCurrentProcess(),
0, False, _winapi.DUPLICATE_CLOSE_SOURCE | _winapi.DUPLICATE_SAME_ACCESS)
finally:
_winapi.CloseHandle(proc)
def reduce_lock_connection(self):
sl = self._semlock
dh = DupSemLockHandle(sl.handle)
return rebuild_lock_connection, (dh, type(self), (sl.kind, sl.maxvalue, sl.name))
def rebuild_lock_connection(dh, t, state):
handle = dh.detach() # Duplicated handle valid in current process's context
# Create a new instance without calling __init__ because we'll supply the state ourselves
lck = t.__new__(t)
lck.__setstate__((handle,)+state)
return lck
# Add our own reduce function to pickle SemLock and it's child classes
synchronize.SemLock.__reduce__ = reduce_lock_connection
class PicklableQueue(multiprocessing.queues.Queue):
"""
A picklable Queue that skips the call to context.assert_spawning because it's no longer needed
"""
def __init__(self, *args, **kwargs):
ctx = get_context()
super().__init__(*args, **kwargs, ctx=ctx)
def __getstate__(self):
return (self._ignore_epipe, self._maxsize, self._reader, self._writer,
self._rlock, self._wlock, self._sem, self._opid)
def is_locked(l):
"""
Returns whether the given lock is acquired or not.
"""
locked = l.acquire(block=False)
if locked is False:
return True
else:
l.release()
return False
if __name__ == '__main__':
# Example that shows that you can now pickle/unpickle locks and they'll still point towards the same object
l1 = Lock()
p = pickle.dumps(l1)
l2 = pickle.loads(p)
print('before acquiring, l1 locked:', is_locked(l1), 'l2 locked', is_locked(l2))
l2.acquire()
print('after acquiring l1 locked:', is_locked(l1), 'l2 locked', is_locked(l2))
# Example that shows how you can pass a queue to Pool and it will work
with Pool() as pool:
q = PicklableQueue()
q.put('laugh')
pool.map(work, (q,))
print("Main: Worker says", q.get())
输出
before acquiring, l1 locked: False l2 locked False
after acquiring l1 locked: True l2 locked True
Worker: Main says laugh
Main: Worker says haha
免责声明: 以上代码仅适用于 Windows。如果您使用的是 UNIX，那么您可以尝试使用 @Booboo's 修改后的代码（报告有效但尚未经过充分测试，完整代码链接 here）：
import os, pickle
from multiprocessing import Pool, Lock, synchronize, get_context, Process
import multiprocessing.queues
import sys
_is_windows= sys.platform == 'win32'
if _is_windows:
import _winapi
.
.
.
class DupSemLockHandle(object):
"""
Picklable wrapper for a handle. Attempts to mirror how PipeConnection objects are pickled using appropriate api
"""
def __init__(self, handle, pid=None):
if pid is None:
# We just duplicate the handle in the current process and
# let the receiving process steal the handle.
pid = os.getpid()
if _is_windows:
proc = _winapi.OpenProcess(_winapi.PROCESS_DUP_HANDLE, False, pid)
try:
self._handle = _winapi.DuplicateHandle(
_winapi.GetCurrentProcess(),
handle, proc, 0, False, _winapi.DUPLICATE_SAME_ACCESS)
finally:
_winapi.CloseHandle(proc)
else:
self._handle = handle
self._pid = pid
def detach(self):
"""
Get the handle, typically from another process
"""
# retrieve handle from process which currently owns it
if self._pid == os.getpid():
# The handle has already been duplicated for this process.
return self._handle
if not _is_windows:
return self._handle
# We must steal the handle from the process whose pid is self._pid.
proc = _winapi.OpenProcess(_winapi.PROCESS_DUP_HANDLE, False,
self._pid)
try:
return _winapi.DuplicateHandle(
proc, self._handle, _winapi.GetCurrentProcess(),
0, False, _winapi.DUPLICATE_CLOSE_SOURCE | _winapi.DUPLICATE_SAME_ACCESS)
finally:
_winapi.CloseHandle(proc)