3、消息传递：通道(Channel)基础、多生产者多消费者、异步通道、select宏

聊到并发编程，消息传递是个绕不开的话题。Rust 的通道（Channel）设计得相当精巧，它不像 Go 那样把通道作为语言内置的原语，而是通过标准库和第三方 crate 提供了灵活的选择。我个人习惯把通道理解为「线程间的水管」——一头往里灌数据，另一头接数据，水流的方向是单向的。

3.1 通道基础：std::sync::mpsc

标准库提供的 mpsc 全称是 Multi-Producer, Single-Consumer（多生产者，单消费者）。说白了，就是允许多个线程往同一个通道里发消息，但只能有一个线程在接收端读取。

先看一个最简单的例子：

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

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        tx.send(String::from("你好，主线程！")).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("收到消息：{}", received);
}

这里 tx 是发送端，rx 是接收端。注意 tx 被 move 进了子线程，因为通道的所有权必须转移过去。我在项目中遇到过新手直接把 tx 留在主线程里用，结果编译报错——所有权规则在这里卡得很严。

3.2 多生产者：克隆发送端

标准库的 mpsc 虽然名字叫多生产者，但它的发送端 Sender 本身并不支持多线程直接共享。怎么办？答案是克隆。

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx1 = tx.clone();

    thread::spawn(move || {
        tx.send("来自线程1").unwrap();
    });

    thread::spawn(move || {
        tx1.send("来自线程2").unwrap();
    });

    for received in rx {
        println!("收到：{}", received);
    }
}

你想想看，每个线程拿到的是独立的 Sender 克隆体，它们内部通过原子操作共享同一个通道缓冲区。我曾经在日志收集系统里用过这个模式——多个工作线程把处理结果发到同一个通道，主线程统一写入文件。嗯，这里要注意：tx 被 move 进第一个线程后，主线程就不能再用了，必须提前 clone。

3.3 多消费者：crossbeam 的通道

标准库的 mpsc 只能单消费者，这在实际项目中往往不够用。比如你有一个任务分发器，多个工作线程需要从同一个通道里取任务——这时候就需要 MPMC（多生产者多消费者）通道。

Rust 社区里最常用的方案是 crossbeam crate。它提供了真正的多生产者多消费者通道：

use crossbeam::channel;
use std::thread;

fn main() {
    let (s, r) = channel::unbounded();

    // 多个生产者
    for i in 0..3 {
        let s = s.clone();
        thread::spawn(move || {
            s.send(format!("生产者 {}", i)).unwrap();
        });
    }

    // 多个消费者
    for _ in 0..2 {
        let r = r.clone();
        thread::spawn(move || {
            for msg in r {
                println!("消费者收到：{}", msg);
            }
        });
    }

    thread::sleep(std::time::Duration::from_secs(1));
}

我个人觉得 crossbeam 的通道比标准库更灵活。它支持有界（bounded）和无界（unbounded）两种模式，而且性能调优做得很好。我在做实时数据流处理时，就用有界通道来防止内存暴涨——缓冲区满了之后发送端会自动阻塞，天然实现了背压。

3.4 异步通道：tokio 的 mpsc

如果你在用异步运行时（比如 tokio），就不能用标准库的阻塞通道了。异步通道的核心是 .await——发送和接收都不会阻塞线程，而是挂起当前任务。

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(32);

    tokio::spawn(async move {
        tx.send("异步消息").await.unwrap();
    });

    if let Some(msg) = rx.recv().await {
        println!("收到：{}", msg);
    }
}

注意 mpsc::channel(32) 里的 32 是缓冲区容量。异步通道必须指定大小，因为 tokio 需要知道什么时候该唤醒等待的任务。我曾经在 WebSocket 服务里用异步通道做消息转发，每个连接一个通道，配合 select 宏处理多个事件源——效果相当丝滑。

3.5 select 宏：同时监听多个通道

实际项目中，你往往需要同时监听多个通道。比如一个网络服务，既要处理客户端消息，又要处理心跳超时，还要处理系统关闭信号。这时候 select! 宏就派上用场了。

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx1, mut rx1) = mpsc::channel::<String>(16);
    let (tx2, mut rx2) = mpsc::channel::<String>(16);

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        tx1.send("来自通道1").await.unwrap();
    });

    tokio::spawn(async move {
        sleep(Duration::from_millis(200)).await;
        tx2.send("来自通道2").await.unwrap();
    });

    loop {
        tokio::select! {
            Some(msg) = rx1.recv() => {
                println!("通道1：{}", msg);
            }
            Some(msg) = rx2.recv() => {
                println!("通道2：{}", msg);
            }
            _ = sleep(Duration::from_millis(500)) => {
                println!("超时，退出循环");
                break;
            }
        }
    }
}

select! 会同时等待多个分支，哪个先准备好就执行哪个。如果多个分支同时就绪，它会随机选择一个（这是 tokio 的设计选择，避免饥饿）。

嗯，这里有个细节：select! 里的每个分支必须是 async 表达式，而且 recv() 返回的是 Option，所以要用 Some(msg) 来模式匹配。我刚开始用的时候老写成 msg = rx.recv()，结果编译报错——模式匹配不完整。

3.6 通道的选择：场景决定方案

说了这么多，到底该用哪种通道？我整理了一个简单的对照表：

场景	推荐方案	理由
简单线程间通信	std::sync::mpsc	零依赖，够用
多消费者工作池	crossbeam::channel	MPMC 支持，有界背压
异步任务通信	tokio::sync::mpsc	非阻塞，与运行时集成
广播/发布订阅	tokio::sync::broadcast	一对多，每个消费者收到相同消息

我个人建议：如果项目里已经用了 tokio，那就统一用 tokio 的通道。如果只是简单的多线程处理，标准库的 mpsc 完全够用。别为了炫技引入不必要的依赖。

3.7 本章知识体系

下面这张图展示了通道相关的核心概念和它们之间的关系：

从这张图可以看出，通道的核心是生产者和消费者之间的解耦。你可以根据场景选择不同的组合：标准库的 SPSC、crossbeam 的 MPMC、tokio 的异步通道，再配合 select! 实现多路复用。

---