Как обрабатывать исключения при использовании потоков и очереди?

Question

Если у меня есть программа, которая использует потоки и очередь, как мне получить исключения, чтобы остановить выполнение? Вот примерная программа, которую невозможно остановить с помощью ctrl-c (в основном вырванный из документов python).

from threading import Thread 
from Queue import Queue 
from time import sleep 

def do_work(item): 
    sleep(0.5) 
    print "working" , item 

def worker(): 
     while True: 
      item = q.get() 
      do_work(item) 
      q.task_done() 

q = Queue() 

num_worker_threads = 10 

for i in range(num_worker_threads): 
    t = Thread(target=worker) 
    # t.setDaemon(True) 
    t.start() 

for item in range(1, 10000): 
    q.put(item) 

q.join()  # block until all tasks are done 

Answer 1

Самый простой способ, чтобы начать все рабочие потоки, как демон потоков, то только ваш основной цикл будет

while True: 
    sleep(1) 

Нажатие Ctrl + C сгенерирует исключение в основном потоке, и все потоки демона выйдут, когда выйдет интерпретатор. Это предполагает, что вы не хотите выполнять очистку во всех этих потоках до их выхода.

Более сложный способ иметь глобальный stoppedEvent:

stopped = Event() 
def worker(): 
    while not stopped.is_set(): 
     try: 
      item = q.get_nowait() 
      do_work(item) 
     except Empty:  # import the Empty exception from the Queue module 
      stopped.wait(1) 

Тогда ваш основной цикл может установить stopped событие в False, когда он получает KeyboardInterrupt

try: 
    while not stopped.is_set(): 
     stopped.wait(1) 
except KeyboardInterrupt: 
    stopped.set() 

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

Обратите внимание, что в этом примере не используется q.join() - это усложняет работу, хотя вы все равно можете использовать его. Если вы это сделаете, то лучше всего использовать обработчики сигналов вместо исключений для обнаружения KeyboardInterrupt s. Например:

from signal import signal, SIGINT 
def stop(signum, frame): 
    stopped.set() 
signal(SIGINT, stop) 

Это позволяет определить, что происходит, когда вы нажмете Ctrl + C, не затрагивая все, что ваш основной цикл находится в середине. Таким образом, вы можете продолжать делать q.join(), не беспокоясь о прерывании Ctrl + C. Конечно, с моими приведенными выше примерами вам не нужно присоединяться, но у вас может быть и другая причина для этого.

Answer 2

A (возможно) оффтоп примечание:

(...) 
for item in range(1, 10000): 
    q.put(item) 
(...) 

Вы можете использовать xrange вместо диапазона здесь (если Вы не используете python3000). Таким образом вы будете экономить процессор и память. Подробнее о xrange можно найти here.