Message Passing with Channels
Go's concurrency motto captures an important idea: do not communicate by sharing memory; instead, share memory by communicating. Rust embraces this philosophy through channels. Instead of multiple threads reaching into the same mutable variable, they send values to each other — and ownership transfer guarantees there is never a data race.
std::sync::mpsc
Rust's standard library provides mpsc (multi-producer, single-consumer)
channels in std::sync::mpsc. The name describes the topology: any number of
threads can send messages, but only one thread receives them.
mpsc::channel() returns a (Sender<T>, Receiver<T>) tuple. The sender and
receiver are split so ownership can be moved to different threads.
use std::sync::mpsc;
use std::thread;
fn main() {
// channel() returns (Sender<T>, Receiver<T>)
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
tx.send(String::from("hello from the thread")).unwrap();
});
// recv() blocks until a message arrives
let received = rx.recv().unwrap();
println!("Got: {}", received);
}Got: hello from the thread
tx.send(value) moves the value into the channel. After the call, the sending thread no longer owns it — ownership has been transferred to the receiver. This is what makes the channel safe with no locks.Sending Multiple Messages
A sender can send as many messages as it likes. The receiver collects them in
order. The simplest way to drain all messages is a for loop — it ends
automatically when all Sender handles have been dropped (channel closed).
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let messages = vec!["one", "two", "three", "four", "five"];
for msg in messages {
tx.send(msg).unwrap();
thread::sleep(Duration::from_millis(50));
}
// tx is dropped here — channel closes
});
// for loop ends when the channel is closed
for received in rx {
println!("Got: {}", received);
}
}Got: one Got: two Got: three Got: four Got: five
Blocking vs Non-blocking Receive
The receiver has two methods for reading messages, with different blocking behaviour:
Method | Blocks? | Returns |
|---|---|---|
rx.recv() | Yes — waits until a message arrives | Ok(T) or Err(RecvError) if channel closed |
rx.try_recv() | No — returns immediately | Ok(T), Err(Empty), or Err(Disconnected) |
rx.recv_timeout(dur) | Yes — up to the given duration | Ok(T), Err(Timeout), or Err(Disconnected) |
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
thread::sleep(Duration::from_millis(200));
tx.send("delayed message").unwrap();
});
// try_recv returns immediately — message not ready yet
match rx.try_recv() {
Ok(msg) => println!("Got immediately: {}", msg),
Err(e) => println!("Not ready yet: {}", e),
}
// recv_timeout waits up to 500 ms
match rx.recv_timeout(Duration::from_millis(500)) {
Ok(msg) => println!("Got with timeout: {}", msg),
Err(e) => println!("Timed out: {}", e),
}
}Not ready yet: channel is empty Got with timeout: delayed message
Multiple Producers
The "multi-producer" part of mpsc means multiple senders can feed the same receiver. Clone the sender to get additional handles — each clone is an independent sender, and the channel closes only when all senders have been dropped.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
// Clone the sender for each additional producer
let tx1 = tx.clone();
let tx2 = tx.clone();
thread::spawn(move || tx.send("from thread 0").unwrap());
thread::spawn(move || tx1.send("from thread 1").unwrap());
thread::spawn(move || tx2.send("from thread 2").unwrap());
// Collect all three messages (order is not guaranteed)
for msg in rx.iter().take(3) {
println!("Received: {}", msg);
}
}Received: from thread 0 Received: from thread 2 Received: from thread 1
Channel Closure and Disconnection
The channel is closed when all Sender handles are dropped. After that,
rx.recv() returns Err(RecvError) and a for loop over rx ends cleanly.
This makes it natural to signal workers that there is no more work.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel::<i32>();
thread::spawn(move || {
for i in 0..3 {
tx.send(i).unwrap();
}
// tx dropped here — channel closes
});
// recv() on a closed, empty channel returns Err
loop {
match rx.recv() {
Ok(v) => println!("value: {}", v),
Err(_) => { println!("channel closed"); break; }
}
}
}value: 0 value: 1 value: 2 channel closed
Bounded Channels: sync_channel
mpsc::channel() is unbounded — the sender never blocks; the buffer grows as
needed. mpsc::sync_channel(n) creates a bounded channel with a buffer of
size n. If the buffer is full, send() blocks until the receiver consumes a
message.
Bounded channels provide back-pressure: a slow consumer naturally throttles a fast producer.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
// Buffer holds at most 2 messages
let (tx, rx) = mpsc::sync_channel(2);
thread::spawn(move || {
for i in 0..5 {
println!("Sending {}", i);
tx.send(i).unwrap(); // blocks when buffer is full
}
});
thread::sleep(Duration::from_millis(100)); // let producer run
for msg in rx {
println!("Received {}", msg);
thread::sleep(Duration::from_millis(50));
}
}Sending 0 Sending 1 Sending 2 Received 0 Sending 3 Received 1 Sending 4 Received 2 Received 3 Received 4
sync_channel(0) for a *rendezvous* channel — the sender blocks until the receiver is ready. This synchronises the two threads at every message.Channels vs Shared State
Both channels and Arc<Mutex<T>> can be used for thread communication. They
model different ownership strategies:
Channels (mpsc) | Arc+Mutex | |
|---|---|---|
Ownership model | Transfer — sender gives up the value | Shared — all threads see the same value |
Locking | None — ownership enforces safety | Mutex lock required for mutation |
Access pattern | Producer → consumer pipeline | Random read/write by any thread |
Back-pressure | Built-in with sync_channel | Manual (check size, sleep, etc.) |
Best for | Pipelines, work queues, event streams | Shared caches, counters, config |
Real Example: Producer-Consumer Work Queue
Here is a practical pattern: a main thread generates work items and sends them through a channel; a pool of worker threads receives and processes them.
use std::sync::mpsc;
use std::thread;
fn main() {
let num_workers = 4;
let (tx, rx) = mpsc::sync_channel::<u64>(num_workers * 2);
// Wrap receiver in Arc+Mutex so multiple workers can share it
let rx = std::sync::Arc::new(std::sync::Mutex::new(rx));
let mut handles = vec![];
for id in 0..num_workers {
let rx = std::sync::Arc::clone(&rx);
handles.push(thread::spawn(move || {
let mut processed = 0u64;
loop {
let job = {
let guard = rx.lock().unwrap();
guard.recv().ok()
};
match job {
Some(n) => {
// Simulate work: compute n squared
let _ = n * n;
processed += 1;
}
None => break, // channel closed, no more work
}
}
println!("Worker {} processed {} jobs", id, processed);
processed
}));
}
// Produce 20 jobs
for i in 0..20u64 {
tx.send(i).unwrap();
}
drop(tx); // close the channel — workers will exit when it empties
let total: u64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
println!("Total jobs processed: {}", total);
}Worker 0 processed 6 jobs Worker 1 processed 5 jobs Worker 2 processed 5 jobs Worker 3 processed 4 jobs Total jobs processed: 20
Sending Across Trait Objects
Channels can carry any type that implements Send. This includes boxed trait
objects — a useful pattern for heterogeneous message types.
use std::sync::mpsc;
use std::thread;
trait Task: Send {
fn run(&self);
}
struct PrintTask(String);
struct SumTask(Vec<i32>);
impl Task for PrintTask {
fn run(&self) { println!("PrintTask: {}", self.0); }
}
impl Task for SumTask {
fn run(&self) { println!("SumTask: {}", self.0.iter().sum::<i32>()); }
}
fn main() {
let (tx, rx) = mpsc::channel::<Box<dyn Task>>();
thread::spawn(move || {
tx.send(Box::new(PrintTask(String::from("hello")))).unwrap();
tx.send(Box::new(SumTask(vec![1, 2, 3, 4, 5]))).unwrap();
});
for task in rx {
task.run();
}
}PrintTask: hello SumTask: 15
Common Patterns at a Glance
Unbounded channel (
mpsc::channel) — use when the producer is faster than the consumer only briefly, and memory growth is acceptable.Bounded channel (
mpsc::sync_channel(n)) — use when you need back-pressure to prevent the producer from running too far ahead.Multiple producers (
tx.clone()) — fan-in: many threads produce, one thread collects.Worker pool (shared receiver behind
Arc<Mutex<Receiver>>) — fan-out: one producer, many consumers.Closing signal (drop all senders) — signal workers to stop without a separate flag.