Summary - Channels
- Rust course
- Advanced
- Lesson 24/81
- 5 min
Combine mpsc and thread to build a threaded work queue.
Channels summary
mpsc::channel opens an unbounded channel, sync_channel(n) a bounded one. send transfers ownership; Err if there's no receiver. recv waits, try_recv doesn't, recv_timeout waits for a duration; once all senders close, the receiver ends. tx.clone() gives many producers. Async code uses tokio::sync::mpsc. Channels are the clean way to build data flow between stages instead of shared state.
Pipeline
Each stage is a thread; stages are connected with channels. When a stage finishes, its sender is dropped and the closure propagates down the chain to the following stages:
let (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();
thread::spawn(move || for i in 1..=5 { tx1.send(i).unwrap(); });
thread::spawn(move || for n in rx1 { tx2.send(n * n).unwrap(); });
let sonuc: Vec<i32> = rx2.iter().collect();Practice Two-stage pipeline
Build the pipeline above: let the first thread send the numbers 1 to 5, and the second thread forward each number's square to the next channel. Let the main thread collect the results and print them with {:?}.
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 (tx1, rx1) = mpsc::channel();
let (tx2, rx2) = mpsc::channel();
thread::spawn(move || {
for i in 1..=5 {
tx1.send(i).unwrap();
}
});
thread::spawn(move || {
for n in rx1 {
tx2.send(n * n).unwrap();
}
});
let sonuc: Vec<i32> = rx2.iter().collect();
println!("{:?}", sonuc);
}Expected output
[1, 4, 9, 16, 25]
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 In a two-stage pipeline (produce → square → collect), why is the order preserved?
- It isn't preserved
- Thanks to a Mutex
- Each stage has a single thread and channels are FIFO Correct answer
- Rust sorts automatically
Explanation: If several workers are added to a stage, the order guarantee is lost.
2 Connect the stage
thread::spawn(move || for n in ____ {
tx2.send(n * n).unwrap();
});thread::spawn(move || for n in rx1 {
tx2.send(n * n).unwrap();
});Explanation: The second stage listens to the first stage's receiver.
3 Group summary
| Term | Meaning |
|---|---|
channel() |
Unbounded channel |
sync_channel(n) |
Bounded channel |
tx.clone() |
Many producers |
try_recv |
Without waiting |