当前位置：首页 > news >正文

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

news 来源：原创 2025/5/16 18:35:47

喜欢的话别忘了点赞、收藏加关注哦（加关注即可阅读全文），对接下来的教程有兴趣的可以关注专栏。谢谢喵！(=^･ω･=)
请添加图片描述

16.2.1. 消息传递

有一种很流行而且能保证安全并发的技术（或者叫机制）叫做消息传递。在这种机制里，线程(或Actor)通过彼此间发送消息（数据）来进行通讯。

Go语言有一句名言是这么说的：Do not communicate by sharing memory; instead, share memory by communicating.（不要用共享内存来通信，要用通信来共享内存）

Go语言的并发机制体现了这种思想。Rust也提供了机遇消息传递的一种并发方式，具体的就是在Rust里实现就是使用Channel（标准库提供）。Go语言里也有Channel，思路差不多。

16.2.2. 理解`Channel`

可以将编程中的Channle想象为定向水道，例如小溪或河流。如果你把橡皮鸭之类的东西放入河中，它会顺流而下，到达水道的尽头。

通道有两部分：发送端和接收端。发射端是将橡皮鸭放入河中的上游位置，接收端是橡皮鸭最终到达下游的位置。代码的一部分使用要发送的数据调用发送端上的方法，另一部分检查接收端是否有到达的消息。如果发送端或接收端其一掉线，则称通道已关闭。

具体的步骤：

调用发送端的方法，发送数据
接收端会检查和接收到达的数据
如果发送端、接收端中的任意一端被丢弃了，那么Channel就关闭了。

16.2.3. 创建`channel`

使用mpsc::channel函数来创建Channel。mpsc表示multiple producer, single consumer（多个生产者、一个消费者），表示可以有多个发送端，但是只能有一个接收端。

调用这个函数返回一个元组(tuple，详见 3.3. 数据类型：复合类型)，有两个元素，分别是发送端和接收端。

看个例子：

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方法（recieve的简写）。它会一直阻塞这个线程，直到有消息被传入为止。
消息被包裹在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 runCompiling message-passing v0.1.0 (file:///projects/message-passing)
error[E0382]: borrow of moved value: `val`--> src/main.rs:10:26|
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|= note: this error originates in the macro `$crate::format_args_nl` which comes from the expansion of the macro `println` (in Nightly builds, run with -Z macro-backtrace for more info)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秒
主线程接收端被当了一个迭代器来使用（因为实现了iterator trait），这样就不需要显式调用recv函数了。每收到一个值就将它打印出来。当发送端执行完毕被丢弃时，Channel就关闭了，循环就不会继续。程序退出。

输出：

Got: hi
Got: from
Got: the
Got: thread

16.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: for
Got: the
Got: thread
Got: you

接收端收到的数据是交替出现的。

16.2.1. 消息传递

16.2.2. 理解Channel

16.2.3. 创建channel

发送端send方法