Многопроцессорный пул возвращает результаты как доступно

Question

Я пытаюсь реализовать многопроцессорность и изо всех сил пытаюсь получить то, что мне нужно получить.

Так что некоторый опыт, который я ранее выполнял многопроцессорным процессом с Celery, поэтому я привык к тому, что могу отправлять задания для рабочего и опроса, когда это делается, и получать результаты работы, даже если другие задания идут. Я пытаюсь связать это с многопроцессорной обработкой. Вот то, что я до сих пор, выкопали с различных сайтов, которые я нашел ...

import urllib2 
import time 
from multiprocessing.dummy import Pool as ThreadPool 
import random 

def openurl(url): 
    time.sleep(random.randrange(1,10)) 
    print url 
    return urllib2.urlopen(url) 


urls = [ 
    'http://www.python.org', 
    'http://www.python.org/about/', 
    'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 
    'http://www.python.org/doc/', 
    'http://www.python.org/download/', 
    'http://www.python.org/getit/', 
    'http://www.python.org/community/', 
    'https://wiki.python.org/moin/', 
    'http://planet.python.org/', 
    'https://wiki.python.org/moin/LocalUserGroups', 
    'http://www.python.org/psf/', 
    'http://docs.python.org/devguide/', 
    'http://www.python.org/community/awards/' 
    # etc.. 
    ] 

pool = ThreadPool(20) 

results = pool.map(openurl, urls) 

pool.close() 
pool.join() 
print 'now what' 

Так что я сбрасывая функцию OpenUrl на моих URLs, но если я нарушу на «печать" теперь, что» , он не прерывается там, пока все мои задания не будут завершены.

Как я могу «опросить» мои темы и вернуть результаты по мере их поступления?

Спасибо!

Answer 1

pool.map распространяет итерируемые элементы над пулом рабочих и агрегирует результаты, когда они все готовы.

Кроме того, pool.close и pool.join поручить Пулу подождать, пока все задачи не будут выполнены.

Если вы хотите обрабатывать результаты по мере их поступления, вы должны использовать pool.apply_async и использовать обратный вызов. Или вы можете собрать AsyncResult объекты, возвращенные pool.apply_async, и перебрать их, чтобы увидеть, когда каждый из них готов, но вся логика будет довольно громоздкой.

from multiprocessing.pool import ThreadPool 

pool = ThreadPool(20) 
tasks = [] 

def callback(result): 
    # handle the result of your function here 
    print result 

for url in urls: 
    pool.apply_async(openurl, args=[url], callback=callback) 

pool.close() 
pool.join()