Rust channels
last modified November 13, 2026
In this article we show how to pass messages between threads with channels in Rust.
A channel is a queue that connects threads. One side, the sender, puts values into the channel; the other side, the receiver, takes them out in the same order. Instead of sharing memory protected by locks, the threads communicate by sending data to each other. Ownership of each value moves with the message, so only one thread can use it at a time.
The standard library provides channels in the std::sync::mpsc
module. The name stands for multiple producer, single consumer:
a channel can have many senders but only one receiver. The
mpsc::channel function creates an unbounded channel,
mpsc::sync_channel a bounded one.
The examples were tested with Rust 1.97 and the 2024 edition. They are run as
Cargo projects with cargo run -q. Some outputs depend on thread
scheduling and change from run to run.
Sending a message
A worker thread sends one string to the main thread.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let msg = String::from("hello from the worker");
tx.send(msg).unwrap();
});
let received = rx.recv().unwrap();
println!("got: {received}");
}
The mpsc::channel function returns a pair: a
Sender<T> and a Receiver<T>. The
type T is inferred from the values we send, here
String.
thread::spawn(move || {
let msg = String::from("hello from the worker");
tx.send(msg).unwrap();
});
The sender is moved into the worker's closure. The send method
puts the value into the channel and returns at once. It returns an error
only if the receiver no longer exists.
let received = rx.recv().unwrap();
The recv method blocks the main thread until a value arrives.
We do not need to join the worker: receiving the message already tells us
it has done its work.
$ cargo run -q got: hello from the worker
Receiving in a loop
The receiver can be used as an iterator. The loop ends when the channel is closed.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
for word in ["sky", "cloud", "rain", "sun"] {
tx.send(word).unwrap();
thread::sleep(Duration::from_millis(100));
}
});
for word in rx {
println!("got: {word}");
}
println!("channel closed");
}
The worker sends four words, pausing after each one.
for word in rx {
println!("got: {word}");
}
Each iteration waits for the next value. When the worker finishes, its
closure is dropped, and with it the sender. A channel with no senders left
is disconnected; the iterator returns None after the
remaining values are received, and the loop ends.
$ cargo run -q got: sky got: cloud got: rain got: sun channel closed
Sending moves the value
Sending a value transfers its ownership to the channel and then to the receiving thread.
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let names = vec![String::from("Paul"), String::from("Lucy")];
tx.send(names).unwrap();
println!("sent {} names", names.len());
});
println!("{:?}", rx.recv().unwrap());
}
After sending the vector, the worker tries to print its length. The vector may already be in use by the receiving thread, so the compiler rejects the access with error E0382. This rule prevents two threads from using the same data at the same time. To keep using the data, we would send a clone or compute the length before sending.
$ cargo run -q
error[E0382]: borrow of moved value: `names`
--> src/main.rs:10:35
|
8 | let names = vec![String::from("Paul"), String::from("Lucy")];
| ----- move occurs because `names` has type `std::vec::Vec<std::string::String>`, which does not implement the `Copy` trait
9 | tx.send(names).unwrap();
| ----- value moved here
10 | println!("sent {} names", names.len());
| ^^^^^ value borrowed here after move
For more information about this error, try `rustc --explain E0382`.
Multiple producers
A sender can be cloned. All clones send into the same channel.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
for id in 1..=3 {
let tx = tx.clone();
thread::spawn(move || {
for n in 1..=3 {
tx.send(format!("producer {id}: message {n}")).unwrap();
thread::sleep(Duration::from_millis(10 * id));
}
});
}
drop(tx);
let messages: Vec<String> = rx.iter().collect();
for msg in &messages {
println!("{msg}");
}
println!("{} messages", messages.len());
}
Three producer threads each send three messages at a different pace.
let tx = tx.clone();
Each thread gets its own clone of the sender. Messages from one producer arrive in the order they were sent, but messages from different producers are interleaved as the threads run.
drop(tx); let messages: Vec<String> = rx.iter().collect();
The original sender is dropped in main. Without this line, the
channel would never be disconnected, because main would still
hold a sender, and collect would wait forever. The
iter method borrows the receiver and returns the same blocking
iterator as the for loop.
$ cargo run -q producer 1: message 1 producer 2: message 1 producer 3: message 1 producer 1: message 2 producer 1: message 3 producer 2: message 2 producer 3: message 2 producer 2: message 3 producer 3: message 3 9 messages
Disconnection
The recv_timeout method waits for a limited time. Its error
tells apart a channel that is merely empty from one that is closed.
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel::<u32>();
let worker_tx = tx.clone();
thread::spawn(move || {
worker_tx.send(42).unwrap();
});
loop {
match rx.recv_timeout(Duration::from_millis(200)) {
Ok(n) => println!("received {n}"),
Err(RecvTimeoutError::Timeout) => {
println!("timeout: a sender is still alive");
break;
}
Err(RecvTimeoutError::Disconnected) => {
println!("all senders are gone");
break;
}
}
}
drop(tx);
println!("after drop: {:?}", rx.recv());
}
The worker sends one value with a cloned sender and ends. The original
sender is still alive in main.
Err(RecvTimeoutError::Timeout) => {
println!("timeout: a sender is still alive");
break;
}
After the first value, no more values arrive. The channel is not
disconnected, because tx exists, so recv_timeout
returns Timeout after 200 milliseconds. A plain
recv would block forever here, which is a common mistake.
drop(tx);
println!("after drop: {:?}", rx.recv());
Once the last sender is dropped, recv returns
Err(RecvError) immediately.
$ cargo run -q received 42 timeout: a sender is still alive after drop: Err(RecvError)
When the receiver is gone
Sending fails when the receiver has been dropped. The error contains the value, so it is not lost.
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
let consumer = thread::spawn(move || {
for n in rx.iter().take(2) {
println!("consumer got {n}");
}
println!("consumer stops");
});
for n in 1..=5 {
if let Err(err) = tx.send(n) {
println!("send failed, value {} returned", err.0);
break;
}
thread::sleep(Duration::from_millis(50));
}
consumer.join().unwrap();
}
The consumer takes two values and ends, which drops the receiver.
if let Err(err) = tx.send(n) {
println!("send failed, value {} returned", err.0);
break;
}
The SendError is a tuple struct that holds the unsent value in
its field 0. A producer can use the error to stop, as here, or
to store the value elsewhere.
$ cargo run -q consumer got 1 consumer got 2 consumer stops send failed, value 3 returned
Non-blocking receive
The try_recv method checks the channel without waiting.
use std::sync::mpsc::{self, TryRecvError};
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let sum: u64 = (1..=50_000_000u64).sum();
thread::sleep(Duration::from_millis(250));
tx.send(sum).unwrap();
});
let mut ticks = 0;
loop {
match rx.try_recv() {
Ok(sum) => {
println!("result: {sum} after {ticks} ticks");
break;
}
Err(TryRecvError::Empty) => {
ticks += 1;
thread::sleep(Duration::from_millis(50));
}
Err(TryRecvError::Disconnected) => {
println!("worker died");
break;
}
}
}
}
The worker computes a sum and sends it. Meanwhile the main thread polls the channel and could do other work between the checks.
Err(TryRecvError::Empty) => {
ticks += 1;
thread::sleep(Duration::from_millis(50));
}
The Empty error means there is no value yet;
Disconnected means there will never be one. The number of
ticks depends on how fast the computation runs. Polling suits loops that
have other things to do, such as a game loop or a user interface event loop.
$ cargo run -q result: 1250000025000000 after 12 ticks
Bounded channels
An unbounded channel grows without limit if the producer is faster than the
consumer. A bounded channel created with sync_channel has a
fixed capacity; when it is full, send blocks until the
consumer makes room. This is called back-pressure.
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let (tx, rx) = mpsc::sync_channel(2);
let start = Instant::now();
let producer = thread::spawn(move || {
for n in 1..=5 {
tx.send(n).unwrap();
println!("{:>4} ms sent {n}", start.elapsed().as_millis());
}
});
thread::sleep(Duration::from_millis(300));
for n in rx {
println!("{:>4} ms received {n}", start.elapsed().as_millis());
thread::sleep(Duration::from_millis(100));
}
producer.join().unwrap();
}
The channel holds two values. The consumer starts after 300 milliseconds and then takes one value every 100 milliseconds.
let (tx, rx) = mpsc::sync_channel(2);
The sync_channel function returns a SyncSender
and a Receiver. The argument is the capacity.
The output shows the times in milliseconds. The first two values go into
the buffer at once. The third send blocks until the consumer
takes the first value at 300 ms. From then on, the producer runs at the
consumer's pace.
$ cargo run -q 0 ms sent 1 0 ms sent 2 300 ms received 1 300 ms sent 3 400 ms received 2 400 ms sent 4 500 ms received 3 500 ms sent 5 600 ms received 4 700 ms received 5
Rendezvous channels
A bounded channel with capacity zero has no buffer at all. Each
send waits until a receiver takes the value. Such a channel is
called a rendezvous channel: the two threads meet at the
handover.
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let (tx, rx) = mpsc::sync_channel(0);
let start = Instant::now();
thread::spawn(move || {
println!("{:>4} ms sender waits", start.elapsed().as_millis());
tx.send("ping").unwrap();
println!("{:>4} ms handed over", start.elapsed().as_millis());
});
thread::sleep(Duration::from_millis(200));
let msg = rx.recv().unwrap();
println!("{:>4} ms received {msg}", start.elapsed().as_millis());
thread::sleep(Duration::from_millis(50));
}
The sender is ready at once, but the main thread waits 200 milliseconds before receiving.
tx.send("ping").unwrap();
println!("{:>4} ms handed over", start.elapsed().as_millis());
The send call returns only when recv has taken
the value, so the sender knows that the message was received. The final
sleep in main gives the worker time to print its last line.
$ cargo run -q 0 ms sender waits 200 ms received ping 200 ms handed over
Commands and replies
Messages are often enums. One thread owns some state and other threads change it only by sending commands. If a command needs an answer, it carries its own reply channel.
use std::collections::HashMap;
use std::sync::mpsc::{self, Sender};
use std::thread;
enum Command {
Deposit(String, u64),
Withdraw(String, u64),
Balance(String, Sender<Option<u64>>),
}
fn bank(rx: mpsc::Receiver<Command>) {
let mut accounts: HashMap<String, u64> = HashMap::new();
for cmd in rx {
match cmd {
Command::Deposit(name, amount) => {
*accounts.entry(name).or_default() += amount;
}
Command::Withdraw(name, amount) => {
let balance = accounts.entry(name.clone()).or_default();
if *balance >= amount {
*balance -= amount;
} else {
println!("bank: {name} cannot withdraw {amount}");
}
}
Command::Balance(name, reply) => {
reply.send(accounts.get(&name).copied()).unwrap();
}
}
}
println!("bank: shutting down with {} accounts", accounts.len());
}
fn main() {
let (tx, rx) = mpsc::channel();
let bank = thread::spawn(move || bank(rx));
tx.send(Command::Deposit("paul".into(), 100)).unwrap();
tx.send(Command::Withdraw("paul".into(), 30)).unwrap();
tx.send(Command::Withdraw("paul".into(), 500)).unwrap();
tx.send(Command::Deposit("lucy".into(), 40)).unwrap();
for name in ["paul", "lucy", "mark"] {
let (reply_tx, reply_rx) = mpsc::channel();
tx.send(Command::Balance(name.into(), reply_tx)).unwrap();
println!("{name}: {:?}", reply_rx.recv().unwrap());
}
drop(tx);
bank.join().unwrap();
}
The bank thread owns the map of accounts. No mutex is needed, because no other thread can access the map.
enum Command {
Deposit(String, u64),
Withdraw(String, u64),
Balance(String, Sender<Option<u64>>),
}
The Balance command contains a sender for the answer. The
caller creates a new channel for each query, sends the command, and waits
on the receiving half.
Command::Balance(name, reply) => {
reply.send(accounts.get(&name).copied()).unwrap();
}
The bank processes the commands one by one in the order they arrived, so a
balance query sees all earlier deposits and withdrawals. When
main drops the sender, the for loop in the bank
ends and the thread shuts down. This design is known as the actor
pattern.
$ cargo run -q bank: paul cannot withdraw 500 paul: Some(70) lucy: Some(40) mark: None bank: shutting down with 2 accounts
A worker pool
A channel can distribute jobs to a fixed set of worker threads. Because a
channel has only one receiver, the workers share it through
Arc<Mutex<Receiver>>. A second channel collects
the results.
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::Duration;
type Job = Box<dyn FnOnce() -> String + Send>;
fn main() {
let (job_tx, job_rx) = mpsc::channel::<Job>();
let (result_tx, result_rx) = mpsc::channel();
let job_rx = Arc::new(Mutex::new(job_rx));
let workers: Vec<_> = (1..=3)
.map(|id| {
let job_rx = Arc::clone(&job_rx);
let result_tx = result_tx.clone();
thread::spawn(move || {
loop {
let job = job_rx.lock().unwrap().recv();
match job {
Ok(job) => result_tx.send(format!("worker {id}: {}", job())).unwrap(),
Err(_) => break,
}
}
})
})
.collect();
drop(result_tx);
for n in 1..=6u64 {
job_tx
.send(Box::new(move || {
thread::sleep(Duration::from_millis(50 * (n % 3 + 1)));
format!("{n}! = {}", (1..=n).product::<u64>())
}))
.unwrap();
}
drop(job_tx);
for line in result_rx {
println!("{line}");
}
for worker in workers {
worker.join().unwrap();
}
}
Three workers compute six factorials. The jobs take different times.
type Job = Box<dyn FnOnce() -> String + Send>;
A job is a boxed closure that returns a string. The closure must be
Send to be sent to another thread.
let job = job_rx.lock().unwrap().recv();
match job {
Ok(job) => result_tx.send(format!("worker {id}: {}", job())).unwrap(),
Err(_) => break,
}
A worker locks the receiver, waits for a job, and runs it. The mutex guard
is a temporary value that is dropped at the end of the let
statement, so the lock is released before the job runs and other workers
can take jobs in the meantime. Writing the lock and the
recv call in a while let condition would keep the
guard alive for the whole loop body, and the workers would run one at a
time. When the job channel is closed, recv fails and the worker
ends.
drop(result_tx); ... drop(job_tx);
The main thread drops its own result sender, so the result channel closes when the last worker ends; and it drops the job sender after queuing all jobs, so the workers stop. Which worker gets which job depends on thread scheduling. The results arrive in the order the jobs finish, not in the order they were queued.
$ cargo run -q worker 3: 3! = 6 worker 1: 1! = 1 worker 2: 2! = 2 worker 3: 4! = 24 worker 2: 6! = 720 worker 1: 5! = 120
A pipeline
Channels connect threads into a pipeline, where each stage receives values from the previous stage and sends results to the next one.
use std::sync::mpsc;
use std::thread;
fn main() {
let (num_tx, num_rx) = mpsc::channel();
let (sq_tx, sq_rx) = mpsc::channel();
let generator = thread::spawn(move || {
for n in 1..=10u64 {
num_tx.send(n).unwrap();
}
});
let squarer = thread::spawn(move || {
for n in num_rx {
if n % 2 == 1 {
sq_tx.send(n * n).unwrap();
}
}
});
let summer = thread::spawn(move || {
let values: Vec<u64> = sq_rx.iter().collect();
println!("squares of odd numbers: {values:?}");
values.iter().sum::<u64>()
});
generator.join().unwrap();
squarer.join().unwrap();
println!("sum: {}", summer.join().unwrap());
}
The first stage generates numbers, the second keeps the odd ones and squares them, and the third collects and sums the squares.
for n in num_rx {
if n % 2 == 1 {
sq_tx.send(n * n).unwrap();
}
}
The middle stage owns the receiver of the first channel and the sender of
the second one. When the generator ends, the first channel closes, the loop
ends, and the middle thread drops sq_tx, which closes the
second channel in turn. The shutdown travels down the pipeline. All stages
run in parallel; each value is processed as soon as it is available.
$ cargo run -q squares of odd numbers: [1, 9, 25, 49, 81] sum: 165
Channels with scoped threads
Scoped threads can borrow local data. Combined with a channel, they can send references to that data.
use std::sync::mpsc;
use std::thread;
fn main() {
let lines = [
"an old falcon",
"the sky is blue",
"rust is fast",
"a quiet afternoon by the river",
];
let (tx, rx) = mpsc::channel();
thread::scope(|s| {
for (i, line) in lines.iter().enumerate() {
let tx = tx.clone();
s.spawn(move || {
let longest = line.split_whitespace().max_by_key(|w| w.len()).unwrap();
tx.send((i, longest)).unwrap();
});
}
});
drop(tx);
let mut results: Vec<(usize, &str)> = rx.iter().collect();
results.sort();
for (i, word) in results {
println!("line {}: {word}", i + 1);
}
}
Each thread finds the longest word of one line and sends it with the line index.
tx.send((i, longest)).unwrap();
The longest value is a &str slice of the
lines array, which lives in main. This is allowed
because the threads cannot outlive the scope, and the array outlives both.
let mut results: Vec<(usize, &str)> = rx.iter().collect(); results.sort();
The results arrive in the order the threads finish. Sending the index with
each result lets us restore the original order. The scope joins all threads
before it returns, and the cloned senders are dropped with them; after we
drop the original sender, collect ends.
$ cargo run -q line 1: falcon line 2: blue line 3: fast line 4: afternoon
Sending errors
Workers can fail. Instead of panicking, a worker can send a
Result, and the receiver decides what to do with the errors.
use std::sync::mpsc;
use std::thread;
fn parse_line(line: &str) -> Result<(String, u32), String> {
let (name, age) = line
.split_once(',')
.ok_or_else(|| format!("missing comma in {line:?}"))?;
let age = age
.trim()
.parse()
.map_err(|e| format!("bad age in {line:?}: {e}"))?;
Ok((name.trim().to_string(), age))
}
fn main() {
let input = vec!["Paul, 34", "Lucy, x", "Mark 41", "Anna, 27"];
let (tx, rx) = mpsc::channel();
for (i, line) in input.into_iter().enumerate() {
let tx = tx.clone();
thread::spawn(move || tx.send((i, parse_line(line))).unwrap());
}
drop(tx);
let mut results: Vec<_> = rx.iter().collect();
results.sort_by_key(|(i, _)| *i);
for (i, result) in results {
match result {
Ok((name, age)) => println!("{i}: {name} is {age}"),
Err(e) => println!("{i}: error: {e}"),
}
}
}
Each thread parses one line of input. The channel carries the index and the
Result of the parsing.
thread::spawn(move || tx.send((i, parse_line(line))).unwrap());
The Result<(String, u32), String> type is
Send, because both of its variants are, so it can travel
through the channel like any other value. The main thread sorts the
results by index and reports the successes and the errors.
$ cargo run -q 0: Paul is 34 1: error: bad age in "Lucy, x": invalid digit found in string 2: error: missing comma in "Mark 41" 3: Anna is 27
Source
The Rust Programming Language: Using Message Passing to Transfer Data Between Threads
std::sync::mpsc module documentation
In this article we have passed messages between threads with channels in
Rust. Threads, locks, and atomics are covered in
Rust threads. For channels with several
consumers or for waiting on several channels at once, see the
crossbeam-channel crate.
Author
List all Rust tutorials.