Channels in Tokio
Tokio provides async channels for task coordination: mpsc for work queues, oneshot for single replies, broadcast for fan-out, and watch for latest-value signaling.
Recipe
Quick-reference recipe card - copy-paste ready.
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(32);
tx.send("job").await.unwrap();
println!("{}", rx.recv().await.unwrap());
}When to reach for this: Worker pools, request/response between tasks, shutdown broadcasts, and config change notifications.
Working Example
use tokio::sync::{broadcast, oneshot, watch};
#[tokio::main]
async fn main() {
// oneshot: single response
let (otx, orx) = oneshot::channel();
tokio::spawn(async move {
otx.send(42).unwrap();
});
println!("oneshot: {}", orx.await.unwrap());
// watch: latest value
let (wtx, mut wrx) = watch::channel("v1".to_string());
wtx.send("v2".into()).unwrap();
wrx.changed().await.unwrap();
println!("watch: {}", *wrx.borrow());
// broadcast: all subscribers
let (btx, mut brx) = broadcast::channel(16);
btx.send("event").unwrap();
println!("broadcast: {}", brx.recv().await.unwrap());
}What this demonstrates:
- Each channel type solves a different coordination pattern.
watchoverwrites - only latest value matters.broadcastmay lag if subscribers are slow (configurable error).
Deep Dive
How It Works
- mpsc: Multi-producer, single consumer; bounded
send().awaitapplies backpressure. - oneshot: One message, perfect for RPC-style callback between two tasks.
- broadcast: Clone receivers; each gets messages; slow consumers can be dropped.
- watch:
Sender::sendupdates;Receiver::changed().awaitwaits for new value.
Channel Picker
| Channel | Messages | Consumers | Typical use |
|---|---|---|---|
| mpsc | many | one | job queue |
| oneshot | one | one | single reply |
| broadcast | many | many | events, ticks |
| watch | latest | many | config, shutdown flag |
Rust Notes
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;
let (tx, rx) = mpsc::channel(8);
let mut stream = ReceiverStream::new(rx);
while let Some(item) = stream.next().await {
println!("{item}");
}- Adapt
mpsc::ReceivertoStreamfor iterator-style processing. Semaphorecomplements channels for limiting in-flight work.- Prefer
mpscoverMutex<VecDeque>for task boundaries.
Gotchas
- Unbounded mpsc -
channel(1)with slow consumer still blocks at capacity; unbounded not in std Tokio mpsc (always bounded). Fix: Size buffer to expected burst. - oneshot used twice - Second
sendfails. Fix: One receiver only; clone data before send if needed elsewhere. - broadcast lag error - Slow subscriber misses messages. Fix: Increase capacity or handle
RecvError::Lagged. - watch without changed() - May read stale without notification. Fix:
changed().awaitbefore relying on updates in loops. - Blocking in channel callbacks - Starves runtime. Fix: Keep handlers async and non-blocking.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
std::sync::mpsc | Sync threads only | Async tasks awaiting recv |
crossbeam | Sync MPMC patterns | Pure async await loops |
Mutex + Condvar | Legacy sync | Idiomatic Tokio services |
| Actor crate | Complex state machines | Simple request/response |
FAQs
mpsc buffer size?
Match burst size and backpressure tolerance - too small stalls producers; too large hides overload.
oneshot vs mpsc of 1?
oneshot is lighter for single reply; mpsc when multiple messages possible.
broadcast vs watch?
broadcast delivers every event; watch keeps only latest snapshot.
Close mpsc?
Drop all Sender handles; recv returns None.
try_recv?
Non-async poll - use in select! or bridging sync code carefully.
fairness?
mpsc FIFO per sender interleaving - not strict global fairness across many producers.
Axum + channels?
Spawn worker task with mpsc receiver; handlers send jobs via cloned Sender in state.
cancel-safe recv?
mpsc::recv in select! loops is cancel-safe per Tokio docs - still design for lost wakeups in complex setups.
Semaphore vs channel?
Semaphore limits concurrency; channel also transports data - often used together.
testing?
Use small buffers to force backpressure in tests; timeout on recv to catch hangs.
Related
- Channels - sync channels
- Streams - ReceiverStream
- select! & Concurrency - channel select
- Shared State in Async - actor pattern
- Graceful Shutdown - shutdown channels
Stack versions: This page was written for Rust 1.97.0 (edition 2024), Tokio 1.x, Axum 0.8, serde 1.0, sqlx 0.8, clap 4, and Polars 0.46+.