《Rust编程入门》12.2 消息传递与mpsc

January 2, 2025

12.2 消息传递与 `mpsc`

在 Rust 中，线程间通信的推荐方式是通过消息传递而非共享状态。这种方法可以显式地管理数据的所有权和生命周期，减少数据竞争。Rust 的标准库提供了一个 mpsc 模块，用于实现线程间的 多生产者单消费者（Multiple Producer, Single Consumer） 模型。

12.2.1 什么是 `mpsc`？

mpsc 是 Rust 标准库中用于实现消息传递的通道（channel）。它允许多个线程发送消息到同一个接收端，从而实现线程之间的数据传递。

多生产者：可以有多个发送端同时发送消息。
单消费者：只有一个接收端接收消息。

12.2.2 使用 `mpsc` 通道

创建通道

使用 mpsc::channel 函数可以创建一个通道，返回一个 (sender, receiver) 元组。

示例：基本消息传递

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


use std::sync::mpsc;
use std::thread;

fn main() {
    // 创建一个通道
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        let message = String::from("Hello from thread!");
        sender.send(message).unwrap(); // 发送消息
    });

    let received = receiver.recv().unwrap(); // 接收消息
    println!("Received: {}", received);
}

sender.send：发送消息。
receiver.recv：阻塞式接收消息。如果没有消息，会一直等待。
recv 返回 Result，需要处理可能的错误。

12.2.3 非阻塞接收

mpsc 提供了一个非阻塞方法 try_recv，如果当前没有消息可用，它会立即返回。

示例：非阻塞接收

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25


use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        thread::sleep(Duration::from_secs(1));
        sender.send("Hello after 1 second!").unwrap();
    });

    loop {
        match receiver.try_recv() {
            Ok(message) => {
                println!("Received: {}", message);
                break;
            }
            Err(_) => {
                println!("No message yet...");
            }
        }
        thread::sleep(Duration::from_millis(200));
    }
}

12.2.4 多生产者

通过克隆发送端，可以实现多个生产者向同一个接收端发送消息。

示例：多生产者单消费者

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel();

    let sender1 = sender.clone();

    thread::spawn(move || {
        sender.send("Message from sender 1").unwrap();
    });

    thread::spawn(move || {
        sender1.send("Message from sender 2").unwrap();
    });

    for received in receiver.iter() {
        println!("Received: {}", received);
    }
}

使用 sender.clone 创建额外的发送端。
receiver.iter 会在所有发送端关闭之前持续接收消息。

12.2.5 发送多种类型的数据

通过使用枚举或结构体，可以在通道中传递多种类型的数据。

示例：传递枚举

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


use std::sync::mpsc;
use std::thread;

enum Message {
    Text(String),
    Number(i32),
}

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        sender.send(Message::Text("Hello".to_string())).unwrap();
        sender.send(Message::Number(42)).unwrap();
    });

    for received in receiver {
        match received {
            Message::Text(text) => println!("Text message: {}", text),
            Message::Number(num) => println!("Number message: {}", num),
        }
    }
}

12.2.6 通道的生命周期

Rust 的通道遵循其所有权规则，当所有发送端被销毁后，接收端会收到一个错误（Err），表示通道已关闭。

示例：通道关闭

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        sender.send("Message 1").unwrap();
    });

    drop(sender); // 显式关闭发送端

    match receiver.recv() {
        Ok(msg) => println!("Received: {}", msg),
        Err(_) => println!("Channel closed."),
    }
}

12.2.7 性能与注意事项

性能优势

通道是线程安全的，适合多线程环境。
消息传递的方式更符合现代编程中的无共享状态设计。

注意事项

阻塞操作
如果接收端一直未准备好，send 操作可能会阻塞，影响程序性能。
数据量大时的性能开销
如果通道中传递的数据较大，可以使用 Arc 等工具只传递数据的引用，避免数据复制。

12.2.8 小结

mpsc 提供了一个简洁易用的通道模型，用于实现线程间的安全通信。
可以通过克隆发送端实现多生产者单消费者模式。
Rust 的所有权规则和通道设计确保了高效、安全的消息传递。

在下一节，我们将探索线程安全与共享状态的管理方式，包括如何使用 Mutex 和 RwLock 来实现线程间的安全数据共享。

12.2 消息传递与 mpsc

12.2.1 什么是 mpsc？

12.2.2 使用 mpsc 通道