Threading en Python

Le module threading de Python vous permet d'exécuter plusieurs tâches dans le même processus en même temps. Chaque tâche s'exécute dans son propre thread — une unité d'exécution légère qui partage l'espace mémoire du processus. Le threading est l'outil adapté lorsque votre programme passe l'essentiel de son temps à attendre (lecture d'un fichier, requête HTTP, interrogation d'une base de données) et que vous souhaitez effectuer un travail utile pendant cette attente plutôt que de bloquer.

Ce chapitre couvre :

• Créer et démarrer des threads avec threading.Thread
• Attendre la fin des threads avec join
• Les threads démons et les tâches en arrière-plan
• Prévenir les courses aux données avec Lock et with
• Coordonner les threads avec Event et Semaphore
• La communication thread-safe avec queue.Queue
• Le ThreadPoolExecutor pour les pools de threads gérés
• Le Global Interpreter Lock (GIL) et pourquoi les threads n'accélèrent pas le code lié au CPU
• Quand choisir le threading plutôt que asyncio

Créer et démarrer un thread

Importez threading et créez un objet Thread en passant la fonction à exécuter comme target. Appelez .start() pour lancer le thread :

import threading
import time

def greet(name):
    time.sleep(0.5)         # simulate some work
    print(f'Hello, {name}!')

t = threading.Thread(target=greet, args=('Alice',))
t.start()
print('Thread started — main continues running')
t.join()                    # wait for the thread to finish
print('Thread finished')
# Thread started — main continues running
# Hello, Alice!
# Thread finished

Points clés :

• args est un tuple d'arguments positionnels passés à target. Utilisez kwargs pour les arguments nommés.
• Sans .join(), le thread principal peut se terminer avant le thread créé.
• .start() retourne immédiatement ; le nouveau thread s'exécute de manière concurrente.

Passer des arguments nommés

import threading

def connect(host, port=80):
    print(f'Connecting to {host}:{port}')

t = threading.Thread(target=connect, kwargs={'host': 'example.com', 'port': 443})
t.start()
t.join()
# Connecting to example.com:443

Exécuter plusieurs threads simultanément

Le véritable intérêt du threading est d'exécuter plusieurs tâches en parallèle. Démarrez d'abord tous les threads, puis attendez-les tous :

import threading
import time

def download(url):
    time.sleep(1)           # simulate a 1-second network request
    print(f'Downloaded: {url}')

urls = [
    'https://example.com/data1',
    'https://example.com/data2',
    'https://example.com/data3',
]

start = time.perf_counter()

threads = [threading.Thread(target=download, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()

elapsed = time.perf_counter() - start
print(f'All downloads finished in {elapsed:.1f}s')
# Downloaded: https://example.com/data1
# Downloaded: https://example.com/data2
# Downloaded: https://example.com/data3
# All downloads finished in 1.0s

Sans threads, cela prendrait 3 secondes (séquentiel). Avec trois threads, cela prend environ 1 seconde car les attentes se chevauchent.

Sous-classer Thread

Pour une logique plus complexe, sous-classez threading.Thread et redéfinissez run(). Stockez les résultats en tant qu'attributs d'instance pour que le code appelant puisse les lire après join() :

import threading
import time

class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None

    def run(self):
        time.sleep(0.5)     # simulate download
        self.result = f'Data from {self.url}'

threads = [DownloadThread(f'https://example.com/page{i}') for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

for t in threads:
    print(t.result)
# Data from https://example.com/page0
# Data from https://example.com/page1
# Data from https://example.com/page2

Threads démons

Un thread démon est un thread en arrière-plan que l'interpréteur tue automatiquement lorsque tous les threads non-démons se sont terminés. Marquez un thread comme démon en passant daemon=True (ou en définissant t.daemon = True avant .start()) :

import threading
import time

def heartbeat():
    while True:
        print('♥ still running')
        time.sleep(1)

t = threading.Thread(target=heartbeat, daemon=True)
t.start()

time.sleep(2.5)
print('Main thread exiting — daemon will be killed')
# ♥ still running
# ♥ still running
# Main thread exiting — daemon will be killed

Utilisez les threads démons pour les tâches de surveillance ou de journalisation en arrière-plan qui ne doivent pas empêcher le programme de se terminer. Ne les utilisez jamais pour des tâches qui doivent se terminer proprement (écriture de fichiers, commits en base de données) — ils sont tués sans nettoyage.

Noms et introspection des threads

Chaque thread possède un nom. Vous pouvez le définir explicitement ou laisser Python en attribuer un automatiquement. Utilisez threading.current_thread() pour inspecter le thread en cours d'exécution et threading.active_count() pour compter les threads actifs :

import threading

def worker():
    t = threading.current_thread()
    print(f'Running in thread: {t.name}')

t = threading.Thread(target=worker, name='WorkerThread-1')
t.start()
t.join()

print(f'Active threads: {threading.active_count()}')
# Running in thread: WorkerThread-1
# Active threads: 1

Synchronisation : prévenir les courses aux données

Les threads partagent la mémoire du processus. Quand deux threads lisent et écrivent simultanément la même variable, le résultat est une course aux données — un comportement non déterministe difficile à reproduire ou à déboguer.

L'exemple suivant sans verrou produit un compteur final imprévisible car les incréments de différents threads peuvent se chevaucher :

import threading

counter = 0

def unsafe_increment():
    global counter
    for _ in range(100_000):
        counter += 1         # read-modify-write: not atomic!

threads = [threading.Thread(target=unsafe_increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# counter is somewhere between 100000 and 500000 — unpredictable
print('Final counter:', counter)

Lock

Un threading.Lock garantit qu'un seul thread exécute la section protégée à la fois. Utilisez-le comme gestionnaire de contexte avec with pour que le verrou soit toujours libéré, même si une exception est levée :

import threading

counter = 0
lock = threading.Lock()

def safe_increment():
    global counter
    for _ in range(100_000):
        with lock:              # acquire before read-modify-write
            counter += 1        # now only one thread at a time can run this

threads = [threading.Thread(target=safe_increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print('Final counter:', counter)   # always 500000

RLock (verrou ré-entrant)

Si un thread doit acquérir le même verrou deux fois (par exemple, une méthode appelle une autre méthode qui acquiert également le verrou), utilisez threading.RLock. Il permet au même thread de ré-acquérir le verrou sans provoquer d'interblocage :

import threading

lock = threading.RLock()

def outer():
    with lock:
        print('Outer acquired')
        inner()             # inner also acquires the same lock

def inner():
    with lock:              # works because RLock counts acquisitions
        print('Inner acquired')

t = threading.Thread(target=outer)
t.start()
t.join()
# Outer acquired
# Inner acquired

Coordination des threads : Event et Semaphore

Event

threading.Event est un signal simple. Un thread appelle .set() pour signaler ; d'autres threads appellent .wait() pour bloquer jusqu'à l'arrivée du signal :

import threading
import time

ready = threading.Event()

def worker():
    print('Worker: waiting for signal...')
    ready.wait()            # blocks here until ready.set() is called
    print('Worker: signal received, starting work')

t = threading.Thread(target=worker)
t.start()

time.sleep(0.5)
print('Main: sending signal')
ready.set()
t.join()
# Worker: waiting for signal...
# Main: sending signal
# Worker: signal received, starting work

Utilisez un Event pour coordonner l'ordre de démarrage — par exemple, pour retarder les threads de travail jusqu'à l'établissement d'une connexion à la base de données.

Semaphore

Un threading.Semaphore limite le nombre de threads pouvant être simultanément dans une section. C'est utile pour limiter le taux d'accès à une ressource partagée comme un pool de connexions :

import threading
import time

# Allow at most 2 threads to enter the critical section at once
semaphore = threading.Semaphore(2)

def use_connection(name):
    with semaphore:
        print(f'{name}: using connection')
        time.sleep(0.5)
        print(f'{name}: releasing connection')

threads = [threading.Thread(target=use_connection, args=(f'T{i}',)) for i in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
# T0: using connection
# T1: using connection          <- only 2 at a time
# T0: releasing connection
# T2: using connection
# T1: releasing connection
# T3: using connection
# T2: releasing connection
# T3: releasing connection

Données locales aux threads

threading.local() crée un objet qui conserve des valeurs séparées par thread. C'est utile pour les caches ou les curseurs de base de données par thread :

import threading

local_data = threading.local()

def set_user(name):
    local_data.user = name       # each thread writes its own copy
    print(f'{threading.current_thread().name}: user = {local_data.user}')

threads = [
    threading.Thread(target=set_user, args=(f'user{i}',), name=f'Thread-{i}')
    for i in range(3)
]
for t in threads:
    t.start()
for t in threads:
    t.join()
# Thread-0: user = user0
# Thread-1: user = user1
# Thread-2: user = user2

Lire local_data.user dans un thread où il n'a jamais été défini lève une AttributeError, comme tout autre accès à un attribut.

Files d'attente thread-safe

La classe queue.Queue (du module queue de la bibliothèque standard, pas de asyncio) est un FIFO thread-safe. Les threads peuvent utiliser put et get sans verrou — toute la synchronisation est gérée en interne.

Le schéma classique est producteur-consommateur : un ou plusieurs threads producteurs génèrent du travail, des threads consommateurs le traitent :

import threading
import queue
import time

q = queue.Queue(maxsize=5)

def producer():
    for i in range(1, 5):
        q.put(f'item-{i}')
        print(f'Produced item-{i}')
        time.sleep(0.05)

def consumer():
    while True:
        item = q.get()
        if item is None:        # sentinel: stop when None is received
            break
        print(f'Consumed {item}')
        q.task_done()

prod = threading.Thread(target=producer)
cons = threading.Thread(target=consumer)

cons.start()
prod.start()
prod.join()
q.put(None)     # signal consumer to stop
cons.join()
# Produced item-1
# Consumed item-1
# Produced item-2
# Consumed item-2
# Produced item-3
# Consumed item-3
# Produced item-4
# Consumed item-4

queue.Queue propose également task_done() et join() pour suivre le traitement de tous les éléments en file, ainsi que queue.LifoQueue / queue.PriorityQueue pour des ordres alternatifs.

ThreadPoolExecutor : pools de threads gérés

Créer un nouvel objet Thread pour chaque tâche est coûteux lorsqu'on a de nombreuses tâches de courte durée. concurrent.futures.ThreadPoolExecutor gère un pool de threads de travail réutilisables et retourne des objets Future pour chaque tâche soumise :

import concurrent.futures
import time

def fetch_url(url):
    time.sleep(0.5)     # simulate network I/O
    return f'Response from {url}'

urls = [
    'https://api.example.com/users',
    'https://api.example.com/posts',
    'https://api.example.com/comments',
]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # submit all tasks and get Future objects
    futures = {executor.submit(fetch_url, url): url for url in urls}

    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        print(future.result())

# Response from https://api.example.com/users    (order may vary)
# Response from https://api.example.com/comments
# Response from https://api.example.com/posts

executor.map(fn, iterable) est une forme plus courte quand vous n'avez pas besoin d'objets Future individuels :

import concurrent.futures
import time

def square(n):
    time.sleep(0.01)
    return n * n

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(square, range(10)))

print(results)
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

executor.map préserve l'ordre des entrées dans sa sortie, contrairement à as_completed qui produit les résultats dans l'ordre de complétion.

Le Global Interpreter Lock (GIL)

CPython (l'interpréteur Python standard) possède un Global Interpreter Lock — un mutex qui n'autorise qu'un seul thread à exécuter du bytecode Python à la fois. Cela signifie que les threads en CPython ne peuvent pas exécuter du code Python en vrai parallèle sur plusieurs cœurs CPU.

L'implication pratique :

• Tâches liées aux entrées/sorties : les threads accélèrent réellement le programme. Pendant qu'un thread attend une réponse réseau, le GIL est libéré et un autre thread s'exécute. Tous les exemples ci-dessus illustrent ce comportement.
• Tâches liées au CPU : les threads n'accélèrent pas les choses et peuvent même être légèrement plus lents en raison du surcoût de la commutation de contexte.

import threading
import time

def cpu_bound(n):
    total = 0
    for i in range(n):
        total += i
    return total

# Sequential
start = time.perf_counter()
cpu_bound(5_000_000)
cpu_bound(5_000_000)
single = time.perf_counter() - start

# Two threads — GIL prevents true parallelism
start = time.perf_counter()
t1 = threading.Thread(target=cpu_bound, args=(5_000_000,))
t2 = threading.Thread(target=cpu_bound, args=(5_000_000,))
t1.start(); t2.start()
t1.join(); t2.join()
threaded = time.perf_counter() - start

print(f'Single-threaded: {single:.2f}s')
print(f'Two threads:     {threaded:.2f}s')
# Two threads are NOT faster (similar elapsed time)

Pour un vrai parallélisme CPU en Python, utilisez multiprocessing ou concurrent.futures.ProcessPoolExecutor — chaque processus possède son propre GIL.

Threading vs. asyncio

threading et asyncio accélèrent tous deux les programmes liés aux entrées/sorties, mais ils fonctionnent différemment :

Règle empirique : si vous utilisez une bibliothèque disposant d'une version compatible async (par exemple aiohttp à la place de requests), optez pour asyncio. Si vous êtes contraint d'utiliser des bibliothèques bloquantes synchrones, utilisez le threading. Pour le travail lié au CPU, utilisez multiprocessing.

Pièges courants

Démarrer un thread deux fois. Appeler .start() plusieurs fois sur le même objet Thread lève une RuntimeError. Créez une nouvelle instance de Thread pour chaque exécution.

Oublier le join. Un thread non joint peut encore être en cours d'exécution à la fin du programme. Toujours join les threads dont l'achèvement importe, ou faites-en des démons s'ils sont vraiment "fire-and-forget".

Tenir un verrou trop longtemps. Verrouiller un grand bloc de code annule l'intérêt de la concurrence. Gardez les sections verrouillées aussi courtes que possible — protégez uniquement l'opération read-modify-write.

Interblocage. Un interblocage survient quand deux threads détiennent chacun un verrou attendu par l'autre. Prévenez-le en acquérant toujours plusieurs verrous dans le même ordre dans tous les threads.

import threading

lock_a = threading.Lock()
lock_b = threading.Lock()

# DEADLOCK: Thread 1 holds lock_a, waits for lock_b
#           Thread 2 holds lock_b, waits for lock_a

# FIX: always acquire locks in the same order (lock_a then lock_b) in every thread

Modifier une liste pendant qu'un autre thread l'itère. Encapsulez tous les accès (lecture et écriture) aux collections partagées avec un verrou pour éviter RuntimeError: list changed size during iteration.

Résumé de référence rapide

Pratique