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

Question

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

我一直在使用来自Filling a queue and managing multiprocessing in python 的信息来指导我跨多个进程使用队列，并使用Using a global variable with a thread 来指导我跨线程使用全局变量。我已经让软件工作了，但是当我在运行进程后检查列表/队列/字典/地图长度时，它总是返回零

我写了一个简单的例子来说明我的意思： 您必须将脚本作为文件运行，map 的 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 函数时，为什么没有任何反应？

Answer 1

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

文档中有一个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())

输出：

Hello world
0
1
2
Hello world
Hello world

Answer 2

您不能使用全局变量进行多处理。

传递给函数多处理队列。

from multiprocessing import Queue
queue= Queue() 

def worker(q):
    q.put(something)

此外，您可能会体验到代码没问题，但由于池创建单独的进程，甚至错误都是分开的，因此您看不到代码不仅无法正常工作，而且会引发错误。

您的输出是 '' 的原因是因为您的 q/global_q 没有附加任何内容。如果它被附加，那么只有一些变量，可能称为 global_q，但它与你的主线程中的 global_q 完全不同

尝试在要进行多进程处理的函数中打印（'Hello world'），您会自己看到，实际上根本没有打印任何内容。该进程只是在您的主线程之外，访问该进程的唯一方法是通过多处理队列。您可以通过 queue.put('something') 和 something = queue.get() 访问队列

试着理解这段代码，你会做得很好：

import multiprocessing as mp

shared_queue = mp.Queue() # This will be shared among all procesess, but you need to pass the queue as an argument in the process. You CANNOT use it as global variable. Understand that the functions kind of run in total different processes and nothing can really access them... Except multiprocessing.Queue - that can be shared across all processes.


def channel(que,channel_num):
    que.put(channel_num)

if __name__ == '__main__':
    processes = [mp.Process(target=channel, args=(shared_queue, channel_num)) for channel_num in range(8)]

    for p in processes:
        p.start()


    for p in processes: # wait for all results to close the pool
        p.join()

    for i in range(8): # Get data from Queue. (you can get data out of it at any time actually)
        print(shared_queue.get())