Python多处理队列非常慢

Question

我正在使用多处理队列来处理我的记录。

queue = multiprocessing.Queue()

def produce(i, item):
    data = process(i, item)
    queue.put(data)

def process(item):
    data = do_processing(item)
    return data

if __name__ == '__main__':
    records = load_records()

    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        print('produce items')
        for i, item in ennumerate(records.items()):
            executor.submit(produce, i, item)

    print('queue size:{}'.format(queue.qsize()))
    while not queue.empty():
        save(queue.get())

队列最初工作得很快。但随着队列的增长，变得非常缓慢。

根据其他 SO 答案，我也尝试使用 multiprocessing.Manager().Queue()，但没有成功。

欢迎指点。

Answer 1

在您的示例中，在所有作业都提交之前，您永远不会从队列中拉出数据，这可能会花费大量时间，从而导致相当多的缓冲区被填满。

Python 的 multiprocessing.Pool.imap 使用仅保存迭代器的线程向工作人员发送输入（阅读迭代器/生成器的效率），而另一个线程在完成时收集输出，并将它们呈现给输出迭代器因此它们可以在创建时被使用，因此您永远不必缓冲大量项目（只要您读取它们的速度比生成它们的速度快）。

from multiprocessing import Pool
from time import sleep

def produce(args):
    i, item = args #unpack tuple returned by enumerate
    data = process(i, item)
    return data

def process(i, item): #added "i" missing from original post
    data = do_processing(i, item) #do something with "i"?
    return data

if __name__ == '__main__':
    records = load_records()

    with Pool() as pool:
        print('produce items')
        for result in pool.imap(produce, enumerate(records.items())):
            save(result)