Channels
A channel is an easy way to use many threads that send to one place.它们相当流行，因为它们很容易组合在一起。你可以在Rust中用std::sync::mpsc创建一个channel。mpsc的意思是 "多个生产者，单个消费者"，所以 "many threads sending to one place"。要启动一个通道，你可以使用 channel()。这将创建一个 Sender 和一个 Receiver，它们被绑在一起。你可以在函数签名中看到这一点。
#![allow(unused)] fn main() { // 🚧 pub fn channel<T>() -> (Sender<T>, Receiver<T>) }
所以你要选择一个发送者的名字和一个接收者的名字。通常你会看到像let (sender, receiver) = channel();这样的开头。因为它是泛型函数，如果你只写这个，Rust不会知道类型。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); // ⚠️ }
编译器说:
error[E0282]: type annotations needed for `(std::sync::mpsc::Sender<T>, std::sync::mpsc::Receiver<T>)`
--> src\main.rs:30:30
|
30 | let (sender, receiver) = channel();
| ------------------ ^^^^^^^ cannot infer type for type parameter `T` declared on the function `channel`
| |
| consider giving this pattern the explicit type `(std::sync::mpsc::Sender<T>, std::sync::mpsc::Receiver<T>)`, where
the type parameter `T` is specified
它建议为Sender和Receiver添加一个类型。如果你愿意的话，可以这样做:
use std::sync::mpsc::{channel, Sender, Receiver}; // Added Sender and Receiver here fn main() { let (sender, receiver): (Sender<i32>, Receiver<i32>) = channel(); }
但你不必这样做: 一旦你开始使用Sender和Receiver，Rust就能猜到类型。
所以我们来看一下最简单的使用通道的方法。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); sender.send(5); receiver.recv(); // recv = receive, not "rec v" }
现在编译器知道类型了。sender是Result<(), SendError<i32>>，receiver是Result<i32, RecvError>。所以你可以用.unwrap()来看看发送是否有效，或者使用更好的错误处理。我们加上.unwrap()，也加上println!，看看得到什么。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); sender.send(5).unwrap(); println!("{}", receiver.recv().unwrap()); }
这样就可以打印出5。
channel就像Arc一样，因为你可以克隆它，并将克隆的内容发送到其他线程中。让我们创建两个线程，并将值发送到receiver。这段代码可以工作，但它并不完全是我们想要的。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); let sender_clone = sender.clone(); std::thread::spawn(move|| { // move sender in sender.send("Send a &str this time").unwrap(); }); std::thread::spawn(move|| { // move sender_clone in sender_clone.send("And here is another &str").unwrap(); }); println!("{}", receiver.recv().unwrap()); }
两个线程开始发送，然后我们println!。它可能会打印 Send a &str this time 或 And here is another &str，这取决于哪个线程先完成。让我们创建一个join句柄来等待它们完成。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); let sender_clone = sender.clone(); let mut handle_vec = vec![]; // Put our handles in here handle_vec.push(std::thread::spawn(move|| { // push this into the vec sender.send("Send a &str this time").unwrap(); })); handle_vec.push(std::thread::spawn(move|| { // and push this into the vec sender_clone.send("And here is another &str").unwrap(); })); for _ in handle_vec { // now handle_vec has 2 items. Let's print them println!("{:?}", receiver.recv().unwrap()); } }
这个将打印:
"Send a &str this time"
"And here is another &str"
现在我们不打印，我们创建一个results_vec。
use std::sync::mpsc::channel; fn main() { let (sender, receiver) = channel(); let sender_clone = sender.clone(); let mut handle_vec = vec![]; let mut results_vec = vec![]; handle_vec.push(std::thread::spawn(move|| { sender.send("Send a &str this time").unwrap(); })); handle_vec.push(std::thread::spawn(move|| { sender_clone.send("And here is another &str").unwrap(); })); for _ in handle_vec { results_vec.push(receiver.recv().unwrap()); } println!("{:?}", results_vec); }
现在结果在我们的vec中:["Send a &str this time", "And here is another &str"]。
现在让我们假设我们有很多工作要做，并且想要使用线程。我们有一个大的VEC，里面有1百万个元素，都是0，我们想把每个0都变成1，我们将使用10个线程，每个线程将做十分之一的工作。我们将创建一个新的VEC，并使用.extend()来收集结果。
use std::sync::mpsc::channel; use std::thread::spawn; fn main() { let (sender, receiver) = channel(); let hugevec = vec![0; 1_000_000]; let mut newvec = vec![]; let mut handle_vec = vec![]; for i in 0..10 { let sender_clone = sender.clone(); let mut work: Vec<u8> = Vec::with_capacity(hugevec.len() / 10); // new vec to put the work in. 1/10th the size work.extend(&hugevec[i*100_000..(i+1)*100_000]); // first part gets 0..100_000, next gets 100_000..200_000, etc. let handle =spawn(move || { // make a handle for number in work.iter_mut() { // do the actual work *number += 1; }; sender_clone.send(work).unwrap(); // use the sender_clone to send the work to the receiver }); handle_vec.push(handle); } for handle in handle_vec { // stop until the threads are done handle.join().unwrap(); } while let Ok(results) = receiver.try_recv() { newvec.push(results); // push the results from receiver.recv() into the vec } // Now we have a Vec<Vec<u8>>. To put it together we can use .flatten() let newvec = newvec.into_iter().flatten().collect::<Vec<u8>>(); // Now it's one vec of 1_000_000 u8 numbers println!("{:?}, {:?}, total length: {}", // Let's print out some numbers to make sure they are all 1 &newvec[0..10], &newvec[newvec.len()-10..newvec.len()], newvec.len() // And show that the length is 1_000_000 items ); for number in newvec { // And let's tell Rust that it can panic if even one number is not 1 if number != 1 { panic!(); } } }