Sichere Concurrency in Rust mit Mutex und Channel

1. Überblick

Rust nennt diese Fähigkeit „fearless concurrency”. Der Compiler sperrt Data Races bereits zur Build-Zeit, bevor das Programm läuft. Dieser Artikel stellt zwei Hauptwerkzeuge der Standardbibliothek vor: Mutex (zusammen mit Arc) zum Teilen von State und mpsc-Channel zum Austausch von Nachrichten zwischen Threads. Nach dem Lesen wirst du wissen, wie du jede Variante einsetzt, Tests dafür schreibst und häufige Fallstricke wie Deadlocks oder zu langes Lock-Halten vermeidest.

Alle Beispiele verwenden nur die Standardbibliothek, es gibt keine externen Abhängigkeiten. Du kannst ein Projekt mit cargo new orders-demo erstellen und dann den Code in src/main.rs einfügen.

2. Das Problem: Daten zwischen Threads teilen

Angenommen, du hast mehrere Threads, die zusammen Bestellungen verarbeiten und du möchtest die Gesamtzahl zählen. Eine naive Implementierung würde so aussehen:

use std::thread;

fn main() {
    let mut total_orders = 0;

    let handles: Vec<_> = (0..4)
        .map(|_| thread::spawn(|| total_orders += 1))
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Dieses Programm kompiliert nicht:

error[E0373]: closure may outlive the current function, but it borrows `total_orders`, which is owned by the current function

Der Compiler gibt zwei Gründe an:

• thread::spawn erfordert, dass die Closure 'static ist. Der Child-Thread kann länger leben als main, also darf er sich keine lokalen Variablen leihen.
• Auch wenn du move hinzufügst, erhält jeder Thread seine eigene Kopie von total_orders. Niemand addiert wirklich zu einer gemeinsamen Variable auf.

Um das zu beheben, brauchst du zwei Dinge: Ownership teilen mit Arc und Zugriff synchronisieren mit Mutex.

3. State mit Arc<Mutex<T>> teilen

3.1. Bestellungen von mehreren Threads zählen

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let total_orders = Arc::new(Mutex::new(0u32));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&total_orders);
        handles.push(thread::spawn(move || {
            for _ in 0..1_000 {
                *counter.lock().unwrap() += 1;
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }

    println!("Gesamtbestellungen: {}", *total_orders.lock().unwrap());
}

Gesamtbestellungen: 4000

Erklärung zu den einzelnen Teilen:

• Arc::new(Mutex::new(0u32)): Arc (Atomic Reference Counted) ermöglicht mehreren Threads, einen Wert zu besitzen. Mutex stellt sicher, dass zu jedem Zeitpunkt nur ein Thread auf den inneren Wert zugreifen kann.
• Arc::clone(&total_orders) erhöht nur den Referenzzähler, kopiert keine Daten. Jeder Thread erhält sein eigenes Arc über move.
• counter.lock() blockiert den aktuellen Thread, bis der Lock erworben ist. Die Funktion gibt Result<MutexGuard<u32>, PoisonError<…>> zurück.
• MutexGuard dereferenziert zu &mut u32, also modifiziert *... += 1 den Wert direkt. Der Guard wird am Ende der Anweisung gedroppt, und der Lock wird automatisch freigegeben.

Du kannst den Lock nicht „vergessen” in Rust. Du kannst auch nicht den Wert lesen, ohne den Lock zu halten, weil die Daten innerhalb von Mutex liegen.

Hinweis: Du kannst Arc nicht durch Rc ersetzen. Rc implementiert nicht Send, und der Compiler wird einen Fehler werfen, wenn du es in thread::spawn einbringst. Ähnlich kannst du Mutex nicht durch RefCell ersetzen, da RefCell nicht Sync ist.

3.2. Scope des MutexGuard

Der Lock wird gehalten, bis der Guard gedroppt wird. Wenn du den Guard einer Variable zuweist, bleibt der Lock für den gesamten Scope dieser Variable bestehen:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

fn reserve(inventory: &Mutex<HashMap<String, u32>>, sku: &str) -> bool {
    let reserved = {
        let mut stock = inventory.lock().unwrap();
        match stock.get_mut(sku) {
            Some(qty) if *qty > 0 => {
                *qty -= 1;
                true
            }
            _ => false,
        }
    }; // Guard wird hier gedroppt, Lock wird freigegeben

    if reserved {
        // Simuliert langsamen PaymentService-Aufruf, hält NICHT den Lock
        thread::sleep(Duration::from_millis(50));
    }
    reserved
}

fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("AO-THUN".to_string(), 2)])));

    let handles: Vec<_> = (0..3)
        .map(|i| {
            let inv = Arc::clone(&inventory);
            thread::spawn(move || println!("Kunde {i}: {}", reserve(&inv, "AO-THUN")))
        })
        .collect();

    for h in handles {
        h.join().unwrap();
    }
}

Kunde 0: true
Kunde 1: true
Kunde 2: false

Die Reihenfolge der Zeilen kann bei verschiedenen Läufen unterschiedlich sein. Das Ergebnis ist immer genau zwei true und ein false.

Der Block { ... } begrenzt die Lock-Dauer auf ein Minimum. Der langsame Teil (Zahlungsaufruf) läuft nach der Lock-Freigabe, sodass andere Threads nicht warten müssen. Wenn du den Lock früher freigeben musst, kannst du drop(stock) explizit aufrufen.

3.3. Mutex-Vergiftung

Wenn ein Thread paniert, während er einen Lock hält, markiert Rust das Mutex als vergiftet (poisoned). Der Grund ist, dass die Daten möglicherweise in einem inkonsistenten Zustand sind:

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let balance = Arc::new(Mutex::new(100i64));

    let b = Arc::clone(&balance);
    let _ = thread::spawn(move || {
        let mut guard = b.lock().unwrap();
        *guard -= 30;
        panic!("PaymentService-Fehler mitten im Prozess");
    })
    .join();

    match balance.lock() {
        Ok(v) => println!("Kontostand: {}", *v),
        Err(poisoned) => {
            let v = poisoned.into_inner();
            println!("Mutex vergiftet, aktueller Kontostand: {}", *v);
        }
    }
}

thread '<unnamed>' panicked at src/main.rs:10:9:
PaymentService-Fehler mitten im Prozess
Mutex vergiftet, aktueller Kontostand: 70

Daher bedeutet .unwrap() nach lock() tatsächlich „wenn ein anderer Thread paniert hat, während er den Lock hielt, dann panic auch”. Für die meisten Anwendungen ist das eine vernünftige Wahl. Wenn du weißt, wie man die Daten wiederherstellt, gibt into_inner() den Guard zurück, damit du überprüfen und reparieren kannst.

4. Nachrichten mit Channels versenden

Mutex beantwortet die Frage „mehrere Threads ändern dieselben Daten”. Channels gehen in die umgekehrte Richtung: jeder Datenwert hat zu einem Zeitpunkt nur einen Besitzer, und das Ownership wird über den Channel übertragen. Bei vielen Problemen ist diese Herangehensweise einfacher und weniger fehleranfällig.

4.1. Mehrere Producer, ein Consumer

std::sync::mpsc ist die Abkürzung für multiple producer, single consumer:

use std::sync::mpsc;
use std::thread;

#[derive(Debug)]
struct Order {
    id: u32,
    customer: String,
    amount: u64,
}

fn main() {
    let (tx, rx) = mpsc::channel::<Order>();

    for shop in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            let order = Order {
                id: shop * 100,
                customer: format!("kunde-{shop}"),
                amount: 150_000 * shop as u64,
            };
            tx.send(order).unwrap();
        });
    }
    drop(tx); // wichtig: original sender freigeben

    let mut revenue = 0;
    for order in rx {
        println!("Bestellung #{} von {} erhalten", order.id, order.customer);
        revenue += order.amount;
    }
    println!("Umsatz: {revenue}");
}

Bestellung #100 von kunde-1 erhalten
Bestellung #300 von kunde-3 erhalten
Bestellung #200 von kunde-2 erhalten
Umsatz: 900000

Erklärung:

• mpsc::channel() gibt ein Paar (Sender, Receiver) zurück. Jeder Producer erhält einen tx.clone().
• tx.send(order) moved order in den Channel. Nach dieser Zeile kann der sendende Thread order nicht mehr verwenden, und der Compiler meldet einen Fehler, wenn du es versuchst. Dadurch können zwei Threads niemals denselben Order ändern.
• for order in rx schleift, bis alle Sender gedroppt wurden. Wenn die drop(tx)-Zeile fehlt, bleibt der ursprüngliche Sender in main alive, und die Schleife wartet ewig.

Die Reihenfolge des Empfangs hängt von der Thread-Planung ab. Der Gesamtumsatz ist immer 900000.

4.2. Begrenzte Channels und Backpressure

mpsc::channel() hat einen unbegrenzten Buffer. Wenn der Producer schneller als der Consumer ist, wächst der Speicher. sync_channel(n) begrenzt den Buffer auf n Elemente, und send blockiert, wenn der Buffer voll ist:

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::sync_channel::<u32>(2);

    let producer = thread::spawn(move || {
        for id in 1..=5 {
            tx.send(id).unwrap();
            println!("Bestellung {id} gesendet");
        }
    });

    thread::sleep(Duration::from_millis(100));
    for id in rx {
        println!("Bestellung {id} wird verarbeitet");
        thread::sleep(Duration::from_millis(20));
    }
    producer.join().unwrap();
}

Bestellung 1 gesendet
Bestellung 2 gesendet
Bestellung 1 wird verarbeitet
Bestellung 3 gesendet
Bestellung 2 wird verarbeitet
...

Der Producer kann zwei Bestellungen senden, muss dann aber warten, dass der Consumer welche herausnimmt. Dies ist ein natürlicher Backpressure-Mechanismus, sehr nützlich für Data-Processing-Pipelines. Bei sync_channel(0) muss sich jeder send blockiert, bis jemand recv aufruft (Rendezvous).

5. Wann du Arc nicht brauchst: Scoped Threads

Wenn Threads sicher vor dem Rückgabewert der Funktion beendet werden, ermöglicht thread::scope dir, Daten direkt zu leihen, ohne Arc zu benötigen:

use std::sync::Mutex;
use std::thread;

fn main() {
    let processed = Mutex::new(Vec::new());
    let orders = vec![101, 102, 103, 104];

    thread::scope(|s| {
        for chunk in orders.chunks(2) {
            s.spawn(|| processed.lock().unwrap().extend_from_slice(chunk));
        }
    }); // alle Threads im Scope sind hier gejoined

    let mut result = processed.into_inner().unwrap();
    result.sort();
    println!("{result:?}");
}

[101, 102, 103, 104]

Dank scope weiß der Compiler, dass die Threads nicht länger leben als processed und orders. Daher ist das Leihen mit & legal. Nach dem Scope nimmt into_inner() die Daten aus Mutex heraus, ohne zu sperren.

6. Tests für nebenläufigen Code schreiben

Du solltest die Logik in reine Funktionen aufteilen, um sie leicht zu testen. Füge das Folgende in src/lib.rs ein:

use std::sync::{mpsc, Arc, Mutex};
use std::thread;

pub fn count_orders(workers: u32, per_worker: u32) -> u32 {
    let total = Arc::new(Mutex::new(0));
    let handles: Vec<_> = (0..workers)
        .map(|_| {
            let total = Arc::clone(&total);
            thread::spawn(move || {
                for _ in 0..per_worker {
                    *total.lock().unwrap() += 1;
                }
            })
        })
        .collect();
    handles.into_iter().for_each(|h| h.join().unwrap());
    let result = *total.lock().unwrap();
    result
}

pub fn sum_payments(payments: Vec<u64>) -> u64 {
    let (tx, rx) = mpsc::channel();
    for amount in payments {
        let tx = tx.clone();
        thread::spawn(move || tx.send(amount).unwrap());
    }
    drop(tx);
    rx.iter().sum()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn counts_every_order_exactly_once() {
        assert_eq!(count_orders(8, 10_000), 80_000);
    }

    #[test]
    fn sums_all_payments_from_channel() {
        assert_eq!(sum_payments(vec![50_000, 120_000, 30_000]), 200_000);
    }

    #[test]
    fn empty_payments_returns_zero() {
        assert_eq!(sum_payments(vec![]), 0);
    }
}

cargo test

running 3 tests
test tests::empty_payments_returns_zero ... ok
test tests::sums_all_payments_from_channel ... ok
test tests::counts_every_order_exactly_once ... ok

test result: ok. 3 passed; 0 failed

Der Test empty_payments_returns_zero prüft einen Grenzfall: Wenn du drop(tx) vergisst, wird dieser Test für immer blockieren, statt 0 zurückzugeben. Das ist ein klares Zeichen, um Fehler zu erkennen.

Es gibt ein kleines Detail in count_orders: Der Wert wird einer result-Variable zugewiesen, bevor er zurückgegeben wird. Der Grund ist, dass der temporäre Guard im letzten Ausdruck der Funktion länger leben könnte als total, und der Compiler gibt einen Borrowing-Fehler aus.

7. Häufige Fehler

Deadlock durch umgekehrte Lock-Reihenfolge. Thread A sperrt orders, dann customers, während Thread B customers sperrt, dann orders. Beide warten ewig aufeinander. Rust verhindert nur Data Races, verhindert aber nicht Deadlocks. Du solltest eine feste Lock-Reihenfolge vereinbaren oder beide Daten in einen gemeinsamen Mutex<(A, B)> zusammenfassen.

Temporäre Guards leben länger als du denkst. Schaue dir diesen Code an:

while let Some(order) = queue.lock().unwrap().pop() {
    process(order); // Lock wird hier immer noch gehalten!
}

Temporaries in while let- (und match-) Ausdrücken leben bis zum Ende des Loop-Body. Jeder andere Thread, der ein Element in queue pushen möchte, wird blockiert, während process läuft. Die Lösung ist, sie zu trennen: let next = queue.lock().unwrap().pop(); dann verarbeiten.

lock() erneut im selben Thread aufrufen. std::sync::Mutex ist nicht reentrant. Zweimaliger Aufruf von lock() auf demselben Thread verursacht einen Deadlock oder Panic.

Lock halten, während auf I/O oder recv() gewartet wird. Jede blockierende Operation während eines Guards macht Mutex zum Engpass. Du solltest die benötigten Daten kopieren oder klonen, den Lock freigeben und dann mit langsamer Arbeit fortfahren.

Sender nicht gedroppt. Wie oben gezeigt, wartet der Consumer ewig.

Hinweis: Im async-Code (Tokio) halte keinen std::sync::MutexGuard über .await hinweg. Verwende stattdessen tokio::sync::Mutex oder Tokios Channels. Außerdem ist AtomicU32 für einfache Ganzzahlen wie Zähler viel schneller als Mutex<u32>.

Wann wählst du was?

• Verwende Mutex, wenn mehrere Threads langfristig denselben State lesen/ändern müssen, z. B. einen Cache oder Lagerbestand. Wenn Lesen deutlich häufiger als Schreiben ist, erwäge RwLock.
• Verwende Channels, wenn Daten von einem Ort zu einem anderen fließen, z. B. Bestellungswarteschlangen, Pipelines oder Worker Pools. Channels stellen sicher, dass jeder Wert nur einen Besitzer hat, daher ist die Logik leichter zu verstehen.

8. Fazit

Rust verwandelt häufige Concurrency-Fehler in Compile-Time-Fehler. Du kannst Daten ohne Arc nicht teilen, Daten ohne Mutex nicht ändern, und einen Wert, der über einen Channel gesendet wurde, nicht wiederverwenden. Arc<Mutex<T>> passt zu gemeinsamen States, während mpsc-Channels zu Ownership-Transfers zwischen Threads passen. sync_channel fügt Backpressure hinzu, und thread::scope ermöglicht dir, Arc zu vermeiden, wenn die Thread-Lebensdauer klar ist. Was der Compiler nicht schützen kann, sind Deadlocks, Guards, die zu lange leben, und nicht-gedroppe Sender. Dies sind die Stellen, an denen du genau hinschauen und sorgfältig testen musst.