За всё время работы скрипта, стартует только 10 потоков, просто каждый из них, отработав со своим 1 данным, не дохнет, а берет из списка следующее 1 данное и делает с ним эту же работу:
# -*- coding: utf-8 -*-
import threading, Queue, time
import traceback
class Worker(threading.Thread):
def __init__(self,queue):
threading.Thread.__init__(self)
self.__queue = queue
def run(self):
while True:
try: item = self.__queue.get_nowait() # ждём данные
except Queue.Empty: break # данные закончились, прекращаем работу
try: self.work(item) # работа
except Exception: traceback.print_exc()
time.sleep(0.5)
self.__queue.task_done() # задача завершена
return
def work(self,item):
print item
#pass
def main():
# Выводим в 5 потоков цифры от 1 до 100.
queue = Queue.Queue()
num_threads = 5 # 5 потоков
for x in xrange(100):
queue.put(x) # заносим данные в очередь
for i in xrange(num_threads):
t = Worker(queue) # создаем поток
t.start() # стартуем
time.sleep(0.1) # чтобы в консоли друг на друга не накладывались
queue.join() #блокируем выполнение программы, пока не будут израсходованы данные.
print "Done!"
if __name__ == '__main__':
main()
И, в общем, хотелось бы еще, чтобы можно всё это в любой момент можно было бы остановить. Но поскольку мы выполнение программы блокируем с помощью queue.join(), то скрипт наши Ctr+C банально "не услышит". Я начал искать, что делать в таком случае, нашел вот такое решение, но у меня оно работало абы как. А потом, поразмыслив, я сам придумал решение, которое оказалось ужасно простым.
# -*- coding: utf-8 -*-
import threading, Queue, time
import traceback
class Worker(threading.Thread):
def __init__(self,queue):
threading.Thread.__init__(self)
self.__queue = queue
self.kill_received = False # флаг прекращения работы
def run(self):
while not self.kill_received:
try: item = self.__queue.get_nowait() # ждём данные
except Queue.Empty: break
try: self.work(item)
except Exception: traceback.print_exc()
time.sleep(0.5)
self.__queue.task_done() # задача завершена
self.__queue.put(item) # зациклим
return
def work(self,item):
print item
def main():
queue = Queue.Queue()
num_threads = 5 # 5 потоков
threads = []
for x in xrange(100):
queue.put(x) # заносим данные в очередь
for i in xrange(num_threads):
t = Worker(queue) # создаем нить
threads.append(t)
t.start() # стартуем
time.sleep(0.1)
#Пока в "живых" не останется только главный поток, ждем.
while threading.activeCount()>1:
try:
time.sleep(1)
except KeyboardInterrupt:
print "Ctrl-c received! Sending kill to threads..."
for t in threads:
t.kill_received = True # даем сигнал о завершении всем потокам
print "Done!"
if __name__ == '__main__':
main()
Рекомендую пользоваться отлаженными решениями:
ОтветитьУдалить* https://bitbucket.org/lorien/grab/src/2cc24bff3a5a/grab/tools/work.py - пул потоков с поддержкой забора заданий из итератора, сам интерфейс пула также реализован в виде итератора
* https://bitbucket.org/lorien/grab/src/2cc24bff3a5a/grab/tools/watch.py - удобная штука для прерывания всех тредов, суть в том, что делается форк в начале программы, единстенная задача которого ловить ^C и убивать второй процесс.
Да я своим пулом тоже давно пользуюсь, пока устраивал. А вот второй пример я только вчера написал, вот его тестить надо. work.py, угу, видел ранее, но не осилил, если честно=) watch.py гляну, спасибо.
ОтветитьУдалитьпроще будет:
ОтветитьУдалитьsignal.signal(signal.SIGINT, lambda *args: exit())