python - 如何管理进程池？

Question

我正在尝试在 Windows 10 上设置多处理池。

基本上一些 cpu(在我的例子中是 12 个)应该从 Qin 读取并将结果写入 Qout。当在 Qin 中写入 'end' 时，进程应该停止。

由于某种原因，进程挂起。

我开发了一个简单的版本:

from multiprocessing import Pool, Queue, Event
import os,time


def worker( Qin, Qout, event):
    time.sleep(5)
    while True:
        item = Qin.get()
        if item == 'end':
            event.set()
        else:
            Qout.put(item)
        time.sleep(1)

def manager():
    Qin,Qout,event= Queue(), Queue(), Event()
    processes = os.cpu_count()
    pool = Pool(processes=processes)
    for _ in range(processes):
        pool.apply_async(worker,args= (Qin,Qout,event,))
    for i in range(100):
        print(i)
        Qin.put(i)

    Qin.put('end')

    pool.close()
    event.wait()
    pool.terminate()
    return Qout

Qout = manager()

Answer 1

您需要了解异步编程在 Python 中的正确工作方式。当您调用 apply_async 时，您会得到 Future 对象。 python 中的 Queue 实现依赖系统管道将数据从一个进程传输到另一个进程和一些信号量来保护此管道上的读写。

from multiprocessing import Pool, Queue, Event
import os
import time
import multiprocessing

def worker( Qin, Qout, event):
    print('worker')
    time.sleep(1)
    event.set()

def manager():
    processes = multiprocessing.cpu_count()
    m = multiprocessing.Manager()
    Qin = m.Queue()
    Qout = m.Queue()
    event = m.Event()
    pool = Pool(processes=processes)
    result = pool.apply_async(worker, (Qin, Qout, event))
    result.get()
    pool.close()
    event.wait()
    return Qout

if __name__ == '__main__':
    Qout = manager()