Python 多处理 : sharing data between processes

Question

我是第一次尝试使用多处理，遇到了一些相当基本的问题。我在下面有一个玩具示例，其中两个进程正在向列表中添加数据:

def add_process(all_nums_class, numbers_to_add):
    for number in numbers_to_add:
        all_nums_class.all_nums_list.append(number)

class AllNumsClass:
    def __init__(self):
        self.all_nums_list = []

all_nums_class = AllNumsClass()

p1 = Process(target=add_process, args=(all_nums_class, [1,3,5]))
p1.start()

p2 = Process(target=add_process, args=(all_nums_class, [2,4,6]))
p2.start()

all_nums_class.all_nums_list

我希望在这些进程之间共享 all_nums_class，以便它们都可以添加到它的 all_nums_list - 所以结果应该是

[1,2,3,4,5,6]

而不是我目前得到的只是好旧

[]

谁能给点建议？我已经尝试过一些命名空间，但我还没有让它在这里工作。

我觉得我最好提一下(以防万一)我是在 Jupyter notebook 上做的。

Answer 1

您可以使用多处理 Queue or a Pipe在进程之间共享数据。队列是线程和进程安全的。使用 Pipe 时必须更加小心，因为如果两个进程(或线程)尝试同时读取或写入管道的同一端，管道中的数据可能会损坏。当然，同时使用管道不同端的进程不会有损坏的风险。

目前，您的实现产生了两个独立的进程，每个进程都有其自己的 self.all_nums_list。因此，您基本上生成了 AllNumsClass 的三个对象:一个在您的主程序中，一个在 p1 中，一个在 p2 中。由于进程是独立的并且不共享相同的内存空间，因此它们正确附加但它附加到每个进程自己的self.all_nums_list。这就是为什么当您在主程序中打印 all_nums_class.all_nums_list 时，您正在打印主进程的 self.all_nums_list 这是一个空列表。要共享数据并将进程附加到同一个列表，我建议使用队列。

使用队列和进程的示例

import multiprocessing as mp

def add_process(queue, numbers_to_add):
    for number in numbers_to_add:
        queue.put(number)

class AllNumsClass:
    def __init__(self):
        self.queue = mp.Queue()
    def get_queue(self):
        return self.queue

if __name__ == '__main__':
    
    all_nums_class = AllNumsClass()

    processes = []
    p1 = mp.Process(target=add_process, args=(all_nums_class.get_queue(), [1,3,5]))
    p2 = mp.Process(target=add_process, args=(all_nums_class.get_queue(), [2,4,6]))

    processes.append(p1)
    processes.append(p2)
    for p in processes:
        p.start()
    for p in processes:
        p.join()

    output = [] 
    while all_nums_class.get_queue().qsize() > 0:
        output.append(all_nums_class.get_queue().get())
    print(output)

此实现是异步，因为它不按顺序应用。每次运行它时，您可能会得到不同的输出。

示例输出

[1, 2, 3, 5, 4, 6]

[1, 3, 5, 2, 4, 6]

[2, 4, 6, 1, 3, 5]

[2, 1, 4, 3, 5, 6]

维护有序或无序列结果列表的更简单方法是使用 mp.Pool类(class)。具体来说，Pool.apply 和 Pool.apply_async 函数。 Pool.apply 将锁定主程序，直到所有进程完成，如果我们想为某些应用程序以特定顺序获取结果，这将非常有用。相比之下，Pool.apply_async 将一次提交所有进程，并在完成后立即检索结果。另一个区别是我们需要在 Pool.apply_async 调用之后使用 get 方法来获取已完成进程的返回值。