到目前为止，每当我需要使用multiprocessing通过手动创建一个“进程池”，并与所有的子进程共享的工作队列我已经这样做了。
例如：
from multiprocessing import Process, Queue
class MyClass:
def __init__(self, num_processes):
self._log = logging.getLogger()
self.process_list = []
self.work_queue = Queue()
for i in range(num_processes):
p_name = 'CPU_%02d' % (i+1)
self._log.info('Initializing process %s', p_name)
p = Process(target = do_stuff,
args = (self.work_queue, 'arg1'),
name = p_name)
这样，我可以添加的东西到队列中，这将由子过程消耗。 然后，我可以监测处理多远是通过检查Queue.qsize()
while True:
qsize = self.work_queue.qsize()
if qsize == 0:
self._log.info('Processing finished')
break
else:
self._log.info('%d simulations still need to be calculated', qsize)
现在我明白这multiprocessing.Pool可以简化很多这样的代码。
我无法找出是如何监控的“工作”量仍然留下许多工作要做。
看看下面的例子：
from multiprocessing import Pool
class MyClass:
def __init__(self, num_processes):
self.process_pool = Pool(num_processes)
# ...
result_list = []
for i in range(1000):
result = self.process_pool.apply_async(do_stuff, ('arg1',))
result_list.append(result)
# ---> here: how do I monitor the Pool's processing progress?
# ...?
有任何想法吗？
使用Manager队列。 这是一个工作进程之间共享的队列。 如果使用普通队列它会被腌制和每个工人拆封，因此复制，从而使队列不能每个工人进行更新。
然后，你有你的工人，增加东西向队列和监视队列的状态，而工人正在工作。 你需要做到这一点使用map_async ，因为这可以让你看到当整个结果是准备好了，让你打破的监视循环。
例：
import time
from multiprocessing import Pool, Manager
def play_function(args):
"""Mock function, that takes a single argument consisting
of (input, queue). Alternately, you could use another function
as a wrapper.
"""
i, q = args
time.sleep(0.1) # mock work
q.put(i)
return i
p = Pool()
m = Manager()
q = m.Queue()
inputs = range(20)
args = [(i, q) for i in inputs]
result = p.map_async(play_function, args)
# monitor loop
while True:
if result.ready():
break
else:
size = q.qsize()
print(size)
time.sleep(0.1)
outputs = result.get()
我有同样的问题，与MapResult对象有些简单的解决方案提出了（虽然使用内部MapResult数据）
pool = Pool(POOL_SIZE)
result = pool.map_async(get_stuff, todo)
while not result.ready():
remaining = result._number_left * result._chunksize
sys.stderr.write('\r\033[2KRemaining: %d' % remaining)
sys.stderr.flush()
sleep(.1)
print >> sys.stderr, '\r\033[2KRemaining: 0'
需要注意的是剩余价值并不总是准确的，因为块大小通常四舍五入根据项目进程数目。
您可以使用此circuvent pool.map_async(get_stuff, todo, chunksize=1)
我想出了以下解决方案async_call。
平凡的玩具脚本示例，但应充分应用，我认为。
基本上在一个无限循环轮询列表中的发电机和你的结果对象的准备价值得到多少你派出池任务剩余的计数。
一旦没有剩余突破和join（）和close（）方法。
根据需要补充睡眠中循环。
同样的原则解决方案上面，但没有一个队列。 如果你还跟踪您最初多少任务发送的游泳池，你可以计算完成百分比，等等。
import multiprocessing
import os
import time
from random import randrange
def worker():
print os.getpid()
#simulate work
time.sleep(randrange(5))
if __name__ == '__main__':
pool = multiprocessing.Pool(processes=8)
result_objs = []
print "Begin dispatching work"
task_count = 10
for x in range(task_count):
result_objs.append(pool.apply_async(func=worker))
print "Done dispatching work"
while True:
incomplete_count = sum(1 for x in result_objs if not x.ready())
if incomplete_count == 0:
print "All done"
break
print str(incomplete_count) + " Tasks Remaining"
print str(float(task_count - incomplete_count) / task_count * 100) + "% Complete"
time.sleep(.25)
pool.close()
pool.join()
从文档，它看起来像你对我想做的事是收集您的result S IN列表或其他序列，然后遍历结果列表检查ready建立自己的输出列表。 然后，您可以通过不处于就绪状态比较剩下的结果对象的数量的派遣就业总人数计算处理状态。 见http://docs.python.org/2/library/multiprocessing.html#multiprocessing.pool.AsyncResult
文章来源: How to get the amount of “work” left to be done by a Python multiprocessing Pool?
