我正在使用多处理模块来拆分一个非常大的任务。它在大多数情况下都有效，但我的设计肯定遗漏了一些明显的东西，因为这样我很难有效地判断何时处理了所有数据。
我有两个单独的任务运行；一个喂另一个。我想这是一个生产者/消费者问题。我在所有进程之间使用共享队列，生产者填满队列，消费者从队列中读取并进行处理。问题是数据量是有限的，所以在某些时候每个人都需要知道所有数据都已处理，以便系统可以正常关闭。
使用 map_async() 函数似乎很有意义，但是由于生产者正在填满队列，所以我不知道前面的所有项目，所以我必须进入一个 while 循环并使用 apply_async()并尝试检测何时一切都以某种超时完成......丑陋。
我觉得我错过了一些明显的东西。如何更好地设计？
生产者
class ProducerProcess(multiprocessing.Process):
def __init__(self, item, consumer_queue):
self.item = item
self.consumer_queue = consumer_queue
multiprocessing.Process.__init__(self)
def run(self):
for record in get_records_for_item(self.item): # this takes time
self.consumer_queue.put(record)
def start_producer_processes(producer_queue, consumer_queue, max_running):
running = []
while not producer_queue.empty():
running = [r for r in running if r.is_alive()]
if len(running) < max_running:
producer_item = producer_queue.get()
p = ProducerProcess(producer_item, consumer_queue)
p.start()
running.append(p)
time.sleep(1)
消费者
def process_consumer_chunk(queue, chunksize=10000):
for i in xrange(0, chunksize):
try:
# don't wait too long for an item
# if new records don't arrive in 10 seconds, process what you have
# and let the next process pick up more items.
record = queue.get(True, 10)
except Queue.Empty:
break
do_stuff_with_record(record)
主要的
if __name__ == "__main__":
manager = multiprocessing.Manager()
consumer_queue = manager.Queue(1024*1024)
producer_queue = manager.Queue()
producer_items = xrange(0,10)
for item in producer_items:
producer_queue.put(item)
p = multiprocessing.Process(target=start_producer_processes, args=(producer_queue, consumer_queue, 8))
p.start()
consumer_pool = multiprocessing.Pool(processes=16, maxtasksperchild=1)
这是它变得俗气的地方。我不能使用地图，因为要消费的列表同时被填满。所以我必须进入一个while循环并尝试检测超时。当生产者仍在尝试填充它时，consumer_queue 可能会变空，所以我不能只检测到一个空队列并就此退出。
timed_out = False
timeout= 1800
while 1:
try:
result = consumer_pool.apply_async(process_consumer_chunk, (consumer_queue, ), dict(chunksize=chunksize,))
if timed_out:
timed_out = False
except Queue.Empty:
if timed_out:
break
timed_out = True
time.sleep(timeout)
time.sleep(1)
consumer_queue.join()
consumer_pool.close()
consumer_pool.join()
我想也许我可以 get() 主线程中的记录并将它们传递给消费者而不是传递队列，但我认为我最终会遇到同样的问题。我仍然需要运行一个 while 循环并使用 apply_async() 提前感谢您的任何建议！