带有管理器和异步方法的 multiprocessing.pool答案

【问题标题】：multiprocessing.pool with manager and async methods带有管理器和异步方法的 multiprocessing.pool
【发布时间】：2020-05-11 19:08:35
【问题描述】：

我正在尝试使用 Manager() 在进程之间共享字典并尝试了以下代码：

from multiprocessing import Manager, Pool

def f(d):
    d['x'] += 2

if __name__ == '__main__':
    manager = Manager()
    d = manager.dict()
    d['x'] = 2
    p= Pool(4)

    for _ in range(2000):
        p.map_async(f, (d,))  #apply_async, map

    p.close()
    p.join()

    print (d)  # expects this result --> {'x': 4002}

使用 map_async 和 apply_async，打印的结果总是不同的（例如 {'x': 3838}, {'x': 3770}）。但是，使用 map 会给出预期的结果。另外，我尝试过使用 Process 而不是 Pool，结果也不同。

有什么见解吗？非阻塞部分和竞态条件不是由经理处理的吗？

【问题讨论】：

标签： python python-3.x multiprocessing multiprocessing-manager

【解决方案1】：

当您调用map（而不是map_async）时，它将阻塞，直到处理器完成您传递的所有请求，在您的情况下，这只是对函数f 的一次调用。因此，即使您的池大小为 4，您实际上也是一次处理 2000 个进程。要真正并行执行，您应该执行单个 p.map(f, [d]*2000) 而不是循环。

但是当您调用map_async 时，您不会阻塞并返回一个结果对象。对结果对象的get 调用将阻塞，直到进程完成并返回函数调用的结果。所以现在您一次最多可以运行 4 个进程。但是字典的更新不会跨处理器序列化。我修改了代码以通过使用多处理锁来强制序列化d[x] += 2。您会看到结果现在是 4002。

from multiprocessing import Manager, Pool, Lock


def f(d):
    lock.acquire()
    d['x'] += 2
    lock.release()

def init(l):
    global lock
    lock = l

if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()
        d['x'] = 2
        lock = Lock()
        p = Pool(4, initializer=init, initargs=(lock,)) # Create the multiprocessing lock that is sharable by all the processes

        results = [] # if the function returnd a result we wanted
        for _ in range(2000):
            results.append(p.map_async(f, (d,)))  #apply_async, map
        """
        for i in range(2000): # if the function returned a result we wanted
            results[i].get() # wait for everything to finish
        """
        p.close()
        p.join()
        print(d)

【讨论】：