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

Если у меня есть программа, которая использует потоки и очередь, как мне получить исключения, чтобы остановить выполнение? Вот примерная программа, которую невозможно остановить с помощью 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

Ответы

Ответ 1

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

while True:
    sleep(1)

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

Более сложным способом является глобальный stopped Event:

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. Конечно, с моими приведенными выше примерами вам не нужно присоединяться, но у вас может быть и другая причина для этого.

Ответ 2

A (возможно) offtopic note:

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

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