Python 多处理和管理器

Question

我正在使用 Pythonmultiprocessing创建并行应用程序。进程需要共享一些数据，为此我使用Manager. 但是，我有一些进程需要调用的常用函数以及需要访问对象存储的数据的函数Manager。我的问题是我是否可以避免需要将Manager实例作为参数传递给这些通用函数，而是像全局函数一样使用它。换句话说，请考虑以下代码：

import multiprocessing as mp

manager = mp.Manager()
global_dict = manager.dict(a=[0])

def add():
    global_dict['a'] += [global_dict['a'][-1]+1]

def foo_parallel(var):
    add()
    print var

num_processes = 5
p = []
for i in range(num_processes):
    p.append(mp.Process(target=foo_parallel,args=(global_dict,)))

[pi.start() for pi in p]
[pi.join() for pi in p]

这运行良好并返回p=[0,1,2,3,4,5]到我的机器上。然而，这是“好形式”吗？这是一个好方法吗，就像定义add(var)和调用add(var)一样好？

Answer 1

您的代码示例似乎比形式有更大的问题。只有运气好才能得到你想要的输出。重复执行会产生不同的结果。那是因为这+=不是原子操作。多个进程可以在其中任何一个进程更新之前相继读取相同的旧值，并且它们将写回相同的值。为了防止这种行为，你必须Manager.Lock额外使用。

关于你最初关于“良好形式”的问题。

在我看来，让子进程的主函数显式foo_parallel传递global_dict到通用函数中会更干净add(var)。这将是依赖注入的一种形式，并且有一些优点。在您的示例中，并非详尽无遗：

• 允许隔离测试

• 提高代码的可重用性

• 更容易调试（检测托管对象的不可访问性不应延迟到add调用（快速失败）

• 更少的样板代码（例如，多个函数需要的资源上的 try-excepts 块）

作为旁注。仅因为其副作用而使用列表推导式被视为“代码味道”。如果不需要列表作为结果，只需使用 for 循环即可。

代码：

import os
from multiprocessing import Process, Manager


def add(l):
    l += [l[-1] + 1]
    return l


def foo_parallel(global_dict, lock):
    with lock:
        l = global_dict['a']
        global_dict['a'] = add(l)
        print(os.getpid(), global_dict)


if __name__ == '__main__':

    N_WORKERS = 5

    with Manager() as manager:

        lock = manager.Lock()
        global_dict = manager.dict(a=[0])

        pool = [Process(target=foo_parallel, args=(global_dict, lock))
                for _ in range(N_WORKERS)]

        for p in pool:
            p.start()

        for p in pool:
            p.join()

        print('result', global_dict)