Threading в python с использованием очереди

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

он ставит бесконечный цикл while. Выполняется ли каждый поток непрерывно с завершением до тех пор, пока все они не будут завершены? Я что-то пропустил.

#!/usr/bin/env python
import Queue
import threading
import urllib2
import time

hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com",
"http://ibm.com", "http://apple.com"]

queue = Queue.Queue()

class ThreadUrl(threading.Thread):
  """Threaded Url Grab"""
  def __init__(self, queue):
    threading.Thread.__init__(self)
    self.queue = queue

  def run(self):
    while True:
      #grabs host from queue
      host = self.queue.get()

      #grabs urls of hosts and prints first 1024 bytes of page
      url = urllib2.urlopen(host)
      print url.read(1024)

      #signals to queue job is done
      self.queue.task_done()

start = time.time()
def main():

  #spawn a pool of threads, and pass them queue instance 
  for i in range(5):
    t = ThreadUrl(queue)
    t.setDaemon(True)
    t.start()

  #populate queue with data   
  for host in hosts:
    queue.put(host)

  #wait on the queue until everything has been processed     
  queue.join()

main()
print "Elapsed Time: %s" % (time.time() - start)

Ответ 1

Настройка потока для потока daemon приводит к тому, что они завершатся, когда основное выполняется. Но да, вы правы в том, что ваши потоки будут работать непрерывно до тех пор, пока что-то в queue еще не будет заблокировано.

Документация объясняет эту деталь Документы очереди

Документация Threading на python объясняет также часть daemon.

Вся программа Python завершается, когда не осталось ни одного живого не-демона.

Итак, когда очередь опустеет, и queue.join возобновится, когда интерпретатор выходит из потоков, тогда умрет.

EDIT: Исправление по умолчанию по умолчанию для queue

Ответ 2

Ваш script отлично работает для меня, поэтому я предполагаю, что вы спрашиваете, что происходит, чтобы вы могли лучше понять его. Да, ваш подкласс помещает каждый поток в бесконечный цикл, ожидая чего-то, что нужно положить в очередь. Когда что-то найдено, оно захватывает его и делает свое дело. Затем, критическая часть, она уведомляет очередь, которую она сделала с queue.task_done, и возобновляет ожидание очередного элемента в очереди.

Пока все это происходит с рабочими потоками, основной поток ожидает (join) до тех пор, пока не будут выполнены все задачи в очереди, которые будут, когда потоки отправят флаг queue.task_done одинаковое количество раз как сообщения в очереди. В этот момент основной поток заканчивается и выходит. Так как это потоки деамонов, они тоже закрываются.

Это классный материал, потоки и очереди. Это одна из действительно хороших частей Python. Вы услышите всевозможные вещи о том, как нарезание резьбы на Python связано с GIL и т.д. Но если вы знаете, где их использовать (например, в этом случае с сетевым вводом-выводом), они действительно ускорят работу для вас. Общее правило: если вы связаны с I/O, попробуйте и протестируйте потоки; если вы связаны cpu, потоки, вероятно, не очень хорошая идея, возможно, вместо этого попробуйте процессы.

удачи,

Mike

Ответ 3

Я не думаю, что Queue необходим в этом случае. Использование только Thread:

import threading, urllib2, time

hosts = ["http://yahoo.com", "http://google.com", "http://amazon.com",
"http://ibm.com", "http://apple.com"]

class ThreadUrl(threading.Thread):
    """Threaded Url Grab"""
    def __init__(self, host):
        threading.Thread.__init__(self)
        self.host = host

    def run(self):
        #grabs urls of hosts and prints first 1024 bytes of page
        url = urllib2.urlopen(self.host)
        print url.read(1024)

start = time.time()
def main():
    #spawn a pool of threads
    for i in range(len(hosts)):
        t = ThreadUrl(hosts[i])
        t.start()

main()
print "Elapsed Time: %s" % (time.time() - start)