Заполнение очереди и управление многопроцессорностью в python

У меня проблема с python:

  • У меня есть очередь URL-адресов, которые мне нужно проверять время от времени
  • Если очередь заполнена, мне нужно обработать каждый элемент в очереди
  • Каждый элемент в очереди должен обрабатываться одним процессом (многопроцессорным)

До сих пор мне удалось добиться этого "вручную" следующим образом:

while 1:
        self.updateQueue()

        while not self.mainUrlQueue.empty():
            domain = self.mainUrlQueue.get()

            # if we didn't launched any process yet, we need to do so
            if len(self.jobs) < maxprocess:
                self.startJob(domain)
                #time.sleep(1)
            else:
                # If we already have process started we need to clear the old process in our pool and start new ones
                jobdone = 0

                # We circle through each of the process, until we find one free ; only then leave the loop 
                while jobdone == 0:
                    for p in self.jobs :
                        #print "entering loop"
                        # if the process finished
                        if not p.is_alive() and jobdone == 0:
                            #print str(p.pid) + " job dead, starting new one"
                            self.jobs.remove(p)
                            self.startJob(domain)
                            jobdone = 1

Однако это приводит к множеству проблем и ошибок. Я задавался вопросом, не лучше ли я, используя пул процессов. Каким будет правильный способ сделать это?

Однако много раз моя очередь пуста, и она может быть заполнена на 300 элементов за секунду, поэтому я не слишком уверен, как это сделать здесь.

Ответы

Ответ 1

Вы можете использовать возможности блокировки queue, чтобы вызвать многократный процесс при запуске (используя multiprocessing.Pool) и позволяя им спать до тех пор, пока некоторые данные не будут доступны в очереди для обработки. Если вы не знакомы с этим, вы можете попробовать "играть" с этой простой программой:

import multiprocessing
import os
import time

the_queue = multiprocessing.Queue()


def worker_main(queue):
    print os.getpid(),"working"
    while True:
        item = queue.get(True)
        print os.getpid(), "got", item
        time.sleep(1) # simulate a "long" operation

the_pool = multiprocessing.Pool(3, worker_main,(the_queue,))
#                            don't forget the coma here  ^

for i in range(5):
    the_queue.put("hello")
    the_queue.put("world")


time.sleep(10)

Протестировано с помощью Python 2.7.3 в Linux

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

При запуске все 3 процесса будут спать до тех пор, пока очередь не будет загружена с некоторыми данными. Когда данные доступны, один из ожидающих работников получает этот элемент и начинает его обрабатывать. После этого он пытается получить другой элемент из очереди, ожидая снова, если ничего не доступно...