强制多处理池迭代参数

Question

我正在使用多处理池一遍又一遍地为多个参数运行一个函数。我使用一个由另一个线程填充的作业列表和一个job_handler 函数来处理每个作业。我的问题是，当列表变空时，池将结束函数。我想让游泳池保持活力并等到列表填满。实际上，有两种情况可以解决这个问题。

1.使用一个池，但列表为空后会结束：

from multiprocessing import Pool
from threading import Thread
from time import sleep


def job_handler(i):
    print("Doing job:", i)
    sleep(0.5)

def job_adder():
    i = 0
    while True:
        jobs.append(i)
        i += 1
        sleep(0.1)


if __name__ == "__main__":
    pool = Pool(4)
    jobs = []
    thr = Thread(target=job_adder)
    thr.start()
    # wait for job_adder to add to list
    sleep(1)
    pool.map_async(job_handler, jobs)
    while True:
        pass

2.多个map_async:

from multiprocessing import Pool
from threading import Thread
from time import sleep


def job_handler(i):
    print("Doing job:", i)
    sleep(0.5)

def job_adder():
    i = 0
    while True:
        jobs.append(i)
        i += 1
        sleep(0.1)


if __name__ == "__main__":
    pool = Pool(4)
    jobs = []
    thr = Thread(target=job_adder)
    thr.start()
    while True:
        for job in jobs:
            pool1 = pool.map_async(job_handler, (job,))
            jobs.remove(job)

两者有什么区别？我认为第一个选项会更好，因为地图本身会处理迭代。我的目标是获得更好的性能来分别处理每项工作。

Answer 1

在许多情况下都需要“放慢”Pool。这种情况比some简单：

q=queue.Queue()
m=pool.imap(iter(q.get,None))

你也可以使用imap_unordered； None 是终止Pool 的哨兵。 Pool 必须使用线程来收集任务（因为这些函数“比map() 更懒惰”），并且它会根据需要阻塞q。