Channels in Tokio

Q: oneshot vs mpsc of 1?

oneshot is lighter for single reply; mpsc when multiple messages possible.

Q: broadcast vs watch?

broadcast delivers every event; watch keeps only latest snapshot.

Q: Close mpsc?

Drop all Sender handles; recv returns None.

Q: try_recv?

Non-async poll - use in select! or bridging sync code carefully.

Q: cancel-safe recv?

mpsc::recv in select! loops is cancel-safe per Tokio docs - still design for lost wakeups in complex setups.

Q: testing?

Use small buffers to force backpressure in tests; timeout on recv to catch hangs.

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.
watch overwrites - only latest value matters.
broadcast may lag if subscribers are slow (configurable error).

Deep Dive

How It Works

mpsc: Multi-producer, single consumer; bounded send().await applies 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::send updates; Receiver::changed().await waits 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::Receiver to Stream for iterator-style processing.
Semaphore complements channels for limiting in-flight work.
Prefer mpsc over Mutex<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 send fails. 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().await before 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.

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+.

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.
watch overwrites - only latest value matters.
broadcast may lag if subscribers are slow (configurable error).

Deep Dive

How It Works

mpsc: Multi-producer, single consumer; bounded send().await applies 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::send updates; Receiver::changed().await waits 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::Receiver to Stream for iterator-style processing.
Semaphore complements channels for limiting in-flight work.
Prefer mpsc over Mutex<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 send fails. 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().await before 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.

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+.

Channels in Tokio

Buscar en la documentación

Channels in Tokio