Использование общей очереди, в которой рабочие могут добавлять задачи к

Question

Я довольно новичок в python (я в основном пишу код на Java). У меня есть скрипт python, который по сути является сканером. Он вызывает phantomjs, который загружает страницу, возвращает ее источник и список URL-адресов, найденных на этой странице.

Я пытался использовать Python 3's multiprocessing, чтобы сделать это, но я не могу понять, как использовать общую очередь, к которой могут добавить и рабочие. Я получаю непредсказуемые результаты.

В моем предыдущем подходе использовался глобальный список URL-адресов, из которого я извлек кусок и отправил работникам, используя map_async. В конце я собирал все возвращенные URL-адреса и добавлял их в глобальный список. Проблема в том, что каждый «кусок» занимает столько же времени, сколько и самый медленный рабочий. Я пытаюсь изменить его, чтобы всякий раз, когда рабочий был сделан, он мог выбрать следующий URL-адрес. Однако я не думаю, что делаю это правильно. Вот что у меня есть до сих пор:

def worker(url, urls): 
    print(multiprocessing.current_process().name + "." + str(multiprocessing.current_process().pid) + " loading " + url) 
    returned_urls = phantomjs(url) 
    print(multiprocessing.current_process().name + "." + str(multiprocessing.current_process().pid) + " returning " + str(len(returned_urls)) + " URLs") 

    for returned_url in returned_urls: 
     urls.put(returned_url, block=True) 

    print("There are " + str(urls.qsize()) + " URLs in total.\n") 

if __name__ == '__main__':  
    manager = multiprocessing.Manager() 
    urls = manager.Queue() 
    urls.append(<some-url>) 

    pool = Pool() 
    while True: 
     url = urls.get(block=True) 
     pool.apply_async(worker, (url, urls)) 

    pool.close() 
    pool.join() 

Если есть лучший способ сделать это, пожалуйста, дайте мне знать. Я просматриваю известный сайт, и конечное условие завершения - это когда нет URL-адресов для обработки. Но сейчас похоже, что я буду продолжать работать навсегда. Я не уверен, буду ли я использовать queue.empty(), потому что он говорит, что он не является надежным.

Answer 1

Вот что я бы, вероятно:

def worker(url, urls): 
    print(multiprocessing.current_process().name + "." + str(multiprocessing.current_process().pid) + " loading " + url) 
    returned_urls = phantomjs(url) 
    print(multiprocessing.current_process().name + "." + str(multiprocessing.current_process().pid) + " returning " + str(len(returned_urls)) + " URLs") 

     for returned_url in returned_urls: 
      urls.put(returned_url, block=True) 

     # signal finished processing this url 
     urls.put('no-url') 

    print("There are " + str(urls.qsize()) + " URLs in total.\n") 

if __name__ == '__main__':  
    manager = multiprocessing.Manager() 
    pool = Pool() 
    urls = manager.Queue() 

    # start first url before entering loop 
    counter = 1 
    pool.apply_async(worker, (<some-url>, urls)) 

    while counter > 0: 
     url = urls.get(block=True) 
     if url == 'no-url': 
      # a url has finished processing 
      counter -= 1 
     else: 
      # a new url needs to be processed 
      counter += 1 
      pool.apply_async(worker, (url, urls)) 

    pool.close() 
    pool.join() 

Всякий раз, когда URL извлекается из очереди, увеличить счетчик. Подумайте об этом как о счетчике «в настоящее время обработки URL». Когда из очереди выдается «нет-url», завершается «текущий обрабатывающий URL-адрес», поэтому уменьшайте счетчик. Пока счетчик больше 0, есть URL-адреса, которые еще не завершили обработку и еще не вернули «нет-url».

EDIT

Как я уже сказал в комментарии (поместить здесь для всех, кто читает его), при использовании multiprocessing.Pool, вместо того чтобы думать о нем, как отдельные процессы, то лучше думать о нем, как единственная конструкция, которая выполняет вашу функцию каждый раз, когда она получает данные (одновременно, когда это возможно). Это наиболее полезно для проблем с данными, в которых вы не отслеживаете или не заботитесь о отдельных рабочих процессах только обрабатываемыми данными.

Answer 2

Вот как я решил проблему. Я первоначально пошел с проектом, опубликованным в this answer, но bj0 упомянул, что он злоупотреблял функцией инициализации. Поэтому я решил сделать это, используя apply_async, по модулю, подобному коду, опубликованному в моем вопросе.

Поскольку мои работники изменить очередь они читают URL, из (они добавляют к нему), я думал, что я мог бы просто запустить свой цикл следующим образом:

while not urls.empty(): 
    pool.apply_async(worker, (urls.get(), urls)) 

Я ожидал, что это будет работать, так как рабочие добавит в очередь, а apply_async будет ждать, если все рабочие будут заняты. Это не сработало, как я ожидал, и цикл закончился раньше. Проблема заключалась в том, что неясно, что apply_asyncне блокирует, если все рабочие заняты. Вместо этого он будет помещать поставленные задачи в очередь, а это значит, что urls со временем станет пустым и цикл завершится. Единственный раз, когда блокирует цикл, если очередь пуста, когда вы пытаетесь выполнить urls.get(). На этом этапе он будет ожидать появления большего количества элементов в очереди. Но мне все еще нужно было выяснить, как закончить цикл. Условие состоит в том, что цикл должен заканчиваться, когда ни один из рабочих не возвращает новые URL-адреса. Для этого я использую общий dict, который устанавливает значение, связанное с именем процесса, равным 0, если процесс не возвратил никаких URL-адресов, и 1 в противном случае. Я проверяю сумму ключей на каждой итерации цикла, и если это когда-либо 0, я знаю, что я сделан.

Основная структура заканчивал тем, как это:

def worker(url, url_queue, proc_user_urls_queue, proc_empty_urls_queue): 

    returned_urls = phantomjs(url) # calls phantomjs and waits for output 
    if len(returned_urls) > 0: 
     proc_empty_urls_queue.put(
      [multiprocessing.current_process().name, 1] 
     ) 
    else: 
     proc_empty_urls_queue.put(
      [multiprocessing.current_process().name, 0] 
     ) 

    for returned_url in returned_urls: 
     url_queue.put(returned_url) 

def empty_url_tallier(proc_empty_urls_queue, proc_empty_urls_dict): 
    while 1: 
     # This may not be necessary. I don't know if this worker is run 
     # by the same process every time. If not, it is possible that 
     # the worker was assigned the task of fetching URLs, and returned 
     # some. So let's make sure that we set its entry to zero anyway. 
     # If this worker is run by the same process every time, then this 
     # stuff is not necessary. 
     id = multiprocessing.current_process().name 
     proc_empty_urls_dict[id] = 0 

     proc_empty_urls = proc_empty_urls_queue.get() 
     if proc_empty_urls == "done": # poison pill 
      break 

     proc_id = proc_empty_urls[0] 
     proc_empty_url = proc_empty_urls[1] 
     proc_empty_urls_dict[proc_id] = proc_empty_url 

manager = Manager() 

urls = manager.Queue() 
proc_empty_urls_queue = manager.Queue() 
proc_empty_urls_dict = manager.dict() 

pool = Pool(33) 

pool.apply_async(writer, (proc_user_urls_queue,)) 
pool.apply_async(empty_url_tallier, (proc_empty_urls_queue, proc_empty_urls_dict)) 

# Run the first apply synchronously 
urls.put("<some-url>") 
pool.apply(worker, (urls.get(), urls, proc_empty_urls_queue)) 
while sum(proc_empty_urls_dict.values()) > 0: 
    pool.apply_async(worker, (urls.get(), urls, proc_empty_urls_queue)) 

proc_empty_urls_queue.put("done") # poison pill 
pool.close() 
pool.join()