Async Channels & Locks

The Channels and Shared State lessons used std::sync with threads. Inside async code those types have a problem: their blocking operations (recv(), lock() on a contended mutex) block the whole worker thread, stalling every other task on it. Tokio provides async versions in tokio::sync, where waiting is an .await that frees the thread.

Need	Threads (`std::sync`)	Async (`tokio::sync`)
queue of messages, many senders, one receiver	`mpsc::channel`	`mpsc::channel(capacity)`
a single reply	a channel used once	`oneshot::channel()`
every subscriber gets every message	(none)	`broadcast::channel(capacity)`
latest value, change notification	`RwLock` + `Condvar`	`watch::channel(initial)`
wake someone up	`Condvar`	`Notify`
mutual exclusion	`Mutex`	`Mutex` (async `lock().await`)
init once	`OnceLock`	`OnceCell` (async initializer)

1. Tokio `mpsc`: a bounded async queue

mpsc::channel(capacity) is bounded: send(...).await waits while the channel is full, and recv().await waits for a message. Both wait asynchronously, so the thread stays free.

Rust

use tokio::sync::mpsc;

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

    tokio::spawn(async move {
        for order in ["order-1", "order-2", "order-3"] {
            tx.send(order.to_string()).await.unwrap();
        }
        // tx dropped here: the channel closes
    });

    while let Some(order) = rx.recv().await {
        println!("processing {order}");
    }
    println!("all orders processed");
}

Output

processing order-1
processing order-2
processing order-3
all orders processed

recv() returns None once every Sender is dropped and the buffer is empty, which ends the while let loop, just like iterating a std receiver.

2. Multiple producers

Sender is cloneable; each clone is another producer. As with threads, messages from different producers interleave, so this example sorts before printing.

Rust

use std::time::Duration;
use tokio::sync::mpsc;

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

    for sensor in ["temp", "humidity", "pressure"] {
        let tx = tx.clone();
        tokio::spawn(async move {
            for reading in 1..=2 {
                tokio::time::sleep(Duration::from_millis(5)).await;
                tx.send(format!("{sensor}#{reading}")).await.unwrap();
            }
        });
    }
    drop(tx); // otherwise recv() never returns None

    let mut readings = Vec::new();
    while let Some(r) = rx.recv().await {
        readings.push(r);
    }
    readings.sort();
    println!("{readings:?}");
}

Output

["humidity#1", "humidity#2", "pressure#1", "pressure#2", "temp#1", "temp#2"]

3. Backpressure: a slow consumer slows the producer

With a small capacity, a fast producer spends its time waiting in send().await instead of piling messages into memory. The wait doesn't hold a thread.

Rust

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::Instant;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(2);
    let start = Instant::now();

    let producer = tokio::spawn(async move {
        for i in 1..=6 {
            tx.send(i).await.unwrap(); // waits while 2 messages are queued
        }
        start.elapsed()
    });

    let mut got = Vec::new();
    while let Some(i) = rx.recv().await {
        tokio::time::sleep(Duration::from_millis(20)).await; // slow consumer
        got.push(i);
    }

    let producer_time = producer.await.unwrap();
    println!("got {got:?}");
    println!("producer had to wait for the consumer: {}", producer_time >= Duration::from_millis(40));
}

Output

got [1, 2, 3, 4, 5, 6]
producer had to wait for the consumer: true

4. `try_send`: shed load instead of waiting

try_send never waits: when the channel is full it returns TrySendError::Full with the message, so you can drop it, count it, or respond "busy". This is how a server protects itself under overload.

Rust

use tokio::sync::mpsc::{self, error::TrySendError};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(3);
    let mut rejected = Vec::new();

    for request in 1..=5 {
        match tx.try_send(request) {
            Ok(()) => {}
            Err(TrySendError::Full(r)) => rejected.push(r), // reply "503 busy"
            Err(TrySendError::Closed(_)) => break,
        }
    }
    drop(tx);

    let mut accepted = Vec::new();
    while let Some(r) = rx.recv().await {
        accepted.push(r);
    }
    println!("accepted {accepted:?}, rejected {rejected:?}");
}

Output

accepted [1, 2, 3], rejected [4, 5]

5. `send_timeout` and unbounded channels

send_timeout waits up to a limit for space. mpsc::unbounded_channel() never makes the sender wait (its send isn't even async), at the cost of unlimited memory growth if the consumer falls behind. Prefer bounded channels in servers.

Rust

use std::time::Duration;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, _rx) = mpsc::channel::<&str>(1);
    tx.send("fills the only slot").await.unwrap();
    let result = tx.send_timeout("no room", Duration::from_millis(30)).await;
    println!("send_timeout on a full channel: {}", result.is_err());

    let (utx, mut urx) = mpsc::unbounded_channel();
    for i in 0..1000 {
        utx.send(i).unwrap(); // never waits
    }
    drop(utx);
    let mut sum = 0;
    while let Some(i) = urx.recv().await {
        sum += i;
    }
    println!("unbounded channel delivered sum {sum}");
}

Output

send_timeout on a full channel: true
unbounded channel delivered sum 499500

6. `oneshot`: exactly one value

A oneshot channel carries a single value from one sender to one receiver. The receiver itself is a future: .await it to get the value.

Rust

use std::time::Duration;
use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel();

    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(20)).await; // compute something
        let _ = tx.send(42); // send consumes the sender: only one value, ever
    });

    match rx.await {
        Ok(v) => println!("got {v}"),
        Err(_) => println!("sender dropped without sending"),
    }
}

Output

got 42

7. `oneshot` tells you when the other side gave up

If the sender is dropped without sending, the receiver gets an error. If the receiver is dropped, send returns the value back as an error, and tx.closed().await lets a worker notice that nobody wants the result anymore and stop early.

Rust

use std::time::Duration;
use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    // Sender dropped: the receiver learns the work was abandoned.
    let (tx, rx) = oneshot::channel::<u32>();
    drop(tx);
    println!("receiver: {:?}", rx.await);

    // Receiver dropped: the worker notices and stops computing.
    let (mut tx, rx) = oneshot::channel::<u32>();
    let worker = tokio::spawn(async move {
        tokio::select! {
            _ = tokio::time::sleep(Duration::from_secs(10)) => "finished (too late)",
            _ = tx.closed() => "caller went away, stopping early",
        }
    });
    drop(rx); // e.g. the HTTP client disconnected
    println!("worker: {}", worker.await.unwrap());
}

Output

receiver: Err(RecvError(()))
worker: caller went away, stopping early

8. Request-reply over `mpsc` + `oneshot`

The async version of the request/response pattern: each request carries a oneshot::Sender for its answer.

Rust

use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};

struct Lookup {
    user_id: u32,
    reply: oneshot::Sender<Option<String>>,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Lookup>(32);

    // A "directory service" task that owns its data.
    tokio::spawn(async move {
        let users = HashMap::from([(1, "ana"), (2, "ben")]);
        while let Some(req) = rx.recv().await {
            let _ = req.reply.send(users.get(&req.user_id).map(|s| s.to_string()));
        }
    });

    for id in [2, 7, 1] {
        let (reply, answer) = oneshot::channel();
        tx.send(Lookup { user_id: id, reply }).await.unwrap();
        println!("user {id}: {:?}", answer.await.unwrap());
    }
}

Output

user 2: Some("ben")
user 7: None
user 1: Some("ana")

9. `broadcast`: every subscriber gets every message

With broadcast, each receiver (created with subscribe()) sees every message sent after it subscribed. Values must be Clone, since each receiver gets its own copy.

Rust

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _) = broadcast::channel::<String>(16);

    let mut handles = Vec::new();
    for name in ["email-service", "audit-log", "analytics"] {
        let mut rx = tx.subscribe();
        handles.push(tokio::spawn(async move {
            let mut seen = Vec::new();
            while let Ok(event) = rx.recv().await {
                seen.push(event);
            }
            format!("{name} saw {seen:?}")
        }));
    }

    for event in ["user.signup", "order.paid"] {
        tx.send(event.to_string()).unwrap();
    }
    drop(tx); // closing the channel ends every subscriber's loop

    for h in handles {
        println!("{}", h.await.unwrap());
    }
}

Output

email-service saw ["user.signup", "order.paid"]
audit-log saw ["user.signup", "order.paid"]
analytics saw ["user.signup", "order.paid"]

10. Slow `broadcast` subscribers lag

A broadcast channel keeps a fixed number of recent messages. A subscriber that falls further behind loses the oldest ones and is told how many with RecvError::Lagged(n). The fast subscribers are never slowed down by it. (Tokio rounds the capacity up to a power of two.)

Rust

use tokio::sync::broadcast::{self, error::RecvError};

#[tokio::main]
async fn main() {
    let (tx, mut slow) = broadcast::channel(4); // keeps the last 4 messages

    for price in [100, 101, 102, 103, 104] {
        tx.send(price).unwrap();
    }

    loop {
        match slow.recv().await {
            Ok(p) => println!("price {p}"),
            Err(RecvError::Lagged(missed)) => println!("lagged: missed {missed} updates"),
            Err(RecvError::Closed) => break,
        }
        if slow.is_empty() {
            break;
        }
    }
}

Output

lagged: missed 1 updates
price 101
price 102
price 103
price 104

11. `watch`: always the latest value

A watch channel holds one value. Senders replace it; receivers can read the current value at any time (borrow()) and wait for the next change (changed().await). Intermediate values can be skipped, which is exactly right for state like configuration or a status flag.

Rust

use std::time::Duration;
use tokio::sync::watch;

#[derive(Debug, Clone)]
struct Config {
    max_connections: u32,
    maintenance: bool,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = watch::channel(Config { max_connections: 100, maintenance: false });

    let server = tokio::spawn(async move {
        let mut seen = vec![rx.borrow().max_connections];
        while rx.changed().await.is_ok() {
            let cfg = rx.borrow_and_update().clone();
            seen.push(cfg.max_connections);
            if cfg.maintenance {
                break;
            }
        }
        seen
    });

    tokio::time::sleep(Duration::from_millis(10)).await;
    tx.send(Config { max_connections: 200, maintenance: false }).unwrap();
    tokio::time::sleep(Duration::from_millis(10)).await;
    tx.send(Config { max_connections: 50, maintenance: true }).unwrap();

    let seen = server.await.unwrap();
    println!("first value: {:?}, last value: {:?}", seen.first(), seen.last());
    println!("saw at most 3 values (updates can be skipped): {}", seen.len() <= 3);
}

Output

first value: Some(100), last value: Some(50)
saw at most 3 values (updates can be skipped): true

12. `watch` as a shutdown signal

Because every receiver sees the latest value, a watch::channel(false) is a clean way to tell many tasks to stop. Tasks can check it at any time, or select! on changed().

Rust

use std::time::Duration;
use tokio::sync::watch;
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
    let mut tasks = JoinSet::new();

    for name in ["http", "grpc", "metrics"] {
        let mut shutdown = shutdown_rx.clone();
        tasks.spawn(async move {
            let mut served = 0u32;
            loop {
                tokio::select! {
                    _ = shutdown.changed() => break,
                    _ = tokio::time::sleep(Duration::from_millis(5)) => served += 1,
                }
            }
            format!("{name} stopped (served something: {})", served > 0)
        });
    }

    tokio::time::sleep(Duration::from_millis(50)).await;
    shutdown_tx.send(true).unwrap(); // e.g. on SIGTERM

    let mut lines = Vec::new();
    while let Some(res) = tasks.join_next().await {
        lines.push(res.unwrap());
    }
    lines.sort();
    println!("{}", lines.join("\n"));
}

Output

grpc stopped (served something: true)
http stopped (served something: true)
metrics stopped (served something: true)

13. `Notify`: a wake-up without data

Notify wakes waiting tasks without sending a value. notify_one stores a permit if nobody is waiting yet, so a notification sent before the wait isn't lost.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;

#[tokio::main]
async fn main() {
    let data_ready = Arc::new(Notify::new());

    let loader = {
        let data_ready = Arc::clone(&data_ready);
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(20)).await; // load data
            data_ready.notify_one();
        })
    };

    data_ready.notified().await; // wait for the signal
    println!("data is ready, starting to serve");
    loader.await.unwrap();

    // A permit stored before anyone waits is used by the next waiter.
    data_ready.notify_one();
    data_ready.notified().await;
    println!("stored permit consumed immediately");
}

Output

data is ready, starting to serve
stored permit consumed immediately

14. A `std::sync::MutexGuard` across `.await` won't compile in `spawn`

A std mutex guard is not Send, so a future that holds one across an .await can't be spawned. The compiler catches this common bug:

Rust✗ does not compile

use std::sync::{Arc, Mutex};

async fn save(_v: u32) {}

#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0));
    let c = Arc::clone(&counter);
    tokio::spawn(async move {
        let mut guard = c.lock().unwrap();
        *guard += 1;
        save(*guard).await; // the guard is still alive here
    })
    .await
    .unwrap();
}

Output

error: future cannot be sent between threads safely
   --> src/main.rs:9:5
    |
  9 | /     tokio::spawn(async move {
 10 | |         let mut guard = c.lock().unwrap();
 11 | |         *guard += 1;
 12 | |         save(*guard).await; // the guard is still alive here
 13 | |     })
    | |______^ future created by async block is not `Send`
    |
    = help: within `{async block@src/main.rs:9:18: 9:28}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:12:22
    |
 10 |         let mut guard = c.lock().unwrap();
    |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
 11 |         *guard += 1;
 12 |         save(*guard).await; // the guard is still alive here
    |                      ^^^^^ await occurs here, with `mut guard` maybe used later
note: required by a bound in `tokio::spawn`
   --> ~/.cargo/registry/src/.../tokio-1.53.1/src/task/spawn.rs:176:21
    |
174 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
175 |     where
176 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

Holding a blocking lock while the task is suspended would block every other task that needs the lock (possibly on the same thread, deadlocking it). The next two examples show the two fixes.

15. Fix 1: keep `std::sync::Mutex`, release before `.await`

For short critical sections, a plain std::sync::Mutex is the recommended choice even in async code: it's faster than the async one. Just make sure the guard is dropped before any .await, typically by locking inside a small block or helper function.

Rust

use std::sync::{Arc, Mutex};
use std::time::Duration;

async fn save(v: u32) -> u32 {
    tokio::time::sleep(Duration::from_millis(5)).await;
    v
}

#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0u32));

    let mut handles = Vec::new();
    for _ in 0..10 {
        let c = Arc::clone(&counter);
        handles.push(tokio::spawn(async move {
            let value = {
                let mut guard = c.lock().unwrap();
                *guard += 1;
                *guard
            }; // guard dropped here, before the await
            save(value).await
        }));
    }
    let mut saved = Vec::new();
    for h in handles {
        saved.push(h.await.unwrap());
    }
    saved.sort();
    println!("saved {saved:?}");
}

Output

saved [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

16. Fix 2: `tokio::sync::Mutex` when you must hold the lock across `.await`

Sometimes the lock must stay held during async work: for example, a single connection that only one task may use at a time. tokio::sync::Mutex::lock() is async (waiting doesn't block the thread), and its guard can be held across .await.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

struct Connection {
    log: Vec<String>,
}

impl Connection {
    async fn query(&mut self, sql: &str) -> usize {
        tokio::time::sleep(Duration::from_millis(5)).await; // network round trip
        self.log.push(sql.to_string());
        sql.len()
    }
}

#[tokio::main]
async fn main() {
    let conn = Arc::new(Mutex::new(Connection { log: Vec::new() }));

    let mut handles = Vec::new();
    for id in 1..=3 {
        let conn = Arc::clone(&conn);
        handles.push(tokio::spawn(async move {
            let mut c = conn.lock().await; // async wait for the lock
            // A transaction: both queries run without other tasks interleaving.
            c.query(&format!("BEGIN; -- tx {id}")).await;
            c.query(&format!("COMMIT; -- tx {id}")).await
        }));
    }
    for h in handles {
        h.await.unwrap();
    }

    let log = &conn.lock().await.log;
    let paired = log.chunks(2).all(|p| p[0].replace("BEGIN;", "") == p[1].replace("COMMIT;", ""));
    println!("{} statements, every BEGIN directly followed by its COMMIT: {paired}", log.len());
}

Output

6 statements, every BEGIN directly followed by its COMMIT: true

Use `std::sync::Mutex`	Use `tokio::sync::Mutex`
the critical section is short and has no `.await`	the lock must be held across `.await`
most shared counters, maps, caches	serializing access to an async resource (a connection, a file)
faster	slower, but never blocks the thread

17. `tokio::sync::RwLock`

The async RwLock works the same way: read().await and write().await. Like the async mutex, reach for it when a guard must live across an .await.

Rust

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

async fn render(template: &str, name: &str) -> String {
    tokio::task::yield_now().await; // pretend rendering is async work
    template.replace("{name}", name)
}

#[tokio::main]
async fn main() {
    let templates = Arc::new(RwLock::new(HashMap::from([("welcome", String::from("Hi {name}!"))])));

    let mut handles = Vec::new();
    for name in ["ana", "ben"] {
        let templates = Arc::clone(&templates);
        handles.push(tokio::spawn(async move {
            let t = templates.read().await; // many readers at once
            render(&t["welcome"], name).await // read guard held across the await
        }));
    }
    for h in handles {
        println!("{}", h.await.unwrap());
    }

    templates.write().await.insert("welcome", String::from("Welcome back, {name}."));
    println!("{}", render(&templates.read().await["welcome"], "cy").await);
}

Output

Hi ana!
Hi ben!
Welcome back, cy.

18. `OnceCell` with an async initializer

tokio::sync::OnceCell is OnceLock with an async initializer: the first caller runs it (for example, connecting to a database), concurrent callers wait for it, and later callers get the cached value immediately.

Rust

use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use tokio::sync::OnceCell;

static CONNECT_ATTEMPTS: AtomicU32 = AtomicU32::new(0);
static DB_POOL: OnceCell<String> = OnceCell::const_new();

async fn pool() -> &'static str {
    DB_POOL
        .get_or_init(|| async {
            CONNECT_ATTEMPTS.fetch_add(1, Ordering::Relaxed);
            tokio::time::sleep(Duration::from_millis(30)).await; // connect
            String::from("pool(size=10)")
        })
        .await
}

#[tokio::main]
async fn main() {
    let (a, b, c) = tokio::join!(pool(), pool(), pool());
    println!("{a} {b} {c}");
    println!("connected {} time(s)", CONNECT_ATTEMPTS.load(Ordering::Relaxed));
}

Output

pool(size=10) pool(size=10) pool(size=10)
connected 1 time(s)

19. Bridging sync and async code

Tokio channels can connect plain threads and async tasks. blocking_send and blocking_recv are the non-async versions for use outside the runtime (in a std::thread or spawn_blocking). Calling them inside async code would panic.

Rust

use tokio::sync::mpsc;

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

    // A legacy, synchronous component running on its own OS thread.
    let legacy = std::thread::spawn(move || {
        for line in ["ERROR disk full", "INFO ok", "ERROR timeout"] {
            if line.starts_with("ERROR") {
                tx.blocking_send(line.to_string()).unwrap();
            }
        }
    });

    // The async side receives normally.
    let mut alerts = Vec::new();
    while let Some(line) = rx.recv().await {
        alerts.push(line);
    }
    legacy.join().unwrap();
    println!("alerts: {alerts:?}");
}

Output

alerts: ["ERROR disk full", "ERROR timeout"]

20. The async actor pattern

An actor task owns its state and handles messages one at a time; a cloneable handle wraps the mpsc::Sender and offers async methods. No locks anywhere, and the state can hold non-Sync data.

Rust

use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};

enum Msg {
    AddToCart { user: String, item: String },
    GetCart { user: String, reply: oneshot::Sender<Vec<String>> },
}

#[derive(Clone)]
struct CartService {
    tx: mpsc::Sender<Msg>,
}

impl CartService {
    fn spawn() -> Self {
        let (tx, mut rx) = mpsc::channel(64);
        tokio::spawn(async move {
            let mut carts: HashMap<String, Vec<String>> = HashMap::new();
            while let Some(msg) = rx.recv().await {
                match msg {
                    Msg::AddToCart { user, item } => carts.entry(user).or_default().push(item),
                    Msg::GetCart { user, reply } => {
                        let _ = reply.send(carts.get(&user).cloned().unwrap_or_default());
                    }
                }
            }
        });
        CartService { tx }
    }

    async fn add(&self, user: &str, item: &str) {
        let _ = self.tx.send(Msg::AddToCart { user: user.into(), item: item.into() }).await;
    }

    async fn cart(&self, user: &str) -> Vec<String> {
        let (reply, rx) = oneshot::channel();
        let _ = self.tx.send(Msg::GetCart { user: user.into(), reply }).await;
        rx.await.unwrap_or_default()
    }
}

#[tokio::main]
async fn main() {
    let carts = CartService::spawn();
    let c2 = carts.clone(); // handles are cheap to clone and share

    carts.add("ana", "book").await;
    c2.add("ana", "lamp").await;
    carts.add("ben", "tent").await;

    println!("ana: {:?}", carts.cart("ana").await);
    println!("ben: {:?}", c2.cart("ben").await);
    println!("cy:  {:?}", carts.cart("cy").await);
}

Output

ana: ["book", "lamp"]
ben: ["tent"]
cy:  []

Tokio's mpsc::Receiver has a single owner. To let several worker tasks pull from one queue, share the receiver behind an async Mutex: a worker holds the lock only while waiting for the next message.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc};

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<u32>(32);
    let rx = Arc::new(Mutex::new(rx));
    let (done_tx, mut done_rx) = mpsc::channel(32);

    for worker in 0..3 {
        let (rx, done_tx) = (Arc::clone(&rx), done_tx.clone());
        tokio::spawn(async move {
            loop {
                let job = rx.lock().await.recv().await; // lock released at the end of this statement
                let Some(job) = job else { break };
                tokio::time::sleep(Duration::from_millis(5)).await;
                done_tx.send((job, worker)).await.unwrap();
            }
        });
    }
    drop(done_tx);

    for job in 1..=9 {
        tx.send(job).await.unwrap();
    }
    drop(tx);

    let mut finished = Vec::new();
    while let Some((job, _worker)) = done_rx.recv().await {
        finished.push(job);
    }
    finished.sort();
    println!("{finished:?}");
}

Output

[1, 2, 3, 4, 5, 6, 7, 8, 9]

The async-channel and flume crates offer multi-consumer channels directly, if you'd rather not share a receiver.

22. Realistic: a chat room with `broadcast`

Each connected client gets a broadcast::Receiver and a clone of the Sender. A message from any client goes to all, and each client filters out its own messages.

Rust

use tokio::sync::broadcast;

#[derive(Clone, Debug)]
struct ChatMsg {
    from: String,
    text: String,
}

async fn client(name: &'static str, tx: broadcast::Sender<ChatMsg>, mut rx: broadcast::Receiver<ChatMsg>, says: &'static [&'static str]) -> Vec<String> {
    for text in says {
        tx.send(ChatMsg { from: name.into(), text: text.to_string() }).unwrap();
        tokio::task::yield_now().await;
    }
    drop(tx);
    let mut inbox = Vec::new();
    while let Ok(msg) = rx.recv().await {
        if msg.from != name {
            inbox.push(format!("{}: {}", msg.from, msg.text));
        }
    }
    inbox
}

#[tokio::main]
async fn main() {
    let (tx, _) = broadcast::channel(64);
    // Subscribe everyone before anyone talks, so nobody misses a message.
    let (ana_rx, ben_rx, cy_rx) = (tx.subscribe(), tx.subscribe(), tx.subscribe());

    let ana = tokio::spawn(client("ana", tx.clone(), ana_rx, &["hi all", "who's up for lunch?"]));
    let ben = tokio::spawn(client("ben", tx.clone(), ben_rx, &["me!"]));
    let cy = tokio::spawn(client("cy", tx.clone(), cy_rx, &[]));
    drop(tx);

    for (name, handle) in [("ana", ana), ("ben", ben), ("cy", cy)] {
        let mut inbox = handle.await.unwrap();
        inbox.sort();
        println!("{name} received {inbox:?}");
    }
}

Output

ana received ["ben: me!"]
ben received ["ana: hi all", "ana: who's up for lunch?"]
cy received ["ana: hi all", "ana: who's up for lunch?", "ben: me!"]

23. Realistic: hot-reloading configuration

A config watcher publishes new settings through a watch channel; request handlers read the latest value on every request with borrow(). There's no lock contention: borrow() is a quick read, and handlers never wait for updates.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;

#[derive(Debug)]
struct Settings {
    greeting: String,
    rate_limit: u32,
}

async fn handle_request(user: &str, settings: &watch::Receiver<Arc<Settings>>) -> String {
    let s = Arc::clone(&settings.borrow()); // take a snapshot, release the borrow
    tokio::task::yield_now().await;
    format!("{} {user} (limit {})", s.greeting, s.rate_limit)
}

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel(Arc::new(Settings { greeting: String::from("Hello"), rate_limit: 10 }));

    println!("{}", handle_request("ana", &rx).await);

    // The file watcher notices a change and publishes new settings.
    let watcher = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(10)).await;
        tx.send(Arc::new(Settings { greeting: String::from("Bonjour"), rate_limit: 50 })).unwrap();
        tx // keep the sender alive until main is done
    });
    let _tx = watcher.await.unwrap();

    println!("{}", handle_request("ben", &rx).await);
}

Output

Hello ana (limit 10)
Bonjour ben (limit 50)

Holding a watch borrow across an .await would block the sender, so the handler clones the Arc out first.

24. Realistic: batching writes with a flush interval

Events arrive on a channel; a writer task buffers them and flushes when the batch is full or when a timer fires, whichever comes first. select! over recv() and interval.tick() expresses this directly.

Rust

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{MissedTickBehavior, interval};

async fn batch_writer(mut rx: mpsc::Receiver<String>, max_batch: usize) -> Vec<Vec<String>> {
    let mut flushes = Vec::new();
    let mut batch = Vec::new();
    let mut timer = interval(Duration::from_millis(50));
    timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
    timer.tick().await; // the first tick fires immediately; skip it

    loop {
        tokio::select! {
            msg = rx.recv() => match msg {
                Some(m) => {
                    batch.push(m);
                    if batch.len() == max_batch {
                        flushes.push(std::mem::take(&mut batch)); // flush: full
                    }
                }
                None => {
                    if !batch.is_empty() {
                        flushes.push(batch); // flush: shutting down
                    }
                    return flushes;
                }
            },
            _ = timer.tick() => {
                if !batch.is_empty() {
                    flushes.push(std::mem::take(&mut batch)); // flush: timer
                }
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel(100);
    let writer = tokio::spawn(batch_writer(rx, 3));

    for i in 1..=4 {
        tx.send(format!("e{i}")).await.unwrap(); // burst: one full batch + 1 left over
    }
    tokio::time::sleep(Duration::from_millis(120)).await; // quiet: the timer flushes e4
    tx.send(String::from("e5")).await.unwrap();
    drop(tx); // shutdown flushes e5

    for (n, batch) in writer.await.unwrap().iter().enumerate() {
        println!("flush {}: {batch:?}", n + 1);
    }
}

Output

flush 1: ["e1", "e2", "e3"]
flush 2: ["e4"]
flush 3: ["e5"]

25. Realistic: an event bus with typed subscribers

A small in-process event bus: publishers send typed events on a broadcast channel, and each subscriber task reacts only to the events it cares about. Adding a subscriber doesn't touch the publishers.

Rust

use tokio::sync::broadcast;

#[derive(Clone, Debug)]
enum Event {
    UserRegistered { email: String },
    OrderPlaced { order_id: u32, cents: u64 },
    OrderShipped { order_id: u32 },
}

fn spawn_subscriber<F>(bus: &broadcast::Sender<Event>, name: &'static str, mut react: F) -> tokio::task::JoinHandle<Vec<String>>
where
    F: FnMut(&Event) -> Option<String> + Send + 'static,
{
    let mut rx = bus.subscribe();
    tokio::spawn(async move {
        let mut actions = Vec::new();
        while let Ok(event) = rx.recv().await {
            if let Some(action) = react(&event) {
                actions.push(format!("[{name}] {action}"));
            }
        }
        actions
    })
}

#[tokio::main]
async fn main() {
    let (bus, _) = broadcast::channel(32);

    let mailer = spawn_subscriber(&bus, "mailer", |e| match e {
        Event::UserRegistered { email } => Some(format!("send welcome mail to {email}")),
        Event::OrderShipped { order_id } => Some(format!("send tracking for order {order_id}")),
        _ => None,
    });
    let mut revenue = 0;
    let accounting = spawn_subscriber(&bus, "accounting", move |e| match e {
        Event::OrderPlaced { cents, .. } => {
            revenue += cents;
            Some(format!("revenue now {revenue}"))
        }
        _ => None,
    });

    bus.send(Event::UserRegistered { email: String::from("ana@example.com") }).unwrap();
    bus.send(Event::OrderPlaced { order_id: 1, cents: 2500 }).unwrap();
    bus.send(Event::OrderPlaced { order_id: 2, cents: 1000 }).unwrap();
    bus.send(Event::OrderShipped { order_id: 1 }).unwrap();
    drop(bus);

    for line in mailer.await.unwrap().into_iter().chain(accounting.await.unwrap()) {
        println!("{line}");
    }
}

Output

[mailer] send welcome mail to ana@example.com
[mailer] send tracking for order 1
[accounting] revenue now 2500
[accounting] revenue now 3500

26. Realistic: a bounded pipeline

Three async stages connected by bounded channels: fetch → parse → store. Each stage is a task; the bounded channels keep memory flat, and closing flows down the pipeline when the first stage finishes.

Rust

use std::time::Duration;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (raw_tx, mut raw_rx) = mpsc::channel::<String>(2);
    let (rec_tx, mut rec_rx) = mpsc::channel::<(String, f64)>(2);

    let fetch = tokio::spawn(async move {
        for line in ["BTC,64000.5", "ETH,oops", "SOL,145.2", "ADA,0.45"] {
            tokio::time::sleep(Duration::from_millis(5)).await; // network
            raw_tx.send(line.to_string()).await.unwrap();
        }
    });

    let parse = tokio::spawn(async move {
        let mut errors = 0;
        while let Some(line) = raw_rx.recv().await {
            match line.split_once(',').and_then(|(s, p)| Some((s.to_string(), p.parse::<f64>().ok()?))) {
                Some(rec) => rec_tx.send(rec).await.unwrap(),
                None => errors += 1,
            }
        }
        errors
    });

    let mut stored = Vec::new();
    while let Some((symbol, price)) = rec_rx.recv().await {
        stored.push(format!("{symbol}={price}"));
    }
    fetch.await.unwrap();
    println!("stored {stored:?}, parse errors: {}", parse.await.unwrap());
}

Output

stored ["BTC=64000.5", "SOL=145.2", "ADA=0.45"], parse errors: 1

27. Choosing a primitive

You want to...	Use
send work to a task, with backpressure	`mpsc::channel(n)`
get one answer back	`oneshot`
notify every subscriber of every event	`broadcast`
share the latest state (config, status, shutdown)	`watch`
wake a task, no data	`Notify`
protect data, no `.await` while locked	`std::sync::Mutex` / `RwLock`
protect data across `.await`	`tokio::sync::Mutex` / `RwLock`
limit concurrency	`Semaphore` (Tokio Patterns)
initialize once, asynchronously	`tokio::sync::OnceCell`
talk between threads and tasks	Tokio channels with `blocking_send` / `blocking_recv`

Summary

In async code, prefer tokio::sync channels: waiting is an .await, not a blocked thread.
mpsc is bounded (backpressure; try_send to shed load); oneshot carries one reply and reports cancellation; broadcast fans out to every subscriber (slow ones lag); watch keeps only the latest value.
A std::sync::Mutex is fine, and faster, as long as the guard is dropped before any .await; the compiler rejects spawned futures that hold it across one.
Use tokio::sync::Mutex/RwLock only when a lock must be held across .await.
Notify wakes tasks; OnceCell initializes once with an async function; blocking_send/blocking_recv bridge to plain threads.
Actors (mpsc + oneshot) and pub/sub (broadcast) keep state owned by one task, with no locks.

Next: Async Traits: async methods in traits, and what changes when you need dyn Trait.

1. Tokio mpsc: a bounded async queue

2. Multiple producers

3. Backpressure: a slow consumer slows the producer

4. try_send: shed load instead of waiting

5. send_timeout and unbounded channels

6. oneshot: exactly one value

7. oneshot tells you when the other side gave up

8. Request-reply over mpsc + oneshot

9. broadcast: every subscriber gets every message

10. Slow broadcast subscribers lag

11. watch: always the latest value

12. watch as a shutdown signal

13. Notify: a wake-up without data

14. A std::sync::MutexGuard across .await won't compile in spawn

15. Fix 1: keep std::sync::Mutex, release before .await

16. Fix 2: tokio::sync::Mutex when you must hold the lock across .await

17. tokio::sync::RwLock

18. OnceCell with an async initializer

19. Bridging sync and async code

20. The async actor pattern

21. Several workers sharing one receiver

22. Realistic: a chat room with broadcast

23. Realistic: hot-reloading configuration

24. Realistic: batching writes with a flush interval

25. Realistic: an event bus with typed subscribers

26. Realistic: a bounded pipeline

27. Choosing a primitive

Summary

1. Tokio `mpsc`: a bounded async queue

4. `try_send`: shed load instead of waiting

5. `send_timeout` and unbounded channels

6. `oneshot`: exactly one value

7. `oneshot` tells you when the other side gave up

8. Request-reply over `mpsc` + `oneshot`

9. `broadcast`: every subscriber gets every message

10. Slow `broadcast` subscribers lag

11. `watch`: always the latest value

12. `watch` as a shutdown signal

13. `Notify`: a wake-up without data

14. A `std::sync::MutexGuard` across `.await` won't compile in `spawn`

15. Fix 1: keep `std::sync::Mutex`, release before `.await`

16. Fix 2: `tokio::sync::Mutex` when you must hold the lock across `.await`

17. `tokio::sync::RwLock`

18. `OnceCell` with an async initializer

22. Realistic: a chat room with `broadcast`