Rust并发编程之使用消息传递进行线程间数据共享方式

半米日暮

目录

• 一、通道（Channel）的基本概念
• 二、创建并使用通道
• 1. 基础用法
• 2. 在子线程中发送消息
• 3. 通道与所有权
• 4. 发送多个消息
• 5. 多个发送端（Multiple Producer）
• 三、总结

一、通道（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（接收端）。运行结果每次可能都不一样，因为线程的调度顺序不可预测，这也正是并发编程“有趣”且需要谨慎之处。

三、总结

1. 创建通道：使用 mpsc::channel() 创建通道，获得 (tx, rx)（发送端和接收端）。
2. 发送数据：tx.send(data) 会转移 data 的所有权，若接收端已关闭，send 会返回错误。
3. 接收数据：rx.recv() 会阻塞等待数据；rx.try_recv() 则不会阻塞，可用于非阻塞检查。也可将 rx 当做迭代器使用，以便持续接收数据，直到通道被关闭。
4. 所有权规则保障安全：发送端在 send 时会移动数据的所有权，避免了多线程中对同一数据的潜在不安全访问。
5. 多发送端：通过克隆发送端（tx.clone()），多个线程可以各自发送数据到同一个接收端，从而实现复杂的多生产者单消费者架构。

Rust 的通道借助所有权系统，帮助我们轻松规避了许多并发陷阱。通过消息传递的思路，不再需要小心翼翼地管理锁和共享数据，编程思路也往往更加清晰简洁。在实际项目中，若需要多个线程之间相互通信，不妨考虑一下通道（channel）方案，也许能带来更加优雅和可靠的并发架构。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持琼殿技术社区。

您可能感兴趣的文章:

• 浅析Rust多线程中如何安全的使用变量
• Rust 多线程编程的实现
• Rust使用Channel实现跨线程传递数据
• rust 创建多线程web server的详细过程
• Rust多线程Web服务器搭建过程
• Rust如何使用线程同时运行代码