El módulo `threading` en Python

Este artículo explica el módulo threading en Python.

YouTube Video

El módulo threading en Python

El módulo threading en Python es una biblioteca estándar que admite la programación multihilo. Usar subprocesos permite que varios procesos se ejecuten simultáneamente, lo que puede mejorar el rendimiento de los programas, especialmente en casos que impliquen operaciones de bloqueo como esperas de E/S. Debido al Global Interpreter Lock (GIL) de Python, la efectividad de la programación multihilo es limitada para operaciones que dependen del CPU, pero funciona eficientemente para operaciones que dependen de E/S.

Las siguientes secciones explican los fundamentos del uso del módulo threading y cómo controlar los hilos.

Uso básico de los subprocesos

Crear y ejecutar subprocesos

Para crear un subproceso y realizar procesamiento concurrente, utiliza la clase threading.Thread. Especifica la función objetivo para crear un subproceso y ejecutarlo.

 1import threading
 2import time
 3
 4# Function to be executed in a thread
 5def worker():
 6    print("Worker thread started")
 7    time.sleep(2)
 8    print("Worker thread finished")
 9
10# Create and start the thread
11thread = threading.Thread(target=worker)
12thread.start()
13
14# Processing in the main thread
15print("Main thread continues to run")
16
17# Wait for the thread to finish
18thread.join()
19print("Main thread finished")

• En este ejemplo, la función worker se ejecuta en un subproceso separado, mientras que el subproceso principal continúa operando. Al llamar al método join(), el subproceso principal espera a que el subproceso secundario termine.

Nombrar hilos

Dar nombres significativos a los hilos facilita el registro y la depuración. Puedes especificarlo con el argumento name.

 1import threading
 2import time
 3
 4# Function to be executed in a thread
 5def worker():
 6    print("Worker thread started")
 7    time.sleep(2)
 8    print("Worker thread finished")
 9
10t = threading.Thread(
11    target=worker,
12    args=("named-worker", 0.3),
13    name="MyWorkerThread"
14)
15
16t.start()
17
18print("Active threads:", threading.active_count())
19for th in threading.enumerate():
20    print(" -", th.name)
21
22t.join()

• threading.enumerate() devuelve una lista de los hilos actuales, lo cual es útil para la depuración y el monitoreo del estado.

Heredar la clase Thread

Si deseas personalizar la clase que ejecuta los subprocesos, puedes definir una nueva clase heredando la clase threading.Thread.

 1import threading
 2import time
 3
 4# Inherit from the Thread class
 5class WorkerThread(threading.Thread):
 6    def __init__(self, name, delay, repeat=3):
 7        super().__init__(name=name)
 8        self.delay = delay
 9        self.repeat = repeat
10        self.results = []
11
12    def run(self):
13        for i in range(self.repeat):
14            msg = f"{self.name} step {i+1}"
15            print(msg)
16            self.results.append(msg)
17            time.sleep(self.delay)
18
19# Create and start the threads
20t1 = WorkerThread("Worker-A", delay=0.4, repeat=3)
21t2 = WorkerThread("Worker-B", delay=0.2, repeat=5)
22
23t1.start()
24t2.start()
25
26t1.join()
27t2.join()
28
29print("Results A:", t1.results)
30print("Results B:", t2.results)

• En este ejemplo, se sobreescribe el método run() para definir el comportamiento del hilo, lo que permite que cada hilo mantenga sus propios datos. Esto es útil cuando los hilos realizan procesamiento complejo o cuando se desea que cada hilo tenga sus propios datos independientes.

Sincronización entre subprocesos

Cuando varios subprocesos acceden simultáneamente a recursos compartidos, pueden ocurrir condiciones de carrera (data races). Para evitar esto, el módulo threading ofrece varios mecanismos de sincronización.

Bloqueo (Lock)

El objeto Lock se usa para implementar control exclusivo de recursos entre subprocesos. Mientras un subproceso bloquea un recurso, otros subprocesos no pueden acceder a ese recurso.

 1import threading
 2
 3lock = threading.Lock()
 4shared_resource = 0
 5
 6def worker():
 7    global shared_resource
 8    with lock:  # Acquire the lock
 9        local_copy = shared_resource
10        local_copy += 1
11        shared_resource = local_copy
12
13threads = [threading.Thread(target=worker) for _ in range(5)]
14
15for t in threads:
16    t.start()
17
18for t in threads:
19    t.join()
20
21print(f"Final value of shared resource: {shared_resource}")

• En este ejemplo, cinco subprocesos acceden a un recurso compartido, pero se usa el Lock para evitar que varios subprocesos modifiquen los datos simultáneamente.

Bloqueo reenrentrant (RLock)

Si un mismo hilo necesita adquirir un bloqueo varias veces, usa un RLock (bloqueo reenrentrant). Esto es útil para llamadas recursivas o para llamadas a bibliotecas que puedan adquirir bloqueos en diferentes llamadas.

 1import threading
 2
 3rlock = threading.RLock()
 4shared = []
 5
 6def outer():
 7    with rlock:
 8        shared.append("outer")
 9        inner()
10
11def inner():
12    with rlock:
13        shared.append("inner")
14
15t = threading.Thread(target=outer)
16t.start()
17t.join()
18print(shared)

• Con RLock, el mismo hilo puede volver a adquirir un bloqueo que ya posee, lo que ayuda a evitar bloqueos muertos (deadlocks) en la adquisición anidada de bloqueos.

Condición (Condition)

Condition se utiliza para que los hilos esperen hasta que se cumpla una condición específica. Cuando un hilo cumple una condición, puedes llamar a notify() para notificar a otro hilo, o a notify_all() para notificar a todos los hilos en espera.

A continuación se muestra un ejemplo de un productor y un consumidor utilizando una Condition.

 1import threading
 2
 3condition = threading.Condition()
 4shared_data = []
 5
 6def producer():
 7    with condition:
 8        shared_data.append(1)
 9        print("Produced an item")
10        condition.notify()  # Notify the consumer
11
12def consumer():
13    with condition:
14        condition.wait()  # Wait until the condition is met
15
16        item = shared_data.pop(0)
17        print(f"Consumed an item: {item}")
18
19# Create the threads
20producer_thread = threading.Thread(target=producer)
21consumer_thread = threading.Thread(target=consumer)
22
23consumer_thread.start()
24producer_thread.start()
25
26producer_thread.join()
27consumer_thread.join()

• Este código utiliza una Condition para que el productor notifique cuando se agregan datos, y el consumidor espera esa notificación antes de recuperar los datos, logrando la sincronización.

Daemonización de hilos

Los hilos daemon se terminan forzosamente cuando finaliza el hilo principal. Mientras que los hilos normales deben esperar para finalizar, los hilos daemon se terminan automáticamente.

 1import threading
 2import time
 3
 4def worker():
 5    while True:
 6        print("Working...")
 7        time.sleep(1)
 8
 9# Create a daemon thread
10thread = threading.Thread(target=worker)
11thread.daemon = True  # Set as a daemon thread
12
13thread.start()
14
15# Processing in the main thread
16time.sleep(3)
17print("Main thread finished")

• En este ejemplo, el hilo worker está daemonizado, por lo que se termina forzosamente cuando finaliza el hilo principal.

Gestión de Hilos con ThreadPoolExecutor

Además del módulo threading, puede usar el ThreadPoolExecutor del módulo concurrent.futures para gestionar un pool de hilos y ejecutar tareas en paralelo.

 1from concurrent.futures import ThreadPoolExecutor
 2import time
 3
 4def worker(seconds):
 5    print(f"Sleeping for {seconds} second(s)")
 6    time.sleep(seconds)
 7    return f"Finished sleeping for {seconds} second(s)"
 8
 9with ThreadPoolExecutor(max_workers=3) as executor:
10    futures = [executor.submit(worker, i) for i in range(1, 4)]
11    for future in futures:
12        print(future.result())

• ThreadPoolExecutor crea un pool de hilos y procesa las tareas de manera eficiente. Especifique el número de hilos que se ejecutarán simultáneamente con max_workers.

Comunicación de Eventos Entre Hilos

Usando threading.Event, puede configurar banderas entre hilos para notificar a otros hilos sobre la ocurrencia de un evento.

 1import threading
 2import time
 3
 4event = threading.Event()
 5
 6def worker():
 7    print("Waiting for event to be set")
 8    event.wait()  # Wait until the event is set
 9
10    print("Event received, continuing work")
11
12thread = threading.Thread(target=worker)
13thread.start()
14
15time.sleep(2)
16print("Setting the event")
17event.set()  # Set the event and notify the thread

• Este código demuestra un mecanismo donde el hilo de trabajo espera la señal del Event y reanuda el procesamiento cuando el hilo principal llama a event.set().

Manejo de excepciones y terminación de hilos en hilos

Cuando ocurren excepciones en los hilos, no se propagan directamente al hilo principal, por lo que se necesita un patrón para capturar y compartir las excepciones.

 1import threading
 2import queue
 3
 4def worker(err_q):
 5    try:
 6        raise ValueError("Something bad")
 7    except Exception as e:
 8        err_q.put(e)
 9
10q = queue.Queue()
11t = threading.Thread(target=worker, args=(q,))
12t.start()
13t.join()
14if not q.empty():
15    exc = q.get()
16    print("Worker raised:", exc)

• Al colocar las excepciones en una Queue y recuperarlas en el hilo principal, puedes detectar fallos de manera fiable. Si usas concurrent.futures.ThreadPoolExecutor, las excepciones se relanzan con future.result(), lo que facilita su manejo.

El GIL (Global Interpreter Lock) y sus efectos

Debido al mecanismo del GIL (Global Interpreter Lock) en CPython, múltiples bytecodes de Python no se ejecutan realmente de forma simultánea dentro del mismo proceso. Para tareas que requieren mucha CPU, como cálculos pesados, se recomienda usar multiprocessing. Por otro lado, para tareas limitadas por E/S, como la lectura de archivos o la comunicación en red, threading funciona de manera eficaz.

Resumen

Usando el módulo threading de Python, puede implementar programas multihilo y ejecutar múltiples procesos simultáneamente. Con mecanismos de sincronización como Lock y Condition, puede acceder de manera segura a recursos compartidos y realizar sincronizaciones complejas. Además, usando hilos daemon o ThreadPoolExecutor, la gestión de hilos y el procesamiento paralelo eficiente se vuelven más sencillos.

Puedes seguir el artículo anterior utilizando Visual Studio Code en nuestro canal de YouTube. Por favor, también revisa nuestro canal de YouTube.

YouTube Video