🎓 A1Learner
🔍
0 of 45 lessons completed

Rust Tokio Channels

A Tokio channel is a pipe that passes values between tasks. A waiting task pauses instead of blocking a thread.

A bounded channel holds n messages

mpsc::channel(2) builds a pipe that holds at most 2 waiting messages:

let (tx, mut rx) = mpsc::channel::<String>(2); // holds at most 2 waiting messages
tx.send(String::from("hello")).await.unwrap();
println!("got: {}", rx.recv().await.unwrap());

A full pipe makes senders wait

That waiting is backpressure: a slow receiver slows its senders, by design. The second send below cannot finish until the receiver takes the first message:

let (tx, mut rx) = mpsc::channel::<&str>(1);
tx.send("one").await.unwrap(); // pipe full: room for no more
let sender = tokio::spawn(async move {
    tx.send("two").await.unwrap(); // waits for a free slot
    println!("second send finished");
});
sleep(Duration::from_millis(20)).await;
let first = rx.recv().await.unwrap(); // frees the slot
sender.await.unwrap();
println!("received: {first}");

An unbounded channel never waits

send is not async here — it cannot wait. The price: a fast producer can pile up memory without limit:

let (tx, mut rx) = mpsc::unbounded_channel::<String>();
tx.send(String::from("no waiting")).unwrap(); // send here is not async
println!("got: {}", rx.recv().await.unwrap());

oneshot carries exactly one value

One sender, one receiver, one value — perfect for "send me the result when you are done":

let (reply_tx, reply_rx) = oneshot::channel::<usize>();
tokio::spawn(async move {
    let _ = reply_tx.send(2); // exactly one value, ever
});
println!("worker handled {} messages", reply_rx.await.unwrap());

Many senders, one channel

Clone a Sender to give many tasks the same pipe. The channel closes only when every sender has been dropped:

let (tx, mut rx) = mpsc::channel::<&str>(4);
let tx2 = tx.clone(); // a second sender into the same channel
tokio::spawn(async move { tx2.send("from the clone").await.unwrap(); });
tx.send("from the original").await.unwrap();
drop(tx); // and when tx2 drops too, the channel closes

Example

// Cargo.toml: tokio = { version = "1", features = ["full"] }
use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<String>(2);
    let (reply_tx, reply_rx) = oneshot::channel::<usize>();

    tokio::spawn(async move {
        let mut count = 0;
        while let Some(msg) = rx.recv().await {
            count += 1;
            println!("got: {}", msg);
        }
        let _ = reply_tx.send(count);
    });

    tx.send("hello".to_string()).await.unwrap();
    tx.send("tokio".to_string()).await.unwrap();
    drop(tx); // close the channel, so the worker can finish

    println!("worker handled {} messages", reply_rx.await.unwrap());
}

Key points

  • mpsc means multi-producer, single-consumer: many senders, one receiver
  • A Tokio channel passes values between tasks, and a waiting receiver pauses instead of blocking a thread.
  • Ownership of each value travels with the value.

In this example, the program builds two channels. The mpsc channel carries words to a worker task, which counts them as they arrive. Dropping tx closes the channel, so the worker's while let loop ends after the second message. The worker then sends the count back through the oneshot channel, and main prints it: 2 messages handled.

Note

Awaiting recv() returns None when the channel is closed and empty — that is your clean shutdown signal.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Rust Playground

Exercise

Why choose a bounded mpsc channel over an unbounded one?