Multiple producers
- Rust course
- Advanced
- Lesson 21/81
- 4 min
Build a multi-producer channel with tx.clone().
Multiple producers
The "multiple producer" part of mpsc: you can duplicate the sender with clone and give each producer a copy. All messages gather at a single receiver (fan-in):
let (tx, rx) = mpsc::channel();
for id in 0..3 {
let tx = tx.clone();
thread::spawn(move || {
for j in 0..2 {
tx.send(format!("üretici {id}: {j}")).unwrap();
}
});
}
drop(tx); // if the main copy isn't dropped, rx never closesMessages from different producers can interleave; only messages from the same producer are ordered among themselves.
Deterministic output
The arrival order of messages changes from run to run. For testable output, messages are collected and sorted, or only order-independent summaries are printed:
let mut hepsi: Vec<String> = rx.iter().collect();
hepsi.sort();
println!("{} {}", hepsi.len(), hepsi[0]);Practice Three producers
Start three producer threads; let each send two messages as üretici ID: J (J = 0, 1). Drop the main sender, collect all messages, sort them; print the message count and the first message separated by a space.
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::channel();
for id in 0..3 {
let tx = tx.clone();
thread::spawn(move || {
for j in 0..2 {
tx.send(format!("üretici {id}: {j}")).unwrap();
}
});
}
drop(tx);
let mut hepsi: Vec<String> = rx.iter().collect();
hepsi.sort();
println!("{} {}", hepsi.len(), hepsi[0]);
}Expected output
6 üretici 0: 0
Questions and answers
Review the lesson questions together with their answers. To test yourself, solve the lesson interactively.
1 Can the receiver be cloned (<code>rx.clone()</code>)?
- Only in tokio
- No; mpsc has a single consumer Correct answer
- Yes
- Only with Arc
Explanation: If many consumers are needed, the receiver is put in an Arc<Mutex<Receiver>>, or crossbeam/tokio broadcast channels are used.
2 Duplicate the sender
for id in 0..3 {
let tx = tx.____();
thread::spawn(move || tx.send(id).unwrap());
}for id in 0..3 {
let tx = tx.clone();
thread::spawn(move || tx.send(id).unwrap());
}Explanation: Sender::clone gives a new sender for the same channel.
3 Concept and meaning
| Term | Meaning |
|---|---|
tx.clone() |
New producer |
fan-in |
Many to one gathering |
drop(tx) |
Close the main copy |
rx.clone() |
Not possible |