Channels in Rust: Nachrichten zwischen Threads senden

Auf der Seite Nebenläufigkeit haben sich Threads Daten geteilt, indem sie sie mit Arc<Mutex<T>> gesperrt haben. Das funktioniert. Es gibt aber noch einen anderen Weg, der oft einfacher ist: Statt Daten zu teilen, schicken sich Threads Nachrichten.

Stell dir einen Channel (Kanal) wie ein Rohr mit zwei Enden vor. Ein Thread wirft Werte in das Ende des Senders. Ein anderer Thread holt sie am Ende des Empfängers wieder heraus, in derselben Reihenfolge. Niemand teilt etwas: Jeder Wert wandert von einem Thread zum anderen. Aus der Go-Community stammt ein Spruch, den Rust sich hier ausleiht: „Kommuniziere nicht, indem du Speicher teilst – teile Speicher, indem du kommunizierst.“

Dein erster Channel

Der Channel der Standardbibliothek liegt in std::sync::mpsc. Der Name steht für multiple producer, single consumer (mehrere Erzeuger, ein Verbraucher): viele Sender, ein Empfänger.

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

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

    thread::spawn(move || {
        let message = String::from("Hallo vom Worker");
        tx.send(message).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("main hat bekommen: {received}");
}

Ausgabe:

main hat bekommen: Hallo vom Worker

▶ Selbst ausprobieren

Schritt für Schritt:

• mpsc::channel() gibt ein Paar zurück: einen Sender und einen Receiver. Traditionell heißen sie tx (transmit, senden) und rx (receive, empfangen).
• move übergibt tx an den neuen Thread – genauso, wie du auf der Seite Nebenläufigkeit Daten in Threads verschoben hast.
• tx.send(value) legt einen Wert in den Channel. Die Methode gibt ein Result zurück, denn sie schlägt fehl, wenn der Empfänger schon weg ist.
• rx.recv() wartet, bis ein Wert ankommt. Ein Err gibt es nur, wenn alle Sender weg sind und der Channel leer ist.

Wir haben hier kein join() gebraucht. recv() wartet ja schon auf die Nachricht, also kann main nicht zu früh enden.

Viele Werte empfangen

Ein Receiver funktioniert als Iterator. Eine for-Schleife darüber ruft für dich immer wieder recv() auf. Und die Schleife endet von selbst, sobald alle Sender gedroppt (freigegeben) wurden:

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

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

    thread::spawn(move || {
        for i in 1..=3 {
            tx.send(i * 10).unwrap();
        }
        // tx wird hier gedroppt, wenn der Thread endet
    });

    for value in rx {
        println!("erhalten: {value}");
    }
    println!("Channel geschlossen, alles erledigt");
}

Ausgabe:

erhalten: 10
erhalten: 20
erhalten: 30
Channel geschlossen, alles erledigt

▶ Selbst ausprobieren

Hier gibt es nur einen Sender, also kommen die Werte genau in der Reihenfolge an, in der sie gesendet wurden. Die Regel „die Schleife hört auf, wenn die Sender weg sind“ ist wichtig. Merk sie dir für den nächsten Abschnitt.

Mehrere Erzeuger: den Sender klonen

Damit mehrere Threads senden können, gibst du jedem mit tx.clone() eine eigene Kopie des Senders. Alle Kopien speisen denselben Channel:

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

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

    for id in 1..=3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(format!("Worker {id} meldet sich")).unwrap();
        });
    }
    drop(tx); // das Original droppen, sonst endet die Schleife unten nie

    let mut messages: Vec<String> = rx.iter().collect();
    messages.sort();
    for m in messages {
        println!("{m}");
    }
}

Ausgabe:

Worker 1 meldet sich
Worker 2 meldet sich
Worker 3 meldet sich

▶ Selbst ausprobieren

Zwei Dinge fallen auf:

• drop(tx) ist wichtig. Jeder Thread droppt seinen Klon, wenn er fertig ist, aber main hält noch das Original. Solange es noch einen Sender gibt, wartet der Empfänger weiter – und das Programm würde für immer hängen. Mit drop sagst du: „main sendet nichts mehr.“
• Die Threads werden in beliebiger Reihenfolge fertig. Die Nachrichten können als Worker 2, Worker 1, Worker 3 ankommen. Wir sammeln sie und sortieren sie, damit die Ausgabe immer gleich ist. Nimm das sort() heraus und führ das Programm ein paarmal aus, dann siehst du, wie sich die Reihenfolge ändert.

Nicht ewig warten: try_recv und recv_timeout

recv() blockiert, bis etwas ankommt. Manchmal willst du aber nicht warten. Dabei helfen zwei Verwandte:

• try_recv() kehrt sofort zurück: mit einem Wert, mit Empty (noch nichts da) oder mit Disconnected (alle Sender sind weg).
• recv_timeout(duration) wartet, aber nur bis zu einer Zeitgrenze. Kommt bis dahin nichts an, bekommst du Timeout.

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

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

    // Noch nichts gesendet: try_recv kehrt sofort zurück.
    println!("{:?}", rx.try_recv());

    thread::spawn(move || tx.send(7).unwrap()).join().unwrap();

    // Ein Wert wartet schon, also kommt er ohne Wartezeit zurück.
    println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));

    // Der Sender ist weg, und der Channel ist leer.
    println!("{:?}", rx.try_recv());
}

Ausgabe:

Err(Empty)
Ok(7)
Err(Disconnected)

▶ Selbst ausprobieren

Wir rufen join() für den Thread auf, bevor wir empfangen. So wartet der Wert schon, und die Ausgabe ist immer dieselbe.

try_recv() ist praktisch in einer Schleife, die noch andere Arbeit zu erledigen hat – zum Beispiel in einer Spielschleife, die einmal pro Frame nach neuen Nachrichten schaut. recv_timeout() passt gut zu „gib auf, wenn der Worker zu lange braucht“.

Begrenzte Channels mit sync_channel

Ein normaler channel() hat keine Größenbegrenzung. Ist der Sender viel schneller als der Empfänger, stapeln sich die Nachrichten im Speicher. mpsc::sync_channel(n) erzeugt einen Channel, der höchstens n Nachrichten aufnimmt. Ist er voll, wartet send(), bis der Empfänger eine herausholt. Das nennt man Backpressure (Gegendruck): Ein langsamer Verbraucher bremst einen schnellen Erzeuger automatisch aus.

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

fn main() {
    // Platz für nur 2 Nachrichten gleichzeitig.
    let (tx, rx) = mpsc::sync_channel(2);

    let producer = thread::spawn(move || {
        for i in 1..=5 {
            tx.send(i).unwrap(); // blockiert, solange der Puffer voll ist
        }
    });

    let total: i32 = rx.iter().sum();
    producer.join().unwrap();
    println!("Summe = {total}");
}

Ausgabe:

Summe = 15

▶ Selbst ausprobieren

Der Erzeuger kann main nie um mehr als zwei Zahlen voraus sein. Mit sync_channel(0) gibt es gar keinen Puffer: Jedes send() wartet, bis der Empfänger bereit ist, den Wert anzunehmen – wie bei einem Paket, das du persönlich übergibst.

Ein kleiner Worker-Pool

Jetzt setzen wir alles zusammen. Ein Worker-Pool besteht aus ein paar Threads, die Jobs aus einer gemeinsamen Warteschlange holen und Ergebnisse zurückschicken. Er nutzt zwei Channels: einen für die Jobs und einen für die Ergebnisse.

Es gibt einen Haken: Einen Receiver kannst du nicht klonen (es gibt ja nur einen Verbraucher). Deshalb teilen sich die Worker den Job-Empfänger hinter Arc<Mutex<_>> und holen abwechselnd Jobs heraus:

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

fn main() {
    let (job_tx, job_rx) = mpsc::channel::<u64>();
    let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();

    // Receiver lassen sich nicht klonen, also teilen sich die Worker einen hinter Arc<Mutex<_>>.
    let job_rx = Arc::new(Mutex::new(job_rx));

    let mut workers = Vec::new();
    for _ in 0..3 {
        let job_rx = Arc::clone(&job_rx);
        let result_tx = result_tx.clone();
        workers.push(thread::spawn(move || loop {
            let job = job_rx.lock().unwrap().recv();
            match job {
                Ok(n) => result_tx.send((n, n * n)).unwrap(),
                Err(_) => break, // keine Jobs mehr
            }
        }));
    }
    drop(result_tx);

    for n in 1..=6 {
        job_tx.send(n).unwrap();
    }
    drop(job_tx); // sagt den Workern, dass keine Arbeit mehr kommt

    let mut results: Vec<(u64, u64)> = result_rx.iter().collect();
    for w in workers {
        w.join().unwrap();
    }
    results.sort();
    for (n, square) in results {
        println!("{n} zum Quadrat ist {square}");
    }
}

Ausgabe:

1 zum Quadrat ist 1
2 zum Quadrat ist 4
3 zum Quadrat ist 9
4 zum Quadrat ist 16
5 zum Quadrat ist 25
6 zum Quadrat ist 36

▶ Selbst ausprobieren

So fährt der Pool sauber herunter:

1. drop(job_tx) schließt den Job-Channel. Das recv() jedes Workers gibt dann Err zurück, und der Worker verlässt seine Schleife mit break.
2. Wenn ein Worker endet, wird sein Klon von result_tx gedroppt. Das Original haben wir schon gedroppt. Sobald also alle drei Worker fertig sind, schließt sich der Ergebnis-Channel.
3. Dann hört result_rx.iter() auf, und wir sortieren die Ergebnisse, weil die Worker in beliebiger Reihenfolge fertig werden.

Ein kleines Detail: job_rx.lock().unwrap().recv() steht mit Absicht in einer einzigen let-Anweisung. Die Sperre wird am Ende dieser Anweisung freigegeben. So können sich die anderen Worker schon den nächsten Job holen, während dieser hier noch seine Zahl quadriert.

Channels oder Arc<Mutex<T>>?

Beides ist sicher. Die beiden passen nur zu unterschiedlichen Arten von Problemen:

Nimm einen Channel, wenn …	Nimm Arc<Mutex<T>>, wenn …
die Arbeit in eine Richtung fließt: Erzeuger stellen etwas her, ein Verbraucher kümmert sich darum	viele Threads denselben Wert lesen und ändern müssen (einen Zähler, einen Cache)
du eine Warteschlange für Jobs, Ereignisse oder Logzeilen willst	die Daten groß sind und Kopieren oder Verschieben Verschwendung wäre
jeder Wert immer nur einen Besitzer hat	Threads den aktuellen Zustand sehen müssen, keinen Strom von Änderungen

Eine gute Faustregel: Wenn du dein Programm mit „diese Threads sagen jenem Thread etwas“ beschreiben kannst, nimm einen Channel. Heißt es eher „diese Threads bearbeiten alle dieses eine Ding“, nimm einen Mutex. Viele echte Programme nutzen beides – wie unser Worker-Pool.

Häufige Fehler

Einen Wert nach dem Senden weiterbenutzen. send() übernimmt die Ownership (Eigentum), genau wie wenn du einen Wert an eine Funktion übergibst. Nach dem Senden gehört der Wert dem Empfänger:

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

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

    thread::spawn(move || {
        let message = String::from("hallo");
        tx.send(message).unwrap();
        println!("gesendet: {message}");
    });

    println!("{}", rx.recv().unwrap());
}

error[E0382]: borrow of moved value: `message`
  --> src/main.rs:10:30
   |
 8 |         let message = String::from("hallo");
   |             ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(message).unwrap();
   |                 ------- value moved here
10 |         println!("gesendet: {message}");
   |                              ^^^^^^^ value borrowed here after move

Der Compiler sagt: message wurde verschoben („moved“) und wird danach trotzdem noch ausgeliehen. Damit schützt dich Rust: Könnte der Sender message noch verändern, würden beide Threads auf dieselben Daten zugreifen. Gib den Wert vor dem Senden aus, oder sende ein .clone().

Den Sender in eine Schleife von Threads verschieben. Der erste Thread nimmt sich tx, und für den zweiten bleibt nichts übrig:

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

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

    for id in 1..=3 {
        thread::spawn(move || {
            tx.send(id).unwrap();
        });
    }

    for value in rx {
        println!("{value}");
    }
}

error[E0382]: use of moved value: `tx`
  --> src/main.rs:8:23
   |
 5 |     let (tx, rx) = mpsc::channel();
   |          -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait
...
 8 |         thread::spawn(move || {
   |                       ^^^^^^^ value moved into closure here, in previous iteration of loop

Die Meldung sagt: tx wurde schon im vorherigen Schleifendurchlauf in das Closure verschoben. Die Lösung ist let tx = tx.clone(); ganz oben in der Schleife, wie im Beispiel mit mehreren Erzeugern.

Ein Programm, das nie endet. Hier gibt es keinen Compilerfehler, und genau das macht diesen Fehler so tückisch. Wenn du über einen Empfänger iterierst (for x in rx), während noch irgendein Sender lebt – oft das ursprüngliche tx in main –, wartet die Schleife für immer. Wenn dein Programm „einfach hängt“, prüf, ob wirklich jeder Sender gedroppt wird.

Wie es weitergeht

std::sync::mpsc deckt schon viel ab, aber dir werden zwei beliebte Alternativen begegnen:

• crossbeam-channel (auf Englisch) bietet Channels mit mehreren Verbrauchern (Empfänger, die du klonen kannst) und ein select!-Makro, mit dem du auf mehrere Channels gleichzeitig wartest.
• tokio::sync::mpsc ist der Channel für Async-Code. Wenn du async/.await benutzt (siehe Async/Await), nimm ihn statt des Channels aus der Standardbibliothek, denn sein recv().await blockiert nicht den ganzen Thread.

Zusammenfassung

• mpsc::channel() gibt dir ein Paar (Sender, Receiver). send() schiebt einen Wert hinein, recv() wartet auf den nächsten.
• Klon den Sender, um mehrere Erzeuger zu haben. Einen Empfänger gibt es immer nur einmal.
• Eine Schleife über einen Empfänger endet, wenn alle Sender gedroppt sind. Denk also daran, das Original mit drop() freizugeben.
• Mit try_recv() und recv_timeout() kannst du nachschauen, ohne ewig zu warten. sync_channel(n) begrenzt die Warteschlange auf n Nachrichten.
• Nachrichten von mehreren Threads kommen in beliebiger Reihenfolge an: Sortiere oder summiere sie, wenn du eine stabile Ausgabe brauchst.
• Nimm Channels, wenn Daten von Thread zu Thread fließen. Nimm Arc<Mutex<T>>, wenn Threads sich einen Wert teilen und ihn bearbeiten.

Kurzer Check

Kurzer Check

1. Wofür steht „mpsc“?

Many processes, shared cache (viele Prozesse, gemeinsamer Cache) Message passing, safe concurrency (Nachrichtenaustausch, sichere Nebenläufigkeit) Multiple producer, single consumer (mehrere Erzeuger, ein Verbraucher) Mutex protected shared channel (durch Mutex geschützter gemeinsamer Channel)

2. Wie lässt du drei Threads in denselben Channel senden?

Du gibst jedem Thread ein tx.clone() Du packst den Empfänger in ein Rc Du rufst dreimal mpsc::channel() auf und führst die Channels zusammen Gar nicht – ein Channel hat nur einen Sender

3. Eine Schleife for value in rx endet nie. Was ist die wahrscheinlichste Ursache?

Der Channel ist voll Ein Sender (oft das ursprüngliche tx in main) wurde nie gedroppt recv() wurde zu früh aufgerufen Die Werte waren nicht Copy

4. Was macht send(), wenn ein sync_channel(2) schon 2 Nachrichten enthält?

Es verwirft die älteste Nachricht Es gibt sofort einen Fehler zurück Es löst eine Panic aus Es wartet, bis der Empfänger eine Nachricht herausholt

Punkte: 0 / 4

Weiterlesen

• Nachrichtenaustausch zwischen Threads – Das Rust-Buch: die offizielle Einführung in Channels, mit weiteren Beispielen.
• std::sync::mpsc – Doku der Standardbibliothek (auf Englisch): alle Methoden von Sender, SyncSender und Receiver.
• Channels – Rust by Example (auf Englisch): ein kurzes, ausführbares Beispiel mit mehreren Erzeuger-Threads.
• Von einem single-threaded zu einem multi-threaded Webserver – Das Rust-Buch: baut Schritt für Schritt einen kompletten Thread-Pool mit Channels und Arc<Mutex<Receiver>>.
• tokio::sync::mpsc – docs.rs (auf Englisch): der Async-Channel, den du mit Tokio benutzt, und wie er sich vom Channel der Standardbibliothek unterscheidet.