mpsc Channels

A channel moves values from one task to another. Instead of sharing data behind a lock, a task sends a message, and another task receives it. mpsc stands for multi-producer, single-consumer: many Sender handles, one Receiver. This lesson covers the bounded channel, its closing rules, and the APIs around it.

API	Purpose
`mpsc::channel(n)`	a bounded channel that buffers up to `n` messages
`send`, `recv`	async send (waits when full) and receive (waits when empty)
`Sender::clone`	more producers for the same channel
`try_send`, `try_recv`	operations that never wait
`reserve`, `Permit`, `try_reserve`	claim space first, send later
`send_timeout`	wait for space, but only for a limited time
`close`, `closed`, `is_closed`	closing a channel and detecting it
`unbounded_channel`	a channel without a limit
`blocking_send`, `blocking_recv`	use from threads outside the runtime
`WeakSender`	a sender handle that does not keep the channel open
`recv_many`, `len`, `capacity`	batch receive and inspection

Examples with start_paused = true use #[tokio::main(flavor = "current_thread", start_paused = true)]: one thread and a virtual clock that jumps forward when every task waits on a timer. It needs the test-util feature and is explained in Time: sleep, timeout, and interval.

Example 1: Send and receive

mpsc::channel(4) returns a (Sender, Receiver) pair. The number is the capacity: how many messages the channel can hold. send(value).await puts a message in. recv().await takes the oldest one out. Messages arrive in the order in which they were sent.

Rust

use tokio::sync::mpsc;

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

    tx.send("first").await.unwrap();
    tx.send("second").await.unwrap();
    tx.send("third").await.unwrap();

    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);
}

Output

Some("first")
Some("second")
Some("third")

recv needs &mut rx, because there is one receiver. recv returns Option<T>: Some(value) for a message, and None when the channel is closed and empty (Example 2).

Example 2: A producer task and a consumer loop

A typical use is one task that produces messages and another that consumes them. The consumer loop while let Some(msg) = rx.recv().await ends when recv returns None, which happens after every sender is dropped and the buffer is empty.

Rust

use tokio::sync::mpsc;

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

    let producer = tokio::spawn(async move {
        for i in 1..=5 {
            tx.send(i * i).await.unwrap();
        }
        // tx is dropped here: the channel starts closing
    });

    let mut total = 0;
    while let Some(value) = rx.recv().await {
        println!("received {value}");
        total += value;
    }
    println!("channel closed, total = {total}");

    producer.await.unwrap();
}

Output

received 1
received 4
received 9
received 16
received 25
channel closed, total = 55

The loop needs no special stop message. Dropping the last Sender ends it. This is the normal way to say "no more work".

Example 3: Closing rule 1: buffered messages survive the senders

The module documentation says: "Once all senders have been dropped and any remaining buffered values have been received, Receiver::recv returns None." Messages that are already in the buffer stay available after the last sender is dropped.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(4);
    tx.send(1).await.unwrap();
    tx.send(2).await.unwrap();
    drop(tx);

    println!("is_closed after dropping the sender: {}", rx.is_closed());
    println!("len = {}", rx.len());
    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);
}

Output

is_closed after dropping the sender: true
len = 2
Some(1)
Some(2)
None
None

recv returns Some(1), Some(2), and then None for every later call. is_closed is already true while two messages are still buffered: "closed" means that no new messages can arrive.

Example 4: Closing rule 2: a live sender keeps recv waiting

While any Sender exists, recv on an empty channel waits. A common bug is to keep the original sender in main after cloning it for the tasks. The receiver then never sees None.

Rust

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

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

    for id in 0..2 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(id).await.unwrap();
        });
    }
    println!("sender count = {}", rx.sender_strong_count());

    println!("{:?}", rx.recv().await);
    println!("{:?}", rx.recv().await);

    // The original `tx` is still alive, so recv keeps waiting.
    let third = timeout(Duration::from_millis(100), rx.recv()).await;
    println!("third recv timed out: {}", third.is_err());
    println!("sender count = {}", rx.sender_strong_count());

    drop(tx);
    println!("after dropping the original: {:?}", rx.recv().await);
}

Output

sender count = 3
Some(0)
Some(1)
third recv timed out: true
sender count = 1
after dropping the original: None

Both tasks finished, so their clones were dropped. The original tx alone kept the channel open. Dropping it produced None. The fix in real programs is to move the original into the last task, or drop it before the receive loop.

Example 5: Closing rule 3: a dropped receiver makes send fail

If the Receiver is dropped, send returns Err(SendError(value)). The value comes back in the error, so nothing is lost silently. The Sender::closed future completes when the receiver is gone, which lets a producer stop early.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<String>(2);
    println!("is_closed before: {}", tx.is_closed());

    drop(rx);
    tx.closed().await; // completes at once: the receiver is gone
    println!("is_closed after: {}", tx.is_closed());

    match tx.send("unsent".to_string()).await {
        Ok(()) => println!("sent"),
        Err(e) => println!("send failed: {e}, value back = {:?}", e.0),
    }
}

Output

is_closed before: false
is_closed after: true
send failed: channel closed, value back = "unsent"

Example 6: The receiver drops unread messages

When the Receiver is dropped, every unread message is dropped with it. The module documentation says the messages "will be drained from the channel and dropped". The example uses a type that prints when it is dropped.

Rust

use tokio::sync::mpsc;

struct Job(u32);

impl Drop for Job {
    fn drop(&mut self) {
        println!("dropping job {}", self.0);
    }
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(4);
    tx.send(Job(1)).await.unwrap();
    tx.send(Job(2)).await.unwrap();
    tx.send(Job(3)).await.unwrap();

    let first = rx.recv().await.unwrap();
    println!("processing job {}", first.0);
    drop(first);

    println!("dropping the receiver");
    drop(rx);
    println!("receiver gone");
}

Output

processing job 1
dropping job 1
dropping the receiver
dropping job 2
dropping job 3
receiver gone

Job 1 was received and processed, then dropped by the program. Jobs 2 and 3 were still in the channel, and they were dropped when the receiver was. Work that must not be lost needs the clean shutdown in the next example.

Example 7: Closing rule 4: close, then drain

Receiver::close stops new messages but keeps the buffered ones. The documentation calls this a "clean" shutdown: close the receiver, read until None, then drop it. After close, send and try_send fail with a closed error.

Rust

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

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(4);
    tx.send("a").await.unwrap();
    tx.send("b").await.unwrap();

    rx.close();
    println!("is_closed after close: {}", rx.is_closed());

    match tx.try_send("c") {
        Err(TrySendError::Closed(v)) => println!("try_send({v}) failed: closed"),
        other => println!("unexpected: {other:?}"),
    }
    println!("send: {:?}", tx.send("d").await);

    while let Some(msg) = rx.recv().await {
        println!("drained {msg}");
    }
    println!("recv after drain: {:?}", rx.recv().await);
}

Output

is_closed after close: true
try_send(c) failed: closed
send: Err(SendError { .. })
drained a
drained b
recv after drain: None

The two messages sent before close are delivered. The two sent after it are rejected, and the caller gets them back in the error.

Example 8: Closing rule 5: an outstanding permit delays None

The recv documentation says that if close is called while Permit values from before the close are outstanding, "the channel is not considered closed by recv until the permits are released". A permit is space that a sender reserved with reserve (Example 14). The example shows recv waiting for a permit that is still held.

Rust

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

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

    let permit = tx.reserve().await.unwrap();
    rx.close();

    let waiting = timeout(Duration::from_millis(50), rx.recv()).await;
    println!("recv while a permit is held timed out: {}", waiting.is_err());

    drop(permit);
    println!("recv after the permit is released: {:?}", rx.recv().await);
}

Output

recv while a permit is held timed out: true
recv after the permit is released: None

The permit may still turn into a message, so recv cannot answer None yet. Dropping the permit without sending frees the space and lets recv finish.

Example 9: Capacity: try_send, len, and capacity

try_send never waits. When the buffer is full it returns TrySendError::Full(value). The receiver reports len (messages waiting), capacity (free space now), and max_capacity (the size given to channel).

Rust

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

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(2);
    println!("max_capacity = {}, capacity = {}", tx.max_capacity(), tx.capacity());

    println!("{:?}", tx.try_send(1));
    println!("{:?}", tx.try_send(2));
    println!("len = {}, capacity = {}", rx.len(), tx.capacity());

    match tx.try_send(3) {
        Err(TrySendError::Full(v)) => println!("buffer full, got {v} back"),
        other => println!("unexpected: {other:?}"),
    }
    println!("error text: {}", tx.try_send(3).unwrap_err());

    rx.recv().await;
    println!("after one recv: len = {}, capacity = {}", rx.len(), tx.capacity());
    println!("{:?}", tx.try_send(3));
}

Output

max_capacity = 2, capacity = 2
Ok(())
Ok(())
len = 2, capacity = 0
buffer full, got 3 back
error text: no available capacity
after one recv: len = 1, capacity = 1
Ok(())

Capacity goes down by one for each queued message and up by one for each received message. max_capacity does not change.

Example 10: A capacity of zero panics

The buffer must hold at least one message. channel(0) panics. The documentation says "Panics if the buffer capacity is 0, or too large", with the maximum equal to Semaphore::MAX_PERMITS (from Semaphore, Notify, Barrier, and OnceCell).

Rustpanics at runtime

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (_tx, _rx) = mpsc::channel::<u32>(0);
}

Output

thread 'main' panicked at src/main.rs:5:22:
mpsc bounded channel requires buffer > 0

std::sync::mpsc::sync_channel(0) is a rendezvous channel in the standard library. The Tokio channel has no equivalent, and a Tokio buffer of one message is the smallest size.

Example 11: Backpressure: send waits when the buffer is full

Backpressure means that a slow consumer slows the producer down. With a bounded channel, send().await does not complete while the buffer is full. It completes when the receiver takes a message. The producer can never run further ahead than the capacity.

The example uses capacity 2. The consumer takes one message every 10 ms of virtual time. Each line shows the virtual time.

Rust

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, Instant};

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

    let producer = tokio::spawn(async move {
        for i in 0..6 {
            tx.send(i).await.unwrap();
            println!("sent {i} at {} ms", start.elapsed().as_millis());
        }
    });

    let consumer = tokio::spawn(async move {
        while let Some(i) = rx.recv().await {
            println!("  received {i} at {} ms", start.elapsed().as_millis());
            sleep(Duration::from_millis(10)).await; // slow processing
        }
    });

    producer.await.unwrap();
    consumer.await.unwrap();
}

Output

sent 0 at 0 ms
sent 1 at 0 ms
  received 0 at 0 ms
sent 2 at 0 ms
  received 1 at 10 ms
sent 3 at 10 ms
  received 2 at 20 ms
sent 4 at 20 ms
  received 3 at 30 ms
sent 5 at 30 ms
  received 4 at 40 ms
  received 5 at 50 ms

The producer finishes its first sends quickly, then waits. After that it advances at the consumer's pace, one message per 10 ms. Without a limit, the producer would finish at once and the queue would hold the difference.

Example 12: try_send sheds load instead of waiting

Sometimes waiting is wrong, for example when a request handler must answer quickly. try_send tells the caller at once whether the message fit, so the caller can drop the message or report "busy". The example sends a burst of 10 messages while the consumer is not yet running and counts the result.

Rust

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

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

    let mut accepted = 0;
    let mut rejected = Vec::new();
    for i in 0..10 {
        match tx.try_send(i) {
            Ok(()) => accepted += 1,
            Err(TrySendError::Full(v)) => rejected.push(v),
            Err(TrySendError::Closed(_)) => unreachable!(),
        }
    }
    println!("accepted {accepted}, rejected {rejected:?}");

    drop(tx);
    let mut got = Vec::new();
    while let Some(v) = rx.recv().await {
        got.push(v);
    }
    println!("consumer received {got:?}");
}

Output

accepted 4, rejected [4, 5, 6, 7, 8, 9]
consumer received [0, 1, 2, 3]

The channel accepted exactly its capacity. The rejected messages came back to the caller, which decides what to do. Backpressure and Load Shedding develops this into a policy.

Example 13: send_timeout: wait, but not forever

send_timeout(value, duration) waits for space like send, but gives up after duration. On failure the error carries the value back. It has two cases: SendTimeoutError::Timeout(value) and SendTimeoutError::Closed(value). It uses Tokio's timer, so it needs the time feature and a Tokio runtime.

Rust

use std::time::Duration;
use tokio::sync::mpsc::{self, error::SendTimeoutError};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);
    tx.send("first").await.unwrap();

    // The buffer is full and nobody receives.
    match tx.send_timeout("second", Duration::from_millis(100)).await {
        Err(SendTimeoutError::Timeout(v)) => println!("timed out, {v} returned"),
        other => println!("unexpected: {other:?}"),
    }
    let err = tx.send_timeout("second", Duration::from_millis(100)).await.unwrap_err();
    println!("error text: {err}");

    rx.recv().await;
    println!("with space: {:?}", tx.send_timeout("second", Duration::from_millis(100)).await);

    drop(rx);
    match tx.send_timeout("third", Duration::from_millis(100)).await {
        Err(SendTimeoutError::Closed(v)) => println!("closed, {v} returned"),
        other => println!("unexpected: {other:?}"),
    }
}

Output

timed out, second returned
error text: timed out waiting on send operation
with space: Ok(())
closed, third returned

Example 14: reserve and Permit: claim space, then send

send(value) needs the value first and can lose it if the future is cancelled while waiting. reserve().await waits for space without holding a value, and returns a Permit. permit.send(value) then sends without waiting and cannot fail. The documentation notes that dropping a Permit without sending "releases the capacity back to the channel".

The pattern fits work that is expensive to start: reserve first, then do the work only when space is certain.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(2);
    println!("capacity at start = {}", tx.capacity());

    let permit = tx.reserve().await.unwrap();
    println!("capacity with one permit = {}", tx.capacity());
    println!("len with one permit = {}", rx.len());
    println!("try_send while one slot is free: {:?}", tx.try_send("direct".into()));
    println!("try_send when the rest is reserved: {}", tx.try_send("other".into()).is_err());

    permit.send("through the permit".into()); // cannot fail
    println!("len after permit.send = {}", rx.len());

    rx.recv().await; // make room: both slots were in use
    let unused = tx.reserve().await;
    drop(unused); // capacity returns without a message
    println!("capacity after dropping a permit = {}", tx.capacity());

    rx.close();
    while let Some(m) = rx.recv().await {
        println!("received {m}");
    }
}

Output

try_reserve is the non-waiting version, and reserve_owned returns an OwnedPermit that owns the Sender and can move into a task.

Example 15: Fan-in: many producers, one consumer

Clone the Sender once per producer. All messages land in the same queue. The channel keeps the order of each sender, and the order between senders depends on scheduling. The example checks per-sender order.

Rust

use std::collections::HashMap;
use tokio::sync::mpsc;

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

    for producer in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            for seq in 0..4 {
                tx.send((producer, seq)).await.unwrap();
            }
        });
    }
    drop(tx); // only the clones remain

    let mut seen: HashMap<u32, Vec<u32>> = HashMap::new();
    while let Some((producer, seq)) = rx.recv().await {
        seen.entry(producer).or_default().push(seq);
    }

    let mut producers: Vec<_> = seen.keys().copied().collect();
    producers.sort();
    for producer in producers {
        println!("producer {producer} sequence: {:?}", seen[&producer]);
    }
}

Output

producer 0 sequence: [0, 1, 2, 3]
producer 1 sequence: [0, 1, 2, 3]
producer 2 sequence: [0, 1, 2, 3]

Each producer's own sequence is in order. The documentation states this guarantee: "All data sent on Sender will become available on Receiver in the same order as it was sent." The program prints results grouped by producer because the interleaving between producers can differ between runs.

Example 16: try_recv, TryRecvError, and recv_many

try_recv returns at once. Its error has two cases: Empty (nothing now, senders exist) and Disconnected (nothing now, and no sender or permit remains). recv_many(&mut vec, limit) moves up to limit messages into a vector and returns how many it moved. It waits only if the channel is empty and open.

Rust

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

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

    match rx.try_recv() {
        Err(TryRecvError::Empty) => println!("empty, senders alive"),
        other => println!("unexpected: {other:?}"),
    }
    println!("error text: {}", rx.try_recv().unwrap_err());

    for i in 1..=5 {
        tx.send(i).await.unwrap();
    }

    let mut batch = Vec::new();
    let n = rx.recv_many(&mut batch, 3).await;
    println!("recv_many moved {n}: {batch:?}");
    let n = rx.recv_many(&mut batch, 3).await;
    println!("recv_many moved {n}: {batch:?}");

    drop(tx);
    match rx.try_recv() {
        Err(TryRecvError::Disconnected) => println!("disconnected"),
        other => println!("unexpected: {other:?}"),
    }
    println!("error text: {}", rx.try_recv().unwrap_err());
    println!("recv_many on a closed, empty channel: {}", rx.recv_many(&mut batch, 3).await);
}

Output

empty, senders alive
error text: receiving on an empty channel
recv_many moved 3: [1, 2, 3]
recv_many moved 2: [1, 2, 3, 4, 5]
disconnected
error text: receiving on a closed channel
recv_many on a closed, empty channel: 0

Batches reduce the number of wakeups when the consumer can process many messages at once, for example when it writes to a file or a socket.

Example 17: WeakSender does not keep the channel open

A WeakSender comes from Sender::downgrade. It does not count as a sender, so it does not keep the channel open. upgrade returns Some(Sender) while at least one strong Sender exists, and None afterwards. This lets a component hold a way to send later without preventing shutdown.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<u32>(2);
    let weak = tx.downgrade();
    println!("strong = {}, weak = {}", tx.strong_count(), tx.weak_count());

    let upgraded = weak.upgrade().unwrap();
    upgraded.send(7).await.unwrap();
    println!("strong with an upgraded handle = {}", tx.strong_count());
    drop(upgraded);

    drop(tx); // the last strong sender
    println!("received {:?}", rx.recv().await);
    println!("then {:?}", rx.recv().await);
    println!("upgrade after the last sender is gone: {:?}", weak.upgrade().is_some());
}

Output

strong = 1, weak = 1
strong with an upgraded handle = 2
received Some(7)
then None
upgrade after the last sender is gone: false

The weak handle stayed alive the whole time, and the receiver still got None. A strong sender is needed for the channel to stay open.

Example 18: Unbounded channels

mpsc::unbounded_channel() has no capacity. UnboundedSender::send is not async: it never waits and returns Err only when the receiver is gone. That makes it usable from synchronous code and from Drop implementations. The cost is that nothing slows the producer. An unbounded channel grows until memory runs out when the producer is faster than the consumer.

Rust

use tokio::sync::mpsc;

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

    // No consumer runs yet. Every send succeeds at once.
    for i in 0..100_000 {
        tx.send(i).unwrap();
    }
    println!("queued messages = {}", rx.len());

    drop(tx);
    let mut sum = 0u64;
    while let Some(v) = rx.recv().await {
        sum += v;
    }
    println!("sum = {sum}");
}

Output

queued messages = 100000
sum = 4999950000

All 100,000 messages sat in the queue. A bounded channel with the same producer would have made the producer wait after capacity messages. Use an unbounded channel only when the number of messages is bounded by something else, such as a fixed number of producers that each send a fixed number of messages, or when the sender cannot be async.

Example 19: blocking_send and blocking_recv: threads and tasks together

Code that runs on a plain thread, outside async code, cannot .await. blocking_send and blocking_recv block the calling thread instead. The module documentation recommends a bounded Tokio channel in both directions between sync and async code. They panic if called inside an async context.

Rust

use std::thread;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (to_async, mut from_thread) = mpsc::channel::<u32>(2);
    let (to_thread, mut from_async) = mpsc::channel::<u32>(2);

    // A plain OS thread: reads numbers, sends back their squares.
    let worker = thread::spawn(move || {
        while let Some(n) = from_async.blocking_recv() {
            to_async.blocking_send(n * n).unwrap();
        }
    });

    let feeder = tokio::spawn(async move {
        for n in 1..=5 {
            to_thread.send(n).await.unwrap();
        }
        // to_thread dropped: the thread's loop ends
    });

    let mut results = Vec::new();
    while let Some(square) = from_thread.recv().await {
        results.push(square);
    }
    feeder.await.unwrap();
    worker.join().unwrap();
    println!("squares = {results:?}");
}

Output

squares = [1, 4, 9, 16, 25]

The worker thread's loop ends when the feeder drops to_thread. Then the worker returns, drops to_async, and the async receive loop ends. Shutdown propagates through the drops.

Example 20: recv is cancel safe in select!

The documentation for Receiver::recv states: "This method is cancel safe. If recv is used as a branch in tokio::select! and another branch completes first, it is guaranteed that no messages were received on this channel." A message is never taken and then lost when another branch wins.

Rust

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, Instant};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let (tx, mut rx) = mpsc::channel(4);
    let start = Instant::now();

    tokio::spawn(async move {
        sleep(Duration::from_millis(25)).await;
        tx.send("late message").await.unwrap();
    });

    let mut ticks = 0;
    loop {
        tokio::select! {
            msg = rx.recv() => {
                match msg {
                    Some(m) => println!("got '{m}' after {ticks} ticks at {} ms", start.elapsed().as_millis()),
                    None => println!("closed after {ticks} ticks"),
                }
                break;
            }
            _ = sleep(Duration::from_millis(10)) => {
                ticks += 1; // the recv future was dropped and is created again
            }
        }
    }
}

Output

got 'late message' after 2 ticks at 25 ms

Each timer branch that wins drops the pending recv future and creates a new one in the next loop turn. No message is lost across those drops. send is different: cancelling send(value) drops the value, and cancelling loses the place in the queue, per the send documentation. Use reserve when a send inside select! might be cancelled.

Example 21: Choosing a capacity

The documentation offers no rule for the number. The capacity sets three things: the largest burst that is absorbed without waiting, the maximum number of messages that wait in memory, and the amount of work lost when the receiver is dropped. The example measures the first: how many messages of a burst of 20 are accepted with no consumer running.

Rust

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let burst = 20;
    println!("capacity  accepted  rejected");
    for capacity in [1, 4, 16, 32] {
        let (tx, _rx) = mpsc::channel::<u32>(capacity);
        let accepted = (0..burst).filter(|&i| tx.try_send(i).is_ok()).count();
        println!("{capacity:>8}  {accepted:>8}  {:>8}", burst as usize - accepted);
    }
}

Output

capacity  accepted  rejected
       1         1        19
       4         4        16
      16        16         4
      32        20         0

Choice	Effect
Small capacity (1 to a few)	tight coupling, fast backpressure, little memory
Capacity near the burst size	bursts pass without waiting, memory grows with the burst
Very large capacity	delays backpressure, hides a slow consumer, holds many messages if the receiver stops
Unbounded	no backpressure, memory grows without limit

The module documentation adds an implementation detail that "may change in future Tokio releases": messages are stored in blocks of 32 (on 64-bit targets), organized in a linked list, and empty blocks are freed or reused.

Interview points

mpsc::channel(n) is bounded, has one Receiver, and clones of Sender. n must be at least 1.
recv returns None only when all senders are dropped (or close was called) and the buffer is empty. A forgotten sender keeps the receiver waiting.
If the Receiver is dropped, send returns SendError with the value, and unread messages are dropped.
Receiver::close rejects new messages and keeps buffered ones. Outstanding Permits delay the None result.
send().await on a full bounded channel waits. That is backpressure. try_send returns Full(value) at once, and send_timeout waits up to a limit.
Cancelling send drops the value and loses the queue position. reserve followed by Permit::send avoids the loss and cannot fail.
recv is cancel safe. It is safe in select! branches.
Messages from one sender arrive in order. Order between senders is not defined.
An unbounded channel never applies backpressure. Its send is synchronous and its memory use is not limited.
blocking_send and blocking_recv serve plain threads and panic inside async code.
WeakSender does not keep a channel open. upgrade returns None when no strong sender remains.
recv_many drains several messages per wakeup.

Summary

A bounded mpsc channel moves messages between tasks and limits how far the producer can run ahead.
Dropping every Sender ends the consumer loop after the buffer drains. Dropping the Receiver makes send fail and drops unread messages.
close supports a clean shutdown: stop new messages, drain the rest.
try_send, send_timeout, and reserve are the three ways to avoid waiting forever on a full channel.
Unbounded channels trade memory safety for a synchronous send.
blocking_send and blocking_recv connect threads and tasks.
Choose the capacity by burst size, memory limit, and acceptable loss on shutdown.
Further reading: Async Channels and Locks in the Rust course. The next lesson, oneshot, broadcast, and watch, covers the other channel types.