Tokio Channels Compared mpsc oneshot broadcast and watch

Tokio gives you four channel types, and picking the wrong one is the source of three recurring bugs: unbounded memory growth, lost updates, and a receiver that sees values it should have skipped. They differ in two questions that are easy to ask and easy to forget: how many senders and receivers can there be, and what happens when the receiver is slow. mpsc is a queue, oneshot is a reply slot, broadcast is a fan-out where every receiver sees every message, and watch is a single shared value where receivers see the latest. This post demonstrates each with code, shows what a slow receiver experiences in each, and ends with a decision table. Everything compiled on Rust 1.99, edition 2024, with tokio 1.

The four at a glance

The last column is the one to read first, because it is the contract you are signing. Each channel makes a different promise about data loss, and your design either fits that promise or has a bug.

mpsc: a work queue

Many producers, one consumer, every message delivered once in send order per sender. The usual shape is a worker task owning the receiver, with senders cloned into whatever produces work. Bounded channels apply backpressure: when the buffer is full, send().await waits.

use tokio::sync::mpsc;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(8);

    let worker = tokio::spawn(async move {
        let mut handled = Vec::new();
        while let Some(job) = rx.recv().await {
            handled.push(job.to_uppercase());
        }
        handled
    });

    let producers: Vec<_> = ["a", "b", "c"]
        .into_iter()
        .map(|name| {
            let tx = tx.clone();
            tokio::spawn(async move {
                for i in 1..=2 {
                    tx.send(format!("{name}{i}")).await.unwrap();
                }
            })
        })
        .collect();
    drop(tx);

    for p in producers {
        p.await.unwrap();
    }
    let mut out = worker.await.unwrap();
    out.sort();
    println!("{out:?}");
}

["A1", "A2", "B1", "B2", "C1", "C2"]

The line drop(tx) matters: the receiver's loop ends when every sender is dropped, and the original handle in main would otherwise keep the channel open forever. All six messages arrived. Capacity choice and what happens at the limit are the subject of bounded channels and backpressure.

What a full mpsc does: wait, fail or grow

The same producer-faster-than-consumer situation ends three ways depending on how you send. send().await waits for space, try_send returns an error immediately, and an unbounded channel never refuses and never stops growing:

use tokio::sync::mpsc::{self, error::TrySendError};
use tokio::time::{sleep, Duration, Instant};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(2);
    let consumer = tokio::spawn(async move {
        while let Some(n) = rx.recv().await {
            sleep(Duration::from_millis(100)).await;
            let _ = n;
        }
    });

    let t = Instant::now();
    for n in 1..=5 {
        tx.send(n).await.unwrap();
        println!("sent {n} at {:>3} ms", t.elapsed().as_millis());
    }

    match tx.try_send(99) {
        Err(TrySendError::Full(v)) => println!("try_send({v}): channel full, caller decides"),
        other => println!("try_send: {other:?}"),
    }
    drop(tx);
    consumer.await.unwrap();

    let (utx, _urx) = mpsc::unbounded_channel::<u32>();
    for n in 0..1_000_000 {
        utx.send(n).unwrap();
    }
    println!("unbounded accepted 1000000 messages with no consumer");
}

sent 1 at   0 ms
sent 2 at   0 ms
sent 3 at   0 ms
sent 4 at 100 ms
sent 5 at 200 ms
try_send(99): channel full, caller decides
unbounded accepted 1000000 messages with no consumer

With capacity 2 the producer sent three messages instantly (two fit in the buffer and the consumer had already taken the first), then had to wait for the slow consumer to make room, and from then on it sent one message per 100 ms, the consumer's pace. That throttling is backpressure, and it is the reason to prefer bounded channels almost everywhere. The unbounded channel swallowed a million messages with nobody reading, which is a million heap slots a slow consumer would have left behind in a real service. Choose the capacity from the delay you can tolerate, not from a round number: it is the number of messages that can be in flight before a producer notices that the consumer is behind, and it is also the number you lose if the process dies. Unbounded channels are fine for a small, known-finite set of messages or where the sender is itself rate limited; they are a memory leak waiting for a traffic spike otherwise.

oneshot: a reply slot

A request that expects a response carries its own return address, a oneshot::Sender. The worker owns the request queue; each request includes the sender for its answer. This is how you build an actor with a call-and-response interface:

use tokio::sync::{mpsc, oneshot};

enum Command {
    Get { key: &'static str, reply: oneshot::Sender<Option<u32>> },
    Set { key: &'static str, value: u32 },
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut rx) = mpsc::channel(16);

    tokio::spawn(async move {
        let mut store = std::collections::HashMap::new();
        while let Some(cmd) = rx.recv().await {
            match cmd {
                Command::Set { key, value } => {
                    store.insert(key, value);
                }
                Command::Get { key, reply } => {
                    let _ = reply.send(store.get(key).copied());
                }
            }
        }
    });

    tx.send(Command::Set { key: "answer", value: 42 }).await.unwrap();

    let (reply, answer) = oneshot::channel();
    tx.send(Command::Get { key: "answer", reply }).await.unwrap();
    println!("{:?}", answer.await.unwrap());

    let (reply, missing) = oneshot::channel();
    tx.send(Command::Get { key: "nothing", reply }).await.unwrap();
    println!("{:?}", missing.await.unwrap());
}

Some(42)
None

If the worker panics or drops the reply sender without sending, the awaiting side gets Err(RecvError) instead of waiting forever, which is the property that makes oneshot safe to build on. A single owner of the state with a message interface removes the need for a mutex, a design explored in sharing state between tasks.

broadcast: every receiver sees every message

A broadcast channel fans messages out to all subscribers, each with its own position in a shared ring buffer of fixed capacity. Slow receivers do not slow the sender; they lose old messages and are told so.

use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (tx, mut fast) = broadcast::channel::<u32>(4);
    let mut slow = tx.subscribe();

    for n in 1..=7 {
        tx.send(n).unwrap();
    }

    let mut seen = Vec::new();
    while !fast.is_empty() || seen.is_empty() {
        seen.push(fast.recv().await);
    }
    println!("fast receiver: {seen:?}");

    loop {
        match slow.recv().await {
            Ok(v) => println!("slow receiver got {v}"),
            Err(RecvError::Lagged(skipped)) => println!("slow receiver lagged, {skipped} messages lost"),
            Err(RecvError::Closed) => break,
        }
        if slow.is_empty() {
            break;
        }
    }
}

fast receiver: [Err(Lagged(3)), Ok(4), Ok(5), Ok(6), Ok(7)]
slow receiver lagged, 3 messages lost
slow receiver got 4
slow receiver got 5
slow receiver got 6
slow receiver got 7

The ring holds four messages. Seven were sent before anyone read, so the first three were overwritten. Both receivers were behind when they started reading, so both were told they lagged by three. The receiver can continue after Lagged, resuming at the oldest message still in the buffer. Always handle that arm explicitly; the unwrap() idiom turns a slow subscriber into a panic. Broadcast suits events where a late receiver can tolerate gaps (live metrics, presence, chat typing indicators), and not state that must be exact.

watch: the latest value, not the history

A watch channel is a variable with change notification. Senders overwrite the value; receivers wait for it to change and read the current one. Intermediate values are skipped, which is exactly right for configuration, status flags and "shutdown requested":

use tokio::sync::watch;
use tokio::time::{sleep, Duration};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = watch::channel("starting");

    let watcher = tokio::spawn(async move {
        let mut observed = vec![*rx.borrow()];
        while rx.changed().await.is_ok() {
            observed.push(*rx.borrow_and_update());
            sleep(Duration::from_millis(100)).await;
        }
        observed
    });

    sleep(Duration::from_millis(10)).await;
    for state in ["loading", "warming", "ready", "serving"] {
        tx.send(state).unwrap();
    }
    sleep(Duration::from_millis(500)).await;
    drop(tx);

    println!("{:?}", watcher.await.unwrap());
}

["starting", "serving"]

Four states were sent back to back, and the watcher saw only the first and last of them, because by the time it called changed() again, the intermediate values had been overwritten. That is the right behaviour for "what is the current config?" and the wrong one for "what happened?". For an audit log of every transition, use mpsc.

Choosing

The row for multiple consumers is the gap in the standard set: Tokio's mpsc has exactly one receiver. Sharing it across workers through Arc<Mutex<Receiver>> works and serialises the recv calls; the cleaner fix is a multi-consumer queue crate, or to shard the load across several mpsc channels. The synchronous equivalents live in the standard library, covered in mpsc channels in threads, and the choice between all of these and shared state is in sharing state.