Async Channels

22 min read

Channels are how independent async tasks talk to each other without sharing memory directly. If you have reached for EventEmitter, a MessageChannel, an RxJS Subject, or a hand-rolled queue in Node.js, Tokio’s channel family covers the same ground — but with compile-time guarantees about who can send, who can receive, and what happens on shutdown.

Quick Overview

A channel is a typed pipe: one or more producers push values in, one or more consumers pull them out. Tokio ships four flavors — mpsc (multi-producer, single-consumer queue), oneshot (a single value, once), broadcast (every subscriber sees every message), and watch (subscribers see only the latest value). They are the idiomatic way to coordinate tasks, because passing ownership of a value through a channel sidesteps the shared-mutable-state problems that Arc<Mutex<T>> solves at a higher cost.

The crucial contrast for a JavaScript developer: these are async channels whose recv/send operations .await and yield to the runtime instead of blocking a thread. The standard library’s std::sync::mpsc looks similar but blocks the OS thread, which would stall a Tokio worker — so inside async code you almost always want the Tokio versions.

Note: Every runnable Rust snippet on this page was compiled and executed with rustc/cargo 1.96.0 (current stable; 2024 edition). Examples use tokio = { version = "1.52", features = ["full"] }. Rust ships no built-in async runtime — see Tokio Setup.

TypeScript/JavaScript Example

JavaScript has no single “channel” primitive, so the same job is done with a grab-bag of tools. A common pattern is an async queue built on Promises, or an EventEmitter for fan-out. Here is a producer/consumer queue and a pub/sub emitter — the two shapes you will most want to translate:

1
import { EventEmitter } from "node:events";
2

3
// --- Producer/consumer: a hand-rolled async queue (mpsc-ish) ---
4
class AsyncQueue<T> {
5
  private items: T[] = [];
6
  private resolvers: Array<(value: T | null) => void> = [];
7
  private closed = false;
8

9
  send(item: T): void {
10
    if (this.closed) throw new Error("queue closed");
11
    const resolve = this.resolvers.shift();
12
    if (resolve) resolve(item);
13
    else this.items.push(item);
14
  }
15

16
  close(): void {
17
    this.closed = true;
18
    // Wake any pending consumers with `null` to signal end-of-stream.
19
    for (const resolve of this.resolvers) resolve(null);
20
    this.resolvers = [];
21
  }
22

23
  // Resolves to the next item, or null once closed and drained.
24
  recv(): Promise<T | null> {
25
    const item = this.items.shift();
26
    if (item !== undefined) return Promise.resolve(item);
27
    if (this.closed) return Promise.resolve(null);
28
    return new Promise((resolve) => this.resolvers.push(resolve));
29
  }
30
}
31

32
// --- Pub/sub fan-out: every listener sees every event (broadcast-ish) ---
33
const bus = new EventEmitter();
34
bus.on("user.created", (id: number) => console.log(`audit log: ${id}`));
35
bus.on("user.created", (id: number) => console.log(`send welcome email: ${id}`));
36
bus.emit("user.created", 42);

Notice what JavaScript does not give you: there is no static guarantee that exactly one consumer drains the queue, no automatic “the channel is closed because every producer went away,” and EventEmitter is untyped (any payload, any event name). Rust’s channels encode all of that in the type system.

Rust Equivalent

The mpsc channel is the workhorse and maps directly onto the producer/consumer queue above:

1
use tokio::sync::mpsc;
2

3
#[derive(Debug)]
4
struct Job {
5
    id: u32,
6
    payload: String,
7
}
8

9
#[tokio::main]
10
async fn main() {
11
    // Bounded channel: at most 8 messages buffered before send() awaits.
12
    let (tx, mut rx) = mpsc::channel::<Job>(8);
13

14
    // Producer task.
15
    let producer = tokio::spawn(async move {
16
        for id in 1..=3 {
17
            let job = Job {
18
                id,
19
                payload: format!("data-{id}"),
20
            };
21
            // send() is async: it waits if the buffer is full (backpressure).
22
            if tx.send(job).await.is_err() {
23
                eprintln!("receiver dropped, stopping producer");
24
                break;
25
            }
26
        }
27
        // tx dropped here when the task ends -> channel closes.
28
    });
29

30
    // Consumer: recv() yields None once all senders are dropped and the buffer drains.
31
    while let Some(job) = rx.recv().await {
32
        println!("processing job {} with {}", job.id, job.payload);
33
    }
34
    println!("channel closed, all jobs done");
35

36
    producer.await.unwrap();
37
}

Real output:

1
processing job 1 with data-1
2
processing job 2 with data-2
3
processing job 3 with data-3
4
channel closed, all jobs done

Two structural facts fall out of the type system for free: mpsc::Receiver is not Clone, so the compiler guarantees a single consumer; and when the last Sender is dropped, recv() returns None, so the while let loop ends on its own. No closed flag, no sentinel null.

Detailed Explanation

`mpsc`: multi-producer, single-consumer

mpsc::channel::<T>(capacity) returns a (Sender<T>, Receiver<T>) pair. The name encodes the contract: the Sender is Clone (many producers), the Receiver is not (one consumer).

tx.send(value).await moves value into the channel. On a bounded channel it .awaits when the buffer is full — this is backpressure, a built-in flow-control mechanism that JavaScript queues lack by default. It returns Result<(), SendError<T>>; the Err arm hands your value back if the receiver is gone.
rx.recv().await returns Option<T>: Some(value) for each message, then None once all senders are dropped and the buffer is empty.

To get multiple producers, clone the sender — but remember to drop the original so the channel can actually close:

1
use tokio::sync::mpsc;
2

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

7
    // Spawn 3 producers, each gets its own clone of the sender.
8
    for worker_id in 1..=3 {
9
        let tx = tx.clone();
10
        tokio::spawn(async move {
11
            tx.send(format!("worker {worker_id} reporting in"))
12
                .await
13
                .unwrap();
14
        });
15
    }
16

17
    // Drop the original sender so the channel can close once the clones finish.
18
    drop(tx);
19

20
    let mut received = Vec::new();
21
    while let Some(msg) = rx.recv().await {
22
        received.push(msg);
23
    }
24

25
    received.sort(); // task ordering is nondeterministic; sort for a stable print.
26
    for msg in received {
27
        println!("{msg}");
28
    }
29
}

Real output:

1
worker 1 reporting in
2
worker 2 reporting in
3
worker 3 reporting in

Tip: There is also mpsc::unbounded_channel(). Its send() is synchronous (no .await, never waits) because the buffer is unlimited — but that means no backpressure, so a fast producer can grow the queue until you run out of memory. Prefer the bounded channel(capacity) unless you have a specific reason not to.

`oneshot`: exactly one value, once

oneshot::channel::<T>() is for a single reply. The Sender::send method takes self by value, so it can only ever be called once — the type system enforces “exactly one message.”

1
use tokio::sync::oneshot;
2

3
#[tokio::main]
4
async fn main() {
5
    let (tx, rx) = oneshot::channel::<u64>();
6

7
    // Worker computes a single value and sends it back.
8
    tokio::spawn(async move {
9
        let result = (1..=100).sum::<u64>();
10
        // send() takes self by value -> can only be called once. Not async.
11
        let _ = tx.send(result);
12
    });
13

14
    // Await the single reply. Errors only if the sender was dropped without sending.
15
    match rx.await {
16
        Ok(value) => println!("computed sum = {value}"),
17
        Err(_) => println!("worker dropped the sender without replying"),
18
    }
19
}

Real output:

1
computed sum = 5050

Note that the Receiver is itself a future — you rx.await it directly rather than calling .recv(). A oneshot is the Rust analogue of a single Promise whose resolve you hand to another task. Its most powerful use is embedding the reply channel inside a request message, which gives you request/response over an mpsc (see the actor pattern below).

`broadcast`: every subscriber sees every message

broadcast::channel::<T>(capacity) is fan-out: clone-free subscription via tx.subscribe(), and each receiver gets a copy of every value sent after it subscribed. The value type must be Clone.

1
use tokio::sync::broadcast;
2

3
#[tokio::main]
4
async fn main() {
5
    // Capacity 16: each receiver has its own 16-slot ring buffer.
6
    let (tx, _rx) = broadcast::channel::<String>(16);
7

8
    // Two independent subscribers. Each gets EVERY message sent after it subscribed.
9
    let mut rx1 = tx.subscribe();
10
    let mut rx2 = tx.subscribe();
11

12
    let sub1 = tokio::spawn(async move {
13
        while let Ok(event) = rx1.recv().await {
14
            println!("sub1 saw: {event}");
15
        }
16
    });
17
    let sub2 = tokio::spawn(async move {
18
        while let Ok(event) = rx2.recv().await {
19
            println!("sub2 saw: {event}");
20
        }
21
    });
22

23
    // send() returns the number of currently-subscribed receivers.
24
    tx.send("user.created".to_string()).unwrap();
25
    tx.send("user.updated".to_string()).unwrap();
26

27
    // Dropping the sender closes the channel; receivers then get RecvError::Closed.
28
    drop(tx);
29

30
    sub1.await.unwrap();
31
    sub2.await.unwrap();
32
}

Real output:

1
sub1 saw: user.created
2
sub1 saw: user.updated
3
sub2 saw: user.created
4
sub2 saw: user.updated

This is the typed, ownership-aware replacement for EventEmitter. The capacity matters: broadcast uses a ring buffer per receiver, so a slow consumer that falls more than capacity messages behind will lag and skip the oldest messages (covered under Common Pitfalls).

`watch`: only the latest value matters

watch::channel(initial) is for state that changes over time where consumers only care about the current value, not the history — config reloads, a “current health” flag, the latest sensor reading. The sender overwrites; receivers borrow() the newest value.

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

4
#[derive(Clone, Debug, PartialEq)]
5
struct Config {
6
    log_level: String,
7
}
8

9
#[tokio::main]
10
async fn main() {
11
    // watch holds a single latest value; receivers see the most recent one.
12
    let (tx, mut rx) = watch::channel(Config {
13
        log_level: "info".to_string(),
14
    });
15

16
    let watcher = tokio::spawn(async move {
17
        loop {
18
            // changed() resolves when the value is updated since the last observation.
19
            if rx.changed().await.is_err() {
20
                println!("config sender dropped, watcher exiting");
21
                break;
22
            }
23
            // borrow() gives a read guard to the latest value.
24
            let cfg = rx.borrow();
25
            println!("config changed -> log_level = {}", cfg.log_level);
26
        }
27
    });
28

29
    sleep(Duration::from_millis(20)).await;
30
    tx.send(Config { log_level: "debug".to_string() }).unwrap();
31
    sleep(Duration::from_millis(20)).await;
32
    tx.send(Config { log_level: "trace".to_string() }).unwrap();
33
    sleep(Duration::from_millis(20)).await;
34

35
    drop(tx); // closes the channel; watcher's changed() returns Err and it exits.
36
    watcher.await.unwrap();
37
}

Real output:

1
config changed -> log_level = debug
2
config changed -> log_level = trace
3
config sender dropped, watcher exiting

If you send three times before a watcher calls changed(), it only ever sees the last value — intermediate values are coalesced. That is the defining difference from broadcast, which delivers every message.

Warning: Do not hold the guard returned by watch::Receiver::borrow() across an .await. It is a read lock; keeping it alive while suspended can block senders. Copy out what you need, drop the guard, then await. The same hazard applies to tokio::sync::RwLock — see sync-primitives.md.

Request/response: `oneshot` inside an `mpsc` message (the actor pattern)

This is the single most useful composition. One task owns some state; everyone else talks to it by sending a request that carries its own private oneshot reply channel:

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

3
/// A request carries a oneshot sender for its individual reply.
4
struct Request {
5
    key: String,
6
    reply_to: oneshot::Sender<Option<String>>,
7
}
8

9
#[tokio::main]
10
async fn main() {
11
    let (tx, mut rx) = mpsc::channel::<Request>(32);
12

13
    // The "actor": owns the state, processes requests one at a time.
14
    let actor = tokio::spawn(async move {
15
        let store = std::collections::HashMap::from([
16
            ("alice".to_string(), "admin".to_string()),
17
            ("bob".to_string(), "user".to_string()),
18
        ]);
19
        while let Some(req) = rx.recv().await {
20
            let answer = store.get(&req.key).cloned();
21
            // Reply on this request's private oneshot channel.
22
            let _ = req.reply_to.send(answer);
23
        }
24
    });
25

26
    // A client sends a request and awaits its dedicated reply.
27
    let role = lookup(&tx, "alice").await;
28
    println!("alice -> {role:?}");
29
    let missing = lookup(&tx, "carol").await;
30
    println!("carol -> {missing:?}");
31

32
    drop(tx);
33
    actor.await.unwrap();
34
}
35

36
async fn lookup(tx: &mpsc::Sender<Request>, key: &str) -> Option<String> {
37
    let (reply_to, reply_rx) = oneshot::channel();
38
    tx.send(Request {
39
        key: key.to_string(),
40
        reply_to,
41
    })
42
    .await
43
    .expect("actor is alive");
44
    reply_rx.await.expect("actor replied")
45
}

Real output:

1
alice -> Some("admin")
2
carol -> None

Because the actor is the only task that touches store, there is no lock and no data race — the channel serializes access. This is how you replace Arc<Mutex<HashMap<..>>> with message passing.

Key Differences

Concern	JavaScript	Rust / Tokio
Producer/consumer queue	hand-rolled, or a userland library	`tokio::sync::mpsc` (built in)
One-shot reply	a single `Promise` you resolve	`tokio::sync::oneshot`
Pub/sub fan-out	`EventEmitter`, RxJS `Subject`	`tokio::sync::broadcast`
”Latest value” state	a variable + manual notify	`tokio::sync::watch`
Type safety of payloads	none (`any`)	fully typed `T`, checked at compile time
Who can receive	not enforced	`mpsc::Receiver` is not `Clone` → single consumer guaranteed
Backpressure	manual	bounded `mpsc::send().await` waits when full
”Channel closed” signal	manual flag / sentinel	automatic when all senders or the receiver drop
Blocking vs async	event loop never blocks	Tokio channels `.await`; `std::sync::mpsc` blocks the thread

`std::sync::mpsc` vs `tokio::sync::mpsc`

The standard library has its own mpsc that looks like Tokio’s, and reaching for it inside async code is a classic mistake.

1
use std::sync::mpsc; // standard library, NOT tokio
2
use std::thread;
3

4
fn main() {
5
    // std channel: multi-producer, single-consumer, BLOCKING (no async).
6
    let (tx, rx) = mpsc::channel::<i32>();
7

8
    for n in 0..3 {
9
        let tx = tx.clone();
10
        thread::spawn(move || {
11
            tx.send(n * 10).unwrap(); // blocks the OS thread if needed
12
        });
13
    }
14
    drop(tx);
15

16
    let mut total = 0;
17
    // recv() blocks the thread until a value arrives or all senders drop.
18
    for value in rx {
19
        total += value;
20
    }
21
    println!("total = {total}");
22
}

Real output:

1
total = 30

The decision rule:

	`std::sync::mpsc`	`tokio::sync::mpsc`
`recv`	blocks the OS thread	`.await`s, yields to the runtime
`send`	synchronous	`.await` on bounded; sync on unbounded
Use it for	plain threads (05-ownership territory)	async tasks under a runtime
Inside a Tokio task?	no — it stalls a worker thread	yes

Note: If you genuinely must call a blocking std channel from async (e.g. bridging a synchronous library), do it on a thread that is allowed to block via tokio::task::spawn_blocking — see spawning-tasks.md.

Common Pitfalls

1. Forgetting to drop the extra `Sender`, so `recv()` never returns `None`

recv() returns None only when every sender is dropped. If you clone a sender for workers but keep the original alive in scope, the consumer loop hangs forever waiting for more. The fix is the explicit drop(tx) shown in the multi-producer example above (or letting the original go out of scope before the loop).

2. Trying to clone the `mpsc::Receiver`

mpsc is single-consumer by design. Cloning the receiver to fan work out to several tasks does not compile:

1
use tokio::sync::mpsc;
2

3
#[tokio::main]
4
async fn main() {
5
    let (_tx, rx) = mpsc::channel::<i32>(8);
6
    // mpsc is multi-producer, SINGLE-consumer: the Receiver is not Clone.
7
    let rx2 = rx.clone(); // does not compile (error[E0599]: no method named `clone`)
8
    drop((rx, rx2));
9
}

Real compiler error:

1
error[E0599]: no method named `clone` found for struct `tokio::sync::mpsc::Receiver` in the current scope
2
 --> src/bin/err_clone_rx.rs:7:18
3
  |
4
7 |     let rx2 = rx.clone();
5
  |                  ^^^^^
6
  |
7
help: there is a method `close` with a similar name

To share one receiver across several worker tasks, wrap it in Arc<tokio::sync::Mutex<Receiver<T>>> (the worker pool below does exactly that), or restructure so each worker has its own channel.

3. Calling `oneshot::Sender::send` twice

send consumes the sender, so a second call references a moved value:

1
use tokio::sync::oneshot;
2

3
#[tokio::main]
4
async fn main() {
5
    let (tx, _rx) = oneshot::channel::<i32>();
6
    tx.send(1).unwrap(); // send consumes `tx` by value
7
    tx.send(2).unwrap(); // does not compile (error[E0382]: use of moved value)
8
}

Real compiler error (abridged):

1
error[E0382]: use of moved value: `tx`
2
   --> src/bin/err_send_twice.rs:7:5
3
    |
4
  5 |     let (tx, _rx) = oneshot::channel::<i32>();
5
    |          -- move occurs because `tx` has type `tokio::sync::oneshot::Sender<i32>`, which does not implement the `Copy` trait
6
  6 |     tx.send(1).unwrap(); // send consumes `tx` by value
7
    |        ------- `tx` moved due to this method call
8
  7 |     tx.send(2).unwrap(); // ERROR: tx already moved
9
    |     ^^ value used here after move
10
    |
11
note: `tokio::sync::oneshot::Sender::<T>::send` takes ownership of the receiver `self`, which moves `tx`

If you need to send more than once, you wanted mpsc (a stream of messages) or watch (the latest value), not oneshot.

4. A slow `broadcast` receiver lagging and skipping messages

broadcast keeps a fixed-size ring buffer per receiver. A consumer that falls behind by more than the capacity loses the oldest messages and recv() returns Err(RecvError::Lagged(n)). Unlike a closed channel, lag is recoverable — you keep receiving after it:

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

3
#[tokio::main]
4
async fn main() {
5
    // Tiny capacity of 2 to force lag.
6
    let (tx, mut rx) = broadcast::channel::<u32>(2);
7

8
    // Send 4 values before the slow receiver reads anything.
9
    for n in 1..=4 {
10
        tx.send(n).unwrap();
11
    }
12

13
    // The first two values were overwritten in the ring buffer -> Lagged(2).
14
    loop {
15
        match rx.recv().await {
16
            Ok(value) => println!("received {value}"),
17
            Err(RecvError::Lagged(skipped)) => {
18
                println!("lagged: skipped {skipped} messages");
19
            }
20
            Err(RecvError::Closed) => {
21
                println!("channel closed");
22
                break;
23
            }
24
        }
25
    }
26
}

Real output:

1
lagged: skipped 2 messages
2
received 3
3
received 4

Always handle Lagged explicitly: log it, increase the capacity, or speed up the consumer. Treating it as a fatal error is usually wrong.

5. Using `std::sync::mpsc::recv()` inside an async task

Because std::sync::mpsc::recv() blocks the calling OS thread, calling it directly in a Tokio task parks one of the runtime’s worker threads. On a single-threaded runtime this deadlocks; on a multi-threaded runtime it silently degrades throughput. There is no compiler error — it just behaves badly. Use tokio::sync::mpsc in async code, or hop to spawn_blocking for genuinely blocking work.

Best Practices

Pick the channel by communication shape, not habit. Stream of work → mpsc; single reply → oneshot; fan-out events → broadcast; latest-value state → watch.
Prefer bounded mpsc. Backpressure protects you from unbounded memory growth. Choose a capacity, and treat a full channel as a real signal about load.
Drop senders you do not need. A lingering Sender clone keeps the channel open and hangs the consumer. Let clones move into tasks; drop the original explicitly when the borrow checker keeps it alive.
Always handle broadcast’s Lagged. A slow subscriber should degrade gracefully, not crash.
Use message passing instead of shared state when you can. The actor pattern (an mpsc of requests, each carrying a oneshot reply) replaces Arc<Mutex<T>> and removes a whole class of locking bugs. When you do need shared state, see arc-mutex-pattern.md.
Keep std::sync::mpsc for plain threads. Inside async, default to the Tokio family.
Combine channels with select! for shutdown. A watch<bool> shutdown flag plus tokio::select! lets a task react to either work or a stop signal — see select-join.md.

Real-World Example

A production graceful-shutdown worker pool brings the whole family together: an mpsc carries jobs (with backpressure), each job carries a oneshot for its individual reply, and a watch<bool> broadcasts the shutdown signal to every worker. The single mpsc::Receiver is shared across workers behind an Arc<tokio::sync::Mutex<_>>, and tokio::select! lets each worker react to either a new job or the shutdown flag.

1
use tokio::sync::{mpsc, oneshot, watch};
2
use tokio::time::{Duration, sleep};
3

4
/// A unit of work submitted to the pool, carrying a private reply channel.
5
struct Task {
6
    id: u32,
7
    reply: oneshot::Sender<String>,
8
}
9

10
#[tokio::main]
11
async fn main() {
12
    // Jobs flow worker-ward over a bounded mpsc (backpressure at 64).
13
    let (job_tx, job_rx) = mpsc::channel::<Task>(64);
14
    // A single shutdown flag broadcast to every worker via watch.
15
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
16

17
    // Spawn a small pool. mpsc is single-consumer, so we share the Receiver
18
    // behind an async Mutex; each worker pulls the next available job.
19
    let job_rx = std::sync::Arc::new(tokio::sync::Mutex::new(job_rx));
20
    let mut workers = Vec::new();
21
    for worker_id in 0..3 {
22
        let job_rx = job_rx.clone();
23
        let mut shutdown_rx = shutdown_rx.clone();
24
        workers.push(tokio::spawn(async move {
25
            loop {
26
                let task = {
27
                    let mut guard = job_rx.lock().await;
28
                    tokio::select! {
29
                        // Stop promptly when the shutdown flag flips to true.
30
                        _ = shutdown_rx.changed() => {
31
                            if *shutdown_rx.borrow() { break; }
32
                            continue;
33
                        }
34
                        maybe = guard.recv() => match maybe {
35
                            Some(task) => task,
36
                            None => break, // all senders dropped
37
                        },
38
                    }
39
                };
40
                // Simulate work, then answer this task's private oneshot.
41
                sleep(Duration::from_millis(5)).await;
42
                let _ = task
43
                    .reply
44
                    .send(format!("task {} handled by worker {worker_id}", task.id));
45
            }
46
        }));
47
    }
48

49
    // Submit 5 jobs and collect their individual replies.
50
    let mut replies = Vec::new();
51
    for id in 0..5 {
52
        let (reply, reply_rx) = oneshot::channel();
53
        job_tx.send(Task { id, reply }).await.unwrap();
54
        replies.push(reply_rx);
55
    }
56

57
    let mut results = Vec::new();
58
    for rx in replies {
59
        results.push(rx.await.unwrap());
60
    }
61
    results.sort();
62
    for line in &results {
63
        println!("{line}");
64
    }
65

66
    // Graceful shutdown: stop accepting work, signal workers, await them.
67
    drop(job_tx);
68
    shutdown_tx.send(true).unwrap();
69
    for w in workers {
70
        w.await.unwrap();
71
    }
72
    println!("pool drained and shut down cleanly");
73
}

Real output (one representative run — which worker handles which task varies between runs, but the task ordering is deterministic because we sort by task id):

1
task 0 handled by worker 0
2
task 1 handled by worker 1
3
task 2 handled by worker 2
4
task 3 handled by worker 2
5
task 4 handled by worker 0
6
pool drained and shut down cleanly

Note: This pattern — bounded mpsc for work, oneshot for replies, watch for shutdown, select! to combine them — is the backbone of countless Rust services. It composes from small, independently testable pieces and never touches a shared lock for the business data.

Exercises

Exercise 1: Sum over an `mpsc`

Difficulty: Easy

Objective: Wire up a basic producer/consumer with a bounded mpsc channel.

Instructions:

Create a bounded mpsc::channel::<i32>(16).
Spawn a producer task that sends the numbers 1..=5.
In main, drain the receiver with while let Some(n) = rx.recv().await and accumulate a sum.
Make sure the consumer loop actually terminates, then print the total.

Solution

1
use tokio::sync::mpsc;
2

3
#[tokio::main]
4
async fn main() {
5
    let (tx, mut rx) = mpsc::channel::<i32>(16);
6

7
    let producer = tokio::spawn(async move {
8
        for n in 1..=5 {
9
            tx.send(n).await.unwrap();
10
        }
11
        // tx moved into the task and dropped here -> channel closes,
12
        // so rx.recv() will return None and the loop below ends.
13
    });
14

15
    let mut sum = 0;
16
    while let Some(n) = rx.recv().await {
17
        sum += n;
18
    }
19
    producer.await.unwrap();
20
    println!("sum = {sum}");
21
}

Output:

1
sum = 15

The key insight: the producer task owns tx after the async move, so when the task finishes the sender is dropped and recv() returns None.

Exercise 2: One-shot square

Difficulty: Medium

Objective: Use oneshot to get a single computed value back from a spawned task, wrapped in a reusable async function.

Instructions:

Write async fn compute(input: u64) -> u64.
Inside it, create a oneshot channel, spawn a task that computes input * input and sends it, and .await the receiver.
Call compute(9) from main and print the result.

Solution

1
use tokio::sync::oneshot;
2

3
async fn compute(input: u64) -> u64 {
4
    let (tx, rx) = oneshot::channel();
5
    tokio::spawn(async move {
6
        let _ = tx.send(input * input);
7
    });
8
    // The Receiver is itself a Future; await it directly (no .recv()).
9
    rx.await.expect("worker replied")
10
}
11

12
#[tokio::main]
13
async fn main() {
14
    let result = compute(9).await;
15
    println!("9 squared = {result}");
16
}

Output:

1
9 squared = 81

rx.await yields Result<T, RecvError>; expect is fine here because we know the task always sends before dropping its sender.

Exercise 3: Fan-out with `broadcast`

Difficulty: Medium–Hard

Objective: Build a pub/sub where two independent subscribers each collect every message, then verify they saw the same sequence.

Instructions:

Create a broadcast::channel::<i32>(8).
Subscribe twice (tx.subscribe()), and spawn a task per subscriber that loops on recv().await, pushing each value into a Vec until the channel closes.
From main, send 1..=3, then drop the sender.
Await both tasks and print each subscriber’s collected Vec. Both must be [1, 2, 3].

Solution

1
use tokio::sync::broadcast;
2

3
#[tokio::main]
4
async fn main() {
5
    let (tx, _) = broadcast::channel::<i32>(8);
6
    let mut a = tx.subscribe();
7
    let mut b = tx.subscribe();
8

9
    let ta = tokio::spawn(async move {
10
        let mut seen = Vec::new();
11
        // recv() returns Err(Closed) once the sender is dropped -> loop ends.
12
        while let Ok(v) = a.recv().await {
13
            seen.push(v);
14
        }
15
        seen
16
    });
17
    let tb = tokio::spawn(async move {
18
        let mut seen = Vec::new();
19
        while let Ok(v) = b.recv().await {
20
            seen.push(v);
21
        }
22
        seen
23
    });
24

25
    for n in 1..=3 {
26
        tx.send(n).unwrap();
27
    }
28
    drop(tx); // close the channel so both receivers stop
29

30
    println!("a saw {:?}", ta.await.unwrap());
31
    println!("b saw {:?}", tb.await.unwrap());
32
}

Output:

1
a saw [1, 2, 3]
2
b saw [1, 2, 3]

Both subscribers must subscribe before any message is sent — broadcast only delivers messages sent after a receiver subscribes. With capacity 8 and only three messages, no lag occurs.

Async Channels

Quick Overview

TypeScript/JavaScript Example

Rust Equivalent

Detailed Explanation

mpsc: multi-producer, single-consumer

oneshot: exactly one value, once

broadcast: every subscriber sees every message

watch: only the latest value matters

Request/response: oneshot inside an mpsc message (the actor pattern)

Key Differences

std::sync::mpsc vs tokio::sync::mpsc

Common Pitfalls

1. Forgetting to drop the extra Sender, so recv() never returns None

2. Trying to clone the mpsc::Receiver

3. Calling oneshot::Sender::send twice

4. A slow broadcast receiver lagging and skipping messages

5. Using std::sync::mpsc::recv() inside an async task

Best Practices

Real-World Example

Further Reading

Exercises

Exercise 1: Sum over an mpsc

Exercise 2: One-shot square

Exercise 3: Fan-out with broadcast

`mpsc`: multi-producer, single-consumer

`oneshot`: exactly one value, once

`broadcast`: every subscriber sees every message

`watch`: only the latest value matters

Request/response: `oneshot` inside an `mpsc` message (the actor pattern)

`std::sync::mpsc` vs `tokio::sync::mpsc`

1. Forgetting to drop the extra `Sender`, so `recv()` never returns `None`

2. Trying to clone the `mpsc::Receiver`

3. Calling `oneshot::Sender::send` twice

4. A slow `broadcast` receiver lagging and skipping messages

5. Using `std::sync::mpsc::recv()` inside an async task

Exercise 1: Sum over an `mpsc`

Exercise 3: Fan-out with `broadcast`