跨多个进程使用双端队列对象

我试图减少读取大约 100,000 个条目的数据库的处理时间，但我需要它们以特定的方式格式化，为了尝试做到这一点，我尝试使用 pythonmultiprocessing.map函数工作完美，只是我似乎无法获得任何形式的队列引用来跨它们工作。

我一直在使用来自在 python 中填充队列并管理多处理 https://stackoverflow.com/questions/17241663/filling-a-queue-and-managing-multiprocessing-in-python指导我跨多个进程使用队列，以及在线程中使用全局变量 https://stackoverflow.com/questions/19790570/using-a-global-variable-with-a-thread指导我跨线程使用全局变量。我已经让软件可以工作了，但是当我在运行该过程后检查列表/队列/字典/映射长度时，它总是返回零

我写了一个简单的例子来说明我的意思： 您必须将脚本作为文件运行，map's initialize函数在解释器中不起作用。

from multiprocessing import Pool
from collections import deque

global_q = deque()

def my_init(q):
    global global_q
    global_q = q
    q.append("Hello world")


def map_fn(i):
    global global_q
    global_q.append(i)


if __name__ == "__main__":
    with Pool(3, my_init, (global_q,)) as pool:
        pool.map(map_fn, range(3))
    for p in range(len(global_q)):
        print(global_q.pop())

理论上，当我使用以下方法将队列对象引用从主线程传递到工作线程时pool函数，然后使用给定函数初始化该线程的全局变量，然后当我从map函数之后，该对象引用应该仍然指向原始队列对象引用（长话短说，所有东西都应该在同一个队列中结束，因为它们都指向内存中的相同位置）。

所以，我期望：

Hello World
Hello World
Hello World
1
2
3

当然，1, 2, 3的顺序是任意的，但是您将在输出中看到的是''.

为什么当我将对象引用传递给pool函数，什么也没发生？

这是一个如何通过扩展在进程之间共享某些内容的示例multiprocessing.managers.BaseManager支持类deques.

有一个定制经理人 https://docs.python.org/3/library/multiprocessing.html#customized-managers文档中有关创建它们的部分。

import collections
from multiprocessing import Pool
from multiprocessing.managers import BaseManager


class DequeManager(BaseManager):
    pass

class DequeProxy(object):
    def __init__(self, *args):
        self.deque = collections.deque(*args)
    def __len__(self):
        return self.deque.__len__()
    def appendleft(self, x):
        self.deque.appendleft(x)
    def append(self, x):
        self.deque.append(x)
    def pop(self):
        return self.deque.pop()
    def popleft(self):
        return self.deque.popleft()

# Currently only exposes a subset of deque's methods.
DequeManager.register('DequeProxy', DequeProxy,
                      exposed=['__len__', 'append', 'appendleft',
                               'pop', 'popleft'])


process_shared_deque = None  # Global only within each process.

def my_init(q):
    """ Initialize module-level global. """
    global process_shared_deque
    process_shared_deque = q
    q.append("Hello world")


def map_fn(i):
    process_shared_deque.append(i)  # deque's don't have a "put()" method.


if __name__ == "__main__":
    manager = DequeManager()
    manager.start()
    shared_deque = manager.DequeProxy()

    with Pool(3, my_init, (shared_deque,)) as pool:
        pool.map(map_fn, range(3))

    for p in range(len(shared_deque)):  # Show left-to-right contents.
        print(shared_deque.popleft())

Output:

Hello world
0
1
2
Hello world
Hello world