tokio::sync::mpsc
- Rust course
- Advanced
- Lesson 23/81
- 4 min
Learn async channels and how they differ from sync mpsc.
tokio::sync::mpsc
Tokio's channel is for async tasks: recv().await waits without blocking the thread. The channel is bounded: when capacity is full, send().await waits until room frees up. This is called backpressure, and it keeps a fast producer from filling memory:
let (tx, mut rx) = tokio::sync::mpsc::channel(2); // capacity 2
tokio::spawn(async move {
for i in 1..=5 {
tx.send(i).await.unwrap(); // waits if full
}
});
while let Some(n) = rx.recv().await {
println!("{n}");
}std counterpart: sync_channel
In the standard library a bounded channel is built with sync_channel(capacity); send to a full channel blocks the thread:
let (tx, rx) = mpsc::sync_channel(2);
thread::spawn(move || {
for i in 1..=5 {
tx.send(i).unwrap(); // waits if 2 messages are in the channel
}
});
let toplam: i32 = rx.iter().sum();sync_channel(0) is a "rendezvous" channel: the sender waits until the receiver takes the message.
Practice Backpressure
Open a sync_channel with capacity 2. Let a producer thread send the numbers 1 to 5; let the main thread sum them with rx.iter().sum() and print.
Starter code
use std::sync::mpsc;
use std::thread;
fn main() {
// buraya yaz
}Sample solution
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::sync_channel(2);
thread::spawn(move || {
for i in 1..=5 {
tx.send(i).unwrap();
}
});
let toplam: i32 = rx.iter().sum();
println!("{}", toplam);
}Expected output
15
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 In an unbounded channel, what happens if the producer is much faster than the consumer?
- The producer slows down automatically
- Messages are dropped
- Panic
- Messages pile up and memory use grows without limit Correct answer
Explanation: std::sync::mpsc::channel is unbounded; the bounded version is sync_channel(n).
2 Bounded channel
let (tx, rx) = mpsc::____(2);let (tx, rx) = mpsc::sync_channel(2);Explanation: sync_channel(n) opens a channel holding at most n messages.
3 Channel and behaviour
| Term | Meaning |
|---|---|
mpsc::channel() |
Unbounded |
mpsc::sync_channel(2) |
Capacity 2 |
sync_channel(0) |
Rendezvous |
tokio mpsc::channel(n) |
Async, bounded |