python多处理池：我怎么知道池中的所有工作人员何时完成？

Posted 2023-02-16

【中文标题】python多处理池：我怎么知道池中的所有工作人员何时完成？

【英文标题】：python multiprocessing pool: how can I know when all the workers in the pool have finished?

【发布时间】：2015-07-30 18:03:35

【问题描述】：

我在 python 中运行一个多处理池，我有大约 2000 个任务，被映射到池中的 24 个工作人员。 每个任务都会根据一些数据分析和 Web 服务创建一个文件。

当池中的所有任务都完成后，我想运行一个新任务。如何判断池中的所有进程何时完成？

【参考方案1】：

您想使用the join method，它会阻止主进程线程向前移动，直到所有子进程结束：

阻塞调用线程，直到调用 join() 方法的进程终止或直到发生可选超时。

from multiprocessing import Process

def f(name):
    print 'hello', name

if __name__ == '__main__':
    processes = []
    for i in range(10):
        p = Process(target=f, args=('bob',))
        processes.append(p)

    for p in processes:
        p.start()
        p.join()

     # only get here once all processes have finished.
     print('finished!')

编辑：

使用join with pools

    pool = Pool(processes=4)  # start 4 worker processes
    result = pool.apply_async(f, (10,))  # do some work
    pool.close()
    pool.join()  # block at this line until all processes are done
    print("completed")

【讨论】：

谢谢，但我问的是池方法：在哪里让多处理自动启动进程。你怎么能在游泳池里做这个“加入”的把戏？

好的更新答案。您只需在 pool 实例上调用 join()。

请注意，您需要先调用pool.close() 或pool.terminate()，然后才能调用pool.join()，因此上面的示例实际上并不起作用。另请注意，如果您以后不再需要使用池，则使用join() 告知工作何时完成只是一个可行的选择，因为它需要关闭或终止池。

如果您希望进程并行运行，您需要先在所有进程上调用 start()，然后再调用 join。

这个接受的答案NOT并行运行，因此不是一个有效的答案。

【参考方案2】：

您可以使用ApplyResult 对象的wait() 方法（这是pool.apply_async 返回的内容）。

import multiprocessing

def create_file(i):
    open(f'i.txt', 'a').close()

if __name__ == '__main__':
    # The default for n_processes is the detected number of CPUs
    with multiprocessing.Pool() as pool:

        # Launch the first round of tasks, building a list of ApplyResult objects
        results = [pool.apply_async(create_file, (i,)) for i in range(50)]
    
        # Wait for every task to finish
        [result.wait() for result in results]

        # start your next task... the pool is still available

    # when you reach here, the pool is closed

即使您计划再次使用您的池并且不想关闭它，此方法仍然有效 - 例如，您可能希望在算法的下一次迭代中保留它。使用 with 语句或使用完后手动调用 pool.close()，否则会发生不好的事情。