TopolearnChannels let threads cooperate by passing ownership instead of sharing memory, which is often simpler and safer.
A channel moves values from producers to a consumer. std::sync::mpsc::channel() returns a (Sender<T>, Receiver<T>) pair; mpsc stands for multiple producer, single consumer. Clone the Sender to give one to each producer thread.
Sending moves the value into the channel, so the sender cannot touch it afterwards: ownership transfers to the receiving thread and no locking of the data is needed.
Receiving:
rx.recv()blocks until a message arrives, returningErronce all senders are dropped and the channel is empty.for msg in rx(orrx.iter()) loops until that happens, which gives you clean shutdown for free: when producers finish and drop their senders, the consumer's loop ends.try_recv()never blocks;recv_timeout(d)waits up to a duration.
The classic bug: keeping the original tx alive in the main thread after cloning it for workers. The consumer then waits forever, because one sender still exists. drop(tx) once you have handed out the clones.
channel() is unbounded: a fast producer can fill memory. `sync_channel(n)` is bounded; send blocks when n messages are queued, applying backpressure. sync_channel(0) is a rendezvous channel where each send waits for a matching receive.
To send different kinds of messages, send an enum. To get a reply, include a Sender for the answer inside the message, which gives you a simple actor pattern.
For multiple consumers or select over several channels, use the crossbeam-channel crate; for async code, use the channels in tokio::sync (mpsc, oneshot, broadcast, watch).
Example
use std::{sync::mpsc, thread};
enum Msg { Line(String), Done(usize) }
fn main() {
let (tx, rx) = mpsc::sync_channel::<Msg>(8); // bounded: backpressure at 8 queued
for id in 0..3 {
let tx = tx.clone();
thread::spawn(move || {
for n in 0..2 { tx.send(Msg::Line(format!("worker {id} item {n}"))).unwrap(); }
tx.send(Msg::Done(id)).unwrap();
}); // this clone of tx is dropped when the thread ends
}
drop(tx); // drop the original, or the loop below never ends
let (mut lines, mut finished) = (Vec::new(), Vec::new());
for msg in rx { // ends when every sender is gone
match msg {
Msg::Line(text) => lines.push(text),
Msg::Done(id) => finished.push(id),
}
}
finished.sort();
println!("{} lines, done: {finished:?}", lines.len()); // 6 lines, done: [0, 1, 2]
}Remember
- Sending moves ownership to the receiver; no locks on the data
- The receiver's loop ends when all
Senders are dropped - Use
sync_channel(n)for backpressure; enums for several message kinds
Watch out for
- Forgetting to drop the original
Sender, so the receiving loop hangs forever.