¿Cómo manejo las excepciones cuando uso threading y Queue?

Question

Si tengo un programa que usa threading y Queue, ¿cómo obtengo excepciones para detener la ejecución? Aquí hay un programa de ejemplo, que no es posible detener con ctrl-c (básicamente arrancado de los documentos de 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

La forma más sencilla es comenzar todos los subprocesos de trabajo en forma de hilos demonio, a continuación, sólo tiene su bucle principal sea

while True: 
    sleep(1) 

Golpear Ctrl + C producirá una excepción en su hilo principal, y todos de los hilos daemon saldrá cuando el intérprete salga. Esto supone que no desea realizar la limpieza en todos los subprocesos antes de que salgan.

Una forma más compleja es tener un mundial 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) 

A continuación, el bucle principal puede establecer el stopped de eventos para False cuando se pone un KeyboardInterrupt

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

Esto le permite al trabajador Los subprocesos terminan lo que están haciendo, en lugar de simplemente hacer que cada subproceso de trabajo sea un daemon y salir en el medio de la ejecución. También puedes hacer la limpieza que quieras.

Tenga en cuenta que este ejemplo no hace uso de q.join() - esto hace las cosas más complejas, aunque todavía puede usarlo. Si lo hace, la mejor opción es utilizar manejadores de señal en lugar de excepciones para detectar KeyboardInterrupt s. Por ejemplo:

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

Esto le permite definir qué ocurre cuando se pulse Ctrl + C sin afectar a cualquiera que sea su bucle principal se encuentra en medio de. De modo que puede seguir haciendo q.join() sin preocuparse de ser interrumpido por Ctrl + C. Por supuesto, con mis ejemplos anteriores, no es necesario que se una, pero es posible que tenga alguna otra razón para hacerlo.

Answer 2

A (posiblemente) nota offtopic:

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

usted podría utilizar en lugar de xrange gama aquí (a menos que utilice python3000). Ahorrará algo de CPU y memoria al hacerlo. Se puede encontrar más información sobre xrange here.