python池apply_async和map_async不会阻塞完整队列

Question

我对python很新.我正在使用多处理模块读取stdin上的文本行,以某种方式转换它们并将它们写入数据库.这是我的代码片段:

batch = []
pool = multiprocessing.Pool(20)
i = 0
for i, content in enumerate(sys.stdin):
    batch.append(content)
    if len(batch) >= 10000:
        pool.apply_async(insert, args=(batch,i+1))
        batch = []
pool.apply_async(insert, args=(batch,i))
pool.close()
pool.join()

现在一切正常,直到我处理我输入我的python程序的巨大输入文件(数亿行).在某些时候,当我的数据库变慢时,我看到内存已满.

经过一番播放后,事实证明pool.apply_async以及pool.map_async永远不会阻塞,因此要处理的调用队列越来越大.

我的问题的正确方法是什么？我希望我能设置一个参数,一旦达到某个队列长度,就会阻塞pool.apply_async调用.Java中的AFAIR可以为ThreadPoolExecutor提供一个具有固定长度的BlockingQueue用于此目的.

谢谢!

Answer 1

为了防止有人在这里结束,这就是我解决问题的方法:我停止使用multiprocessing.Pool.我现在就是这样做的:

#set amount of concurrent processes that insert db data
processes = multiprocessing.cpu_count() * 2

#setup batch queue
queue = multiprocessing.Queue(processes * 2)

#start processes
for _ in range(processes): multiprocessing.Process(target=insert, args=(queue,)).start() 

#fill queue with batches    
batch=[]
for i, content in enumerate(sys.stdin):
    batch.append(content)
    if len(batch) >= 10000:
        queue.put((batch,i+1))
        batch = []
if batch:
    queue.put((batch,i+1))

#stop processes using poison-pill
for _ in range(processes): queue.put((None,None))

print "all done."

在insert方法中,每个批处理的处理都包含在一个循环中,该循环从队列中拉出,直到它收到毒丸:

while True:
    batch, end = queue.get()
    if not batch and not end: return #poison pill! complete!
    [process the batch]
print 'worker done.'

Answer 2

这些apply_async和map_async功能的目的不是阻止主要过程.为了做到这一点,Pool维持一个内部Queue,不幸的是,这个尺寸无法改变.

解决问题的方法是使用Semaphore您想要队列的大小初始化.您在进入池之前以及在工作人员完成任务之后获取并释放信号量.

这是一个使用Python 2.6或更高版本的示例.

from threading import Semaphore
from multiprocessing import Pool

def task_wrapper(f):
    """Python2 does not allow a callback for method raising exceptions,
    this wrapper ensures the code run into the worker will be exception free.

    """
    try:
        return f()
    except:
        return None

class TaskManager(object):
    def __init__(self, processes, queue_size):
        self.pool = Pool(processes=processes)
        self.workers = Semaphore(processes + queue_size)

    def new_task(self, f):
        """Start a new task, blocks if queue is full."""
        self.workers.acquire()
        self.pool.apply_async(task_wrapper, args=(f, ), callback=self.task_done))

    def task_done(self):
        """Called once task is done, releases the queue is blocked."""
        self.workers.release()

使用池实现的另一个示例concurrent.futures.

Answer 3

apply_async返回一个AsyncResult对象，您可以wait对其进行：

if len(batch) >= 10000:
    r = pool.apply_async(insert, args=(batch, i+1))
    r.wait()
    batch = []

不过，如果您想以更简洁的方式执行此操作，则应该使用 a为 10000 的multiprocessing.Queuea maxsize，并从从此类队列中获取的内容派生一个Worker类。multiprocessing.Process