使用multiprocessing控制Python中的进程数量

Question

我想控制在使用多进程包时同时运行的进程数量。

比如说，我只想让三个进程同时活跃。我知道的唯一方法是：

import multiprocessing
import Queue
def worker(arg):
    ## Do stuff
    return returnvalue

argument = list(1,2,3,4,5,6)
aliveprocesses = 0
jobs = Queue.Queue()
for arg in argument:
    while jobs.qsize() > 2:
        jobs.get().join()
    p = multiprocessing.Process(target=worker,args=(arg,))
    jobs.put(p)
    p.start()

基本上，我只知道如何一次监控一个进程，使用的是Process.join()这个函数。我会监控最早启动的进程，等它完成后再创建一个新进程。对于我的程序来说，最早的进程应该会比其他进程先完成，平均来说是这样的。但谁知道呢？也许其他进程先完成了，而我根本没办法知道。

我能想到的唯一替代方法是这样的：

import multiprocessing
import time
def worker(arg):
    ## Do stuff
    return returnvalue

argument = list(1,2,3,4,5,6)
aliveprocesses = 0
jobs = set()
for arg in argument:
    while aliveprocesses > 2:
        for j in jobs:
            if not j.is_alive():
                aliveprocesses -= 1
                break
            time.sleep(1)
    p = multiprocessing.Process(target=worker,args=(arg,))
    jobs.put(p)
    p.start()
    aliveprocesses += 1

在上面的函数中，你会检查所有的进程是否还在运行。如果它们都还在运行，你就稍微等一会儿，然后再检查，直到有一个进程结束，这时你就可以启动一个新进程。这里的问题是，按照我的理解，time.sleep()这个函数并不是等待进程结束的高效方法。

理想情况下，我希望有一个叫“superjoin()”的函数，类似于Process.join()，但它可以处理一组进程对象，当其中一个进程完成时，superjoin()也就返回了。而且superjoin()本身不使用time.sleep()函数，也就是说，它不是在“推卸责任”。

Answer 1

看起来你只有一个（并行）任务，所以与其单独管理每个进程，不如使用更高级的 multiprocessing.Pool，这样可以更轻松地管理进程的数量。

你不能对一个池进行 join 操作，但你可以使用一些阻塞调用（比如 Pool.map），这些调用可以完成这类任务。

如果你需要更细致的控制，可以考虑修改 Pool的源代码。