Архив Октябрь, 2009
Пул потоков. Модуль Threading.
Я думаю, пример из SEO-софта на тему пула потоков долго искать не придется. Взять тот же Хрумер. Указал количество потоков, например 20, запустил — все 20 работают, как только один из них завершит работу — запускается следующий и так, пока не закончатся ссылки. Наверное в 9 из 10 случаев я пишу скрипт с использованием пула потоков. Рассмотрим некторые примеры.
Пример 1. «В лоб».
threads = [] for x in range(THREADS): new_thread = threading.Thread(target=step_one) threads.append(new_thread) new_thread.start() time.sleep(1) while True: # какое-то условие for y in threads[:]: if not y.isAlive(): threads.remove(y) new_thread = threading.Thread(target=step_one) threads.append(new_thread) new_thread.start() time.sleep(0.1) while threading.activeCount()>0: time.sleep(5)
Функцию step_one запускаем в несколько потоков, все созданные потоки храним в списке. Затем провереяем в цикле какие потоки завершили работу и, если такие есть, создаем новые и так, пока не достигнем какого-либо результата. Криво, неудобно, зато сразу понятно. Достаточно долго пользовался такой реализацией, а всё из-за лени.
Пример 2. Семафоры.
import time import threading max_potokov = 20 maxconnections = 5 akkiQueue = threading.BoundedSemaphore(value=maxconnections) def doform(): akkiQueue.acquire() # код функции akkiQueue.release() for i in xrange(max_potokov): time.sleep(1) p = threading.Thread(target=doform) p.setDaemon(True) p.start() while threading.activeCount()>0: time.sleep(5)
Уже лучше, количество потоков контролируется, но надо заранее знать сколько будет потоков, в общем как-то изворачиваться если тебе надо будет прервать все эти потоки по условию.
Пример 3. Очереди.
queue = Queue.Queue() # создаем очередь def register(item): #Функция с основным кодом return def repeat(): while True: try: item = queue.get_nowait() # ждём данные except Queue.Empty: break register(item) # передаем данные в нашу функцию time.sleep(0.5) queue.task_done() # задача завершена for item in range(MAX_THREADS): queue.put(item) # заносим данные в очередь for i in xrange(THREADS): t = threading.Thread(target=repeat) # создаем нить t.start() # стартуем time.sleep(0.5) queue.join() # блокируем очередь до завершения
Здесь используется модуль Queue. На данный для меня самый удобный варинат — и выглядит красиво, и данные передавать можно.
Помимо тех вариантов, что я описал, можно погуглить на тему threadpool и найти какие-нибудь модули для организации пула потоков.