Python Threading

Das threading-Modul von Python ermöglicht es, mehrere Aufgaben im selben Prozess gleichzeitig auszuführen. Jede Aufgabe läuft in einem eigenen Thread — einer leichtgewichtigen Ausführungseinheit, die den Speicherbereich des Prozesses teilt. Threading ist das richtige Werkzeug, wenn Ihr Programm die meiste Zeit wartet (eine Datei liest, eine HTTP-Anfrage stellt, eine Datenbank abfragt) und Sie während dieser Wartezeit sinnvolle Arbeit erledigen möchten, anstatt zu blockieren.

Dieses Kapitel behandelt:

• Threads erstellen und starten mit threading.Thread
• Auf das Ende von Threads warten mit join
• Daemon-Threads und Hintergrundaufgaben
• Data Races verhindern mit Lock und with
• Threads koordinieren mit Event und Semaphore
• Thread-sichere Kommunikation mit queue.Queue
• Der ThreadPoolExecutor für verwaltete Thread-Pools
• Das Global Interpreter Lock (GIL) und warum Threads CPU-lastigen Code nicht beschleunigen
• Wann Threading gegenüber asyncio zu bevorzugen ist

Einen Thread erstellen und starten

Importieren Sie threading und erstellen Sie ein Thread-Objekt, wobei Sie die auszuführende Funktion als target übergeben. Rufen Sie .start() auf, um den Thread zu starten:

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

Wichtige Punkte:

• args ist ein Tupel von Positionsargumenten, die an target übergeben werden. Verwenden Sie kwargs für Schlüsselwortargumente.
• Ohne .join() kann der Haupt-Thread beendet werden, bevor der gestartete Thread abgeschlossen ist.
• .start() kehrt sofort zurück; der neue Thread läuft parallel.

Schlüsselwortargumente übergeben

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

Mehrere Threads gleichzeitig ausführen

Der eigentliche Vorteil von Threading besteht darin, mehrere Aufgaben parallel auszuführen. Starten Sie zuerst alle Threads und warten Sie dann auf alle:

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

Ohne Threads würde dies 3 Sekunden dauern (sequenziell). Mit drei Threads dauert es etwa 1 Sekunde, da sich die Wartezeiten überlappen.

Thread als Unterklasse

Für komplexere Logik können Sie threading.Thread ableiten und run() überschreiben. Speichern Sie Ergebnisse als Instanzattribute, damit der aufrufende Code sie nach join() lesen kann:

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

Daemon-Threads

Ein Daemon-Thread ist ein Hintergrund-Thread, den der Interpreter automatisch beendet, wenn alle Nicht-Daemon-Threads abgeschlossen sind. Markieren Sie einen Thread als Daemon, indem Sie daemon=True übergeben (oder t.daemon = True vor .start() setzen):

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

Verwenden Sie Daemon-Threads für Hintergrundüberwachung oder Logging-Aufgaben, die das Beenden des Programms nicht verhindern sollen. Verwenden Sie sie niemals für Aufgaben, die sauber abgeschlossen werden müssen (Datei-Schreibvorgänge, Datenbank-Commits) — sie werden ohne jegliches Aufräumen beendet.

Thread-Namen und Inspektion

Jeder Thread hat einen Namen. Sie können ihn explizit setzen oder Python automatisch einen zuweisen lassen. Verwenden Sie threading.current_thread(), um den laufenden Thread zu inspizieren, und threading.active_count(), um aktive Threads zu zählen:

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

Synchronisierung: Data Races verhindern

Threads teilen den Speicher des Prozesses. Wenn zwei Threads gleichzeitig dieselbe Variable lesen und schreiben, entsteht ein Data Race — nicht-deterministisches Verhalten, das schwer zu reproduzieren oder zu debuggen ist.

Das folgende Beispiel ohne Lock liefert einen unvorhersehbaren Endwert, da Inkrementierungen verschiedener Threads sich überlappen können:

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

Ein threading.Lock stellt sicher, dass nur ein Thread den geschützten Abschnitt gleichzeitig ausführt. Verwenden Sie ihn als Kontextmanager mit with, damit der Lock immer freigegeben wird, auch wenn eine Ausnahme ausgelöst wird:

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 (re-entranter Lock)

Wenn ein Thread denselben Lock zweimal erwerben muss (z. B. ruft eine Methode eine andere Methode auf, die ebenfalls den Lock erwirbt), verwenden Sie threading.RLock. Er erlaubt es demselben Thread, den Lock erneut zu erwerben, ohne einen Deadlock zu verursachen:

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

Threads koordinieren: Event und Semaphore

Event

threading.Event ist ein einfaches Signal. Ein Thread ruft .set() auf, um zu signalisieren; andere Threads rufen .wait() auf, um zu blockieren, bis das Signal eintrifft:

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

Verwenden Sie ein Event, um die Startreihenfolge zu koordinieren — beispielsweise um Worker-Threads zu verzögern, bis eine Datenbankverbindung hergestellt wurde.

Semaphore

Ein threading.Semaphore begrenzt die Anzahl der Threads, die gleichzeitig einen Abschnitt betreten können. Dies ist nützlich für die Ratenbegrenzung des Zugriffs auf eine gemeinsame Ressource wie einen Verbindungspool:

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

Thread-lokale Daten

threading.local() erstellt ein Objekt, das pro Thread separate Werte hält. Dies ist nützlich für Thread-spezifische Caches oder Datenbank-Cursor:

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

Das Lesen von local_data.user in einem Thread, in dem es noch nie gesetzt wurde, löst einen AttributeError aus, genau wie jeder andere Attributzugriff.

Thread-sichere Queues

Die Klasse queue.Queue (aus dem Standardbibliotheks-Modul queue, nicht asyncio) ist eine thread-sichere FIFO. Threads können Elemente mit put und get einfügen und entnehmen, ohne einen Lock — die gesamte Synchronisierung wird intern gehandhabt.

Das klassische Muster ist Producer-Consumer: Ein oder mehrere Producer-Threads erzeugen Arbeit, Consumer-Threads verarbeiten sie:

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 bietet auch task_done() und join() zum Verfolgen, wann alle in der Warteschlange befindlichen Elemente verarbeitet wurden, sowie queue.LifoQueue / queue.PriorityQueue für alternative Reihenfolgen.

ThreadPoolExecutor: verwaltete Thread-Pools

Das Erstellen eines neuen Thread-Objekts für jede Aufgabe ist verschwenderisch, wenn Sie viele kurzlebige Aufgaben haben. concurrent.futures.ThreadPoolExecutor verwaltet einen Pool wiederverwendbarer Worker-Threads und gibt Future-Objekte für jede übermittelte Aufgabe zurück:

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) ist eine kürzere Form, wenn Sie keine einzelnen Future-Objekte benötigen:

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 bewahrt die Eingabereihenfolge in der Ausgabe, anders als as_completed, das in der Abschlussreihenfolge liefert.

Das Global Interpreter Lock (GIL)

CPython (der Standard-Python-Interpreter) besitzt ein Global Interpreter Lock — einen Mutex, der nur einem Thread erlaubt, Python-Bytecode gleichzeitig auszuführen. Das bedeutet, dass Threads in CPython keinen Python-Code wirklich parallel auf mehreren CPU-Kernen ausführen können.

Die praktische Konsequenz:

• I/O-gebundene Aufgaben: Threads beschleunigen das Programm tatsächlich. Während ein Thread auf eine Netzwerkantwort wartet, wird der GIL freigegeben und ein anderer Thread läuft. Alle obigen Beispiele demonstrieren dieses Verhalten.
• CPU-gebundene Aufgaben: Threads beschleunigen die Ausführung nicht und können durch den Overhead des Kontextwechsels sogar leicht langsamer sein.

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)

Für echte CPU-Parallelität in Python verwenden Sie stattdessen multiprocessing oder concurrent.futures.ProcessPoolExecutor — jeder Prozess hat seinen eigenen GIL.

Threading vs. asyncio

Sowohl threading als auch asyncio machen I/O-gebundene Programme schneller, aber sie funktionieren unterschiedlich:

Faustregel: Wenn Sie eine Bibliothek verwenden, die eine async-kompatible Version hat (z. B. aiohttp statt requests), greifen Sie auf asyncio zurück. Wenn Sie auf synchrone Blocking-Bibliotheken angewiesen sind, verwenden Sie Threading. Für CPU-gebundene Arbeit verwenden Sie multiprocessing.

Häufige Fallstricke

Einen Thread zweimal starten. Das Aufrufen von .start() für dasselbe Thread-Objekt mehr als einmal löst einen RuntimeError aus. Erstellen Sie für jede Ausführung eine neue Thread-Instanz.

join vergessen. Ein Thread, der nicht gejoined wird, kann noch laufen, wenn das Programm endet. Joinen Sie immer Threads, deren Abschluss wichtig ist, oder machen Sie sie zu Daemons, wenn sie wirklich "fire-and-forget" sind.

Einen Lock zu lange halten. Das Sperren eines großen Codeblocks macht den Zweck der Nebenläufigkeit zunichte. Halten Sie gesperrte Abschnitte so kurz wie möglich — schützen Sie nur die eigentliche Lese-Modifiziere-Schreib-Operation.

Deadlock. Ein Deadlock tritt auf, wenn zwei Threads jeweils einen Lock halten, auf den der andere wartet. Verhindern Sie dies, indem Sie mehrere Locks immer in derselben Reihenfolge in allen Threads erwerben.

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

Eine Liste modifizieren, während ein anderer Thread darüber iteriert. Schützen Sie alle Zugriffe (Lesen und Schreiben) auf gemeinsam genutzte Sammlungen mit einem Lock, um RuntimeError: list changed size during iteration zu vermeiden.

Kurzübersicht

Übung