Introduzione
Quando si parla di multithreading e multiprocessing in Python, una delle difficoltà più comuni non è tanto avviare i task, quanto gestirli in modo ordinato. In molti progetti reali non basta creare thread o processi “a mano”: serve un meccanismo che limiti il numero di worker, raccolga i risultati, gestisca gli errori e permetta di scalare il codice senza complicarlo troppo.
Un sotto-argomento molto utile, pratico e spesso sottovalutato è l’uso di pool di esecuzione per coordinare lavoro concorrente. In Python questo approccio si realizza bene con il modulo concurrent.futures, che offre interfacce semplici sia per i thread sia per i processi. È una soluzione ideale quando vuoi:
- eseguire molte operazioni indipendenti in parallelo;
- limitare il numero di worker attivi;
- semplificare la raccolta dei risultati;
- scrivere codice più pulito rispetto alla gestione manuale di thread/processi.
In questo tutorial vedremo un esempio concreto: elaborare una lista di file simulando un flusso di lavoro reale, dove alcune operazioni sono I/O-bound e altre CPU-bound. Useremo un pool di thread per la parte di lettura e un pool di processi per la parte di calcolo, così da capire quando conviene ciascun modello.
Codice completo
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
from pathlib import Path
import hashlib
import time
import random
# -----------------------------
# Fase 1: simuliamo la lettura
# -----------------------------
def leggi_file(percorso: str) -> str:
"""
Simula la lettura di un file.
In un caso reale qui ci sarebbe I/O su disco, rete o database.
"""
time.sleep(random.uniform(0.1, 0.4)) # simula latenza I/O
return f"contenuto del file {percorso} con dati casuali {random.randint(1, 1000)}"
# -----------------------------
# Fase 2: simuliamo un calcolo CPU-bound
# -----------------------------
def calcola_hash(contenuto: str) -> str:
"""
Calcola un hash SHA-256 del contenuto.
Questa operazione è CPU-bound e si presta bene al multiprocessing.
"""
digest = hashlib.sha256(contenuto.encode("utf-8")).hexdigest()
return digest
def elabora_file(percorso: str) -> tuple[str, str]:
"""
Pipeline completa per un file:
1) lettura
2) calcolo hash
"""
contenuto = leggi_file(percorso)
hash_calcolato = calcola_hash(contenuto)
return percorso, hash_calcolato
def main():
# Lista di "file" da elaborare
file_da_processare = [f"file_{i}.txt" for i in range(1, 11)]
risultati = []
print("=== Fase 1: lettura parallela con thread ===")
# I thread sono adatti a I/O-bound: molte attese, poco calcolo.
with ThreadPoolExecutor(max_workers=4) as executor:
future_to_file = {
executor.submit(leggi_file, percorso): percorso
for percorso in file_da_processare
}
contenuti = {}
for future in as_completed(future_to_file):
percorso = future_to_file[future]
try:
contenuto = future.result()
contenuti[percorso] = contenuto
print(f"Lettura completata: {percorso}")
except Exception as e:
print(f"Errore nella lettura di {percorso}: {e}")
print("n=== Fase 2: calcolo parallelo con processi ===")
# I processi sono adatti a CPU-bound: sfruttano più core.
with ProcessPoolExecutor(max_workers=4) as executor:
future_to_file = {
executor.submit(calcola_hash, contenuto): percorso
for percorso, contenuto in contenuti.items()
}
for future in as_completed(future_to_file):
percorso = future_to_file[future]
try:
hash_calcolato = future.result()
risultati.append((percorso, hash_calcolato))
print(f"Hash calcolato per: {percorso}")
except Exception as e:
print(f"Errore nel calcolo di {percorso}: {e}")
print("n=== Risultati finali ===")
for percorso, hash_calcolato in sorted(risultati):
print(f"{percorso} - {hash_calcolato}")
if __name__ == "__main__":
main()
Spiegazione
Il codice mostra una strategia molto comune nei programmi reali: separare il lavoro I/O-bound dal lavoro CPU-bound.
1. Lettura parallela con ThreadPoolExecutor
La funzione leggi_file() simula un’operazione lenta, come la lettura da disco o una chiamata a una API. In questi casi i thread sono spesso la scelta migliore, perché mentre un thread aspetta, un altro può continuare a lavorare. Questo rende il programma più reattivo senza sprecare tempo di CPU.
Nel blocco:
with ThreadPoolExecutor(max_workers=4) as executor:stiamo creando un pool con massimo 4 thread attivi. Questo è importante: non conviene creare un numero illimitato di thread, perché aumenterebbe l’overhead e peggiorerebbe le prestazioni.
2. Calcolo parallelo con ProcessPoolExecutor
La funzione calcola_hash() rappresenta un’operazione CPU-bound. In Python, il GIL limita l’esecuzione simultanea di thread quando il lavoro è principalmente computazionale. Per questo, quando devi sfruttare più core della CPU, i processi sono spesso la scelta giusta.
Con ProcessPoolExecutor ogni task viene eseguito in un processo separato. Questo comporta un piccolo costo di serializzazione dei dati, ma permette di ottenere parallelismo reale nei calcoli.
3. Raccolta dei risultati con as_completed
La funzione as_completed() restituisce i future man mano che finiscono. Questo è molto utile perché non devi aspettare che tutti i task terminino nell’ordine di invio: puoi elaborare subito i risultati pronti. In un sistema reale questo approccio migliora la reattività e semplifica il logging degli errori.
4. Gestione delle eccezioni
Ogni chiamata a future.result() è racchiusa in un blocco try/except. È una best practice fondamentale: nei programmi concorrenti gli errori non devono bloccare tutto il flusso. Un singolo task fallito non dovrebbe compromettere l’intera elaborazione.
Best practice
- Scegli il modello giusto: usa i thread per I/O-bound e i processi per CPU-bound.
- Limita il numero di worker: un valore troppo alto può peggiorare le prestazioni invece di migliorarle.
- Evita stato condiviso non necessario: con i processi i dati vengono copiati o serializzati, quindi il design deve essere il più possibile immutabile.
- Usa
with: i context manager chiudono correttamente pool e risorse anche in caso di errore. - Gestisci sempre le eccezioni: in ambienti concorrenti gli errori vanno raccolti e registrati con attenzione.
- Misura prima di ottimizzare: non dare per scontato che più concorrenza significhi più velocità. Dipende dal carico, dall’I/O e dal costo di serializzazione.
- Preferisci funzioni pure nei processi: il codice inviato a un
ProcessPoolExecutordeve essere serializzabile e meglio se privo di effetti collaterali.
Riepilogo
In questo tutorial abbiamo visto un approccio pratico per coordinare lavoro concorrente in Python usando i pool di esecuzione. Il punto chiave è questo:
- ThreadPoolExecutor è ideale per attività che attendono risorse esterne, come file, rete o database;
- ProcessPoolExecutor è più adatto a calcoli intensivi che possono sfruttare più core;
- as_completed() aiuta a gestire i risultati appena disponibili;
- una buona architettura separa chiaramente le fasi del lavoro e rende il codice più manutenibile.
Questo schema è molto utile in applicazioni reali come pipeline di elaborazione documenti, download massivi, analisi dati e automazioni che combinano I/O e calcolo. Imparare a usare bene i pool significa scrivere programmi concorrenti più semplici, robusti e facili da scalare.
Approfondisci con risorse ufficiali
- Documentazione Python - concurrent.futures:
https://docs.python.org/3/library/concurrent.futures.html - Documentazione Python - threading:
https://docs.python.org/3/library/threading.html - Documentazione Python - multiprocessing:
https://docs.python.org/3/library/multiprocessing.html - Documentazione Python - hashlib:
https://docs.python.org/3/library/hashlib.html - PEP 3148 - futures:
https://peps.python.org/pep-3148/