【Rust自学】16.2. 使用消息传递来跨线程传递数据

16.2 使用消息传递来跨线程传递数据16.2.1. 消息传递有一种很流行而且能保证安全并发的技术叫做消息传递。在这种机制里线程或 Actor通过彼此间发送消息数据来进行通讯。Go 语言有一句名言是这么说的Do not communicate by sharing memory; instead, share memory by communicating.不要用共享内存来通信要用通信来共享内存Go 语言的并发模型体现了这种思想。Rust 也提供了基于消息传递的一种并发方式具体就是使用标准库中的Channel。Go 语言里也有Channel思路差不多。16.2.2. 理解Channel可以将编程中的Channel想象为定向水道例如小溪或河流。如果你把橡皮鸭之类的东西放入河中它会顺流而下到达水道的尽头。通道有两部分发送端和接收端。发送端是将橡皮鸭放入河中的上游位置接收端是橡皮鸭最终到达下游的位置。代码的一部分使用要发送的数据调用发送端上的方法另一部分检查接收端是否有到达的消息。如果发送端或接收端其一消失则称通道已关闭。具体的步骤- 调用发送端的方法发送数据- 接收端会检查和接收到达的数据- 如果发送端、接收端中的任意一端被丢弃了那么Channel就关闭了。16.2.3. 创建channel使用mpsc::channel函数来创建Channel。mpsc表示 multiple producer, single consumer多个生产者、一个消费者表示可以有多个发送端但是只能有一个接收端。调用这个函数返回一个元组有两个元素分别是发送端和接收端。看个例子use std::sync::mpsc; use std::thread; fn main() { let (tx, rx) mpsc::channel(); thread::spawn(move || { let val String::from(hi); tx.send(val).unwrap(); }); let received rx.recv().unwrap(); println!(Got: {received}); }首先使用mpsc::channel函数来创建Channel返回的元组使用模式匹配进行解构分别用tx和rx表示发送端和接收端。接下来创建了一个线程使用move关键字表示发送端tx的所有权被移至分线程内因为线程必须拥有通道发送端的所有权才能往通道里发消息。使用send方法来发送消息返回类型是Result类型如果接收端被丢弃了那么返回值就是Err反之就是Ok。在这里面就简单地使用unwrap进行错误处理即可这样如果接收端被丢弃就会恐慌。接收端有两个方法来获取消息这里使用了recv方法receive的简写。它会一直阻塞这个线程直到有消息被传入为止。消息被包裹在Result类型中有消息就返回Ok反之就是Err一样使用unwrap简单地处理错误即可。输出Got: hi发送端的send方法send方法的参数是想要发送的数据返回Result类型。如果有问题例如接收端已经被丢弃就会返回Err。接收端的方法recv方法阻止当前线程执行直到Channel中有值传来一旦收到值就返回Result类型如果发送端关闭了就会收到Err。try_recv方法不会阻塞当前线程执行立即返回Result类型有数据到达就是Ok变体包裹着传过来的数据否则就返回错误。通常是使用循环调用来检查try_recv的结果。一旦有消息来了就开始处理如果没来那么这时候也可以执行其他指令。16.2.4.channel和所有权转移所有权在消息传递中非常重要它能帮你编写安全、并发的代码。看个例子use std::sync::mpsc; use std::thread; fn main() { let (tx, rx) mpsc::channel(); thread::spawn(move || { let val String::from(hi); tx.send(val).unwrap(); println!(val is {val}); }); let received rx.recv().unwrap(); println!(Got: {received}); }在刚才的代码上加了println!(val is {val});这句话。把值传入send函数后想继续在线程里使用值。输出$ cargo run Compiling message-passing v0.1.0 (/tmp/projects/message-passing) error[E0382]: borrow of moved value: val -- src/main.rs:10:27 | 8 | let val String::from(hi); | --- move occurs because val has type String, which does not implement the Copy trait 9 | tx.send(val).unwrap(); | --- value moved here 10 | println!(val is {val}); | ^^^ value borrowed here after move For more information about this error, try rustc --explain E0382. error: could not compile message-passing (bin message-passing) due to 1 previous error错误在于借用了已经移动的值val。它的所有权已经在传入send时移交出去了所以就会报错。下一个例子通过发送多个值来观察接收者等待的过程use std::sync::mpsc; use std::thread; use std::time::Duration; fn main() { let (tx, rx) mpsc::channel(); thread::spawn(move || { let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(thread), ]; for val in vals { tx.send(val).unwrap(); thread::sleep(Duration::from_secs(1)); } }); for received in rx { println!(Got: {received}); } }分线程以循环的方式发送Vector里的各个元素每次发送完之后会暂停 1 秒主线程把接收端当作一个迭代器来使用因为实现了Iteratortrait这样就不需要显式调用recv函数了。每收到一个值就将它打印出来。当发送端执行完毕被丢弃时Channel就关闭了循环就不会继续。程序退出。输出Got: hi Got: from Got: the Got: thread16.2.5. 通过克隆创建多个发送者继续在上一个代码的基础上稍作修改use std::sync::mpsc; use std::thread; use std::time::Duration; fn main() { let (tx, rx) mpsc::channel(); let tx1 tx.clone(); thread::spawn(move || { let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(thread), ]; for val in vals { tx1.send(val).unwrap(); thread::sleep(Duration::from_secs(1)); } }); thread::spawn(move || { let vals vec![ String::from(more), String::from(messages), String::from(for), String::from(you), ]; for val in vals { tx.send(val).unwrap(); thread::sleep(Duration::from_secs(1)); } }); for received in rx { println!(Got: {received}); } }这里多了一个分线程现在有 2 个分线程都想要给主线程发消息所以就需要两个发送端。针对这种情况只需要对代表发送端的变量tx使用clone方法即可也就是原文的let tx1 tx.clone();这一句。输出接收顺序不确定以下为一次代表性的本地运行结果Got: hi Got: more Got: from Got: messages Got: the Got: for Got: thread Got: you接收端收到的数据会交错来自两个发送端。