Rust并发编程之使用消息传递进行线程间数据共享方式
作者:Hello.Reader
一、通道(Channel)的基本概念
一个通道可以想象成一条单向水道或河流:有一个 发送端(transmitter) 和一个 接收端(receiver)。发送端好比河流上游,负责把“橡皮鸭”丢进水里;接收端在河流下游,收到这只“橡皮鸭”。在编程中,线程之间的通信即是这样——把数据发到通道的一端,另外一个(或多个)线程在通道的另一端接收。
Rust 通过 std::sync::mpsc
(Multiple Producer, Single Consumer)来提供通道功能:
- Multiple Producer:可以有多个发送端同时发送数据;
- Single Consumer:但只能有一个接收端来接收数据。
通过克隆发送端可以允许多个线程一起发送数据给同一个接收端。
二、创建并使用通道
1. 基础用法
创建一个 mpsc::channel
use std::sync::mpsc; use std::thread; fn main() { // 创建一个通道 let (tx, rx) = mpsc::channel(); // 这里的 tx 是 transmitter(发送端),rx 是 receiver(接收端)。 // 我们先不发送任何数据,因此代码暂时无法编译, // 因为编译器不知道通道要发送什么类型的数据。 }
mpsc::channel()
函数会返回一个元组 (tx, rx)
,分别代表发送端和接收端。在之后,我们会看到如何把 tx
移动到不同线程去发送消息,rx
则留在当前线程用于接收消息。
2. 在子线程中发送消息
下面的例子中,我们在主线程创建了通道,然后把发送端 tx
移动(move
)到子线程中,子线程通过 tx.send()
发送一条字符串“hi”给主线程。
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(); // 如果接收端已关闭,则 send 会返回错误,这里直接 unwrap 处理 }); // 在主线程接收消息 let received = rx.recv().unwrap(); // recv 会阻塞主线程,直到收到一条消息或者发送端被关闭 println!("Got: {}", received); }
运行后,主线程会打印:
Got: hi
这表示主线程成功地收到了子线程通过通道发送的字符串。
3. 通道与所有权
当我们调用 tx.send(val)
时,send
方法会获取 val
的所有权。这样做能够避免在另一个线程修改数据后,我们在原线程又使用这段数据的潜在风险。
例如,下面这段示例代码(示意)试图在发送之后继续使用 val
,就会导致编译错误:
use std::sync::mpsc; use std::thread; fn main() { let (tx, rx) = mpsc::channel(); let val = String::from("hello"); thread::spawn(move || { tx.send(val).unwrap(); // 发送后 val 的所有权已转移到通道 // println!("val is: {}", val); // 编译错误: val 所有权已经被移动 }); let received = rx.recv().unwrap(); println!("Got: {}", received); }
由于所有权已经转移,主线程可以安全地接收并处理这条消息,而子线程也不会再访问已经移出的数据。这种严格的所有权规则能有效避免数据竞争和其他并发错误。
4. 发送多个消息
我们可以让子线程发送不止一条消息。下面的例子让子线程依次发送多条字符串,并在发送之间加上 sleep
用来模拟耗时操作:
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)); } }); // 主线程中,通过 iter 的方式持续接收消息 for received in rx { println!("Got: {}", received); } }
由于接收端可以被当做迭代器来使用,当所有发送端都关闭时,for
循环会自动结束。这段程序会像下面这样依次打印每条消息:
Got: hi
Got: from
Got: the
Got: thread
5. 多个发送端(Multiple Producer)
mpsc
的含义之一就是 Multiple Producer。如果我们希望有多个不同的子线程来发送消息给同一个接收端,只需要克隆发送端即可。如下:
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("first 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(); // 这里使用原先的 tx thread::sleep(Duration::from_secs(1)); } }); for received in rx { println!("Got: {}", received); } }
- 第一个子线程使用
tx1
; - 第二个子线程使用原本的
tx
;
所有发送过来的数据都将流向同一个 rx
(接收端)。运行结果每次可能都不一样,因为线程的调度顺序不可预测,这也正是并发编程“有趣”且需要谨慎之处。
三、总结
- 创建通道:使用
mpsc::channel()
创建通道,获得(tx, rx)
(发送端和接收端)。 - 发送数据:
tx.send(data)
会转移data
的所有权,若接收端已关闭,send
会返回错误。 - 接收数据:
rx.recv()
会阻塞等待数据;rx.try_recv()
则不会阻塞,可用于非阻塞检查。也可将rx
当做迭代器使用,以便持续接收数据,直到通道被关闭。 - 所有权规则保障安全:发送端在
send
时会移动数据的所有权,避免了多线程中对同一数据的潜在不安全访问。 - 多发送端:通过克隆发送端(
tx.clone()
),多个线程可以各自发送数据到同一个接收端,从而实现复杂的多生产者单消费者架构。
Rust 的通道借助所有权系统,帮助我们轻松规避了许多并发陷阱。通过消息传递的思路,不再需要小心翼翼地管理锁和共享数据,编程思路也往往更加清晰简洁。在实际项目中,若需要多个线程之间相互通信,不妨考虑一下通道(channel)方案,也许能带来更加优雅和可靠的并发架构。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。