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

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 вызовет исключение в вашем основном потоке, и все потоки демона выйдут при выходе из интерпретатора. Это предполагает, что вы не хотите выполнять очистку во всех этих потоках перед их выходом.

Более сложный способ - создать глобальное stopped событие :

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