消息传递:channel - 技术学习笔记

一句话总结

除了共享内存(Arc/Mutex), 并发的另一种范式是消息传递: 线程之间不共享数据, 而是通过通道(channel)互相发送消息.
标准库的 mpsc(多生产者, 单消费者)是其实现.
在 Rust 里, 消息的所有权随发送一起转移给接收方--所有权系统让"通过通信共享内存"这句话有了编译期保障.

并发的两种范式

到现在已见过共享状态(Arc/Mutex). 并发世界还有另一条路线, 由 Go 发扬光大的那句口号点明:

不要通过共享内存来通信, 而要通过通信来共享内存.
(Do not communicate by sharing memory; instead, share memory by communicating.)

两种范式的区别:

	共享状态(上一篇)	消息传递(本篇)
线程如何协作	多个线程访问同一份加锁的数据	线程间发送/接收消息, 不共享数据
核心工具	`Arc<Mutex<T>>`	channel(`mpsc`)
心智负担	要管锁, 防死锁	无锁, 数据所有权清晰流转

共享状态像一块多人共用的白板: 谁要写就得先拿到唯一的笔(锁), 用完归还.
消息传递像邮件往来: 每人有自己的桌子(各自的数据), 要协作就把东西寄给对方--东西寄出后就不在手上了, 不存在"两人同时改一份"的问题.
后者从源头上回避了数据竞争.

mpsc: 创建一个通道

标准库的通道在 std::sync::mpsc. 名字 mpsc = Multiple Producer, Single Consumer(多生产者, 单消费者): 可以有多个发送端, 但只有一个接收端. channel() 返回一对发送端 tx 和接收端 rx:

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

let (tx, rx) = mpsc::channel(); // tx = 发送端, rx = 接收端

thread::spawn(move || {
    let msg = String::from("来自子线程");
    tx.send(msg).unwrap(); // 发送, msg 的所有权被移交出去
});

let received = rx.recv().unwrap(); // 阻塞接收
println!("收到: {received}");

send 把消息发出, recv 阻塞等待直到收到一条消息(类似 Go 的 <-ch).

所有权随消息转移: 这才是关键

这是 Rust channel 与 Go channel 最深刻的差异, 也是消息传递在 Rust 里"天然安全"的原因. 看 send 之后会发生什么:

let (tx, rx) = mpsc::channel();
let msg = String::from("hello");
tx.send(msg).unwrap();
// println!("{msg}"); // 编译错误! msg 的所有权已随 send 转移走了

send(msg) 把 msg 的所有权移交给了通道(进而移交给接收方). 发送之后, 发送方再也不能访问 msg--这是所有权的移动规则在起作用. 于是"一份数据被发送方和接收方同时持有"的情况根本不可能发生, 数据竞争从源头消失.

'通过通信共享内存'在 Rust 里是编译期事实

Go 也提倡这句口号, 但它不强制--完全可以把一个指针塞进 channel 发出去, 然后发送方继续用那个指针, 造成共享 + 竞争(Go 编译器不拦, 要靠 race detector).

Rust 不同: 消息所有权随 send 转移是编译期强制的, 发送后碰原数据就编译不过. 所以这句口号在 Rust 里不是"建议", 而是"类型系统保证的事实"--发送即放手.

把接收端当迭代器

接收端 rx 可以直接当迭代器遍历, 它会持续产出消息, 直到所有发送端关闭(被 drop):

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

let (tx, rx) = mpsc::channel();

thread::spawn(move || {
    for i in 1..=3 {
        tx.send(i).unwrap();
    }
    // tx 在闭包结束时被 drop, 通道关闭
});

for received in rx { // 迭代, 直到通道关闭后自动结束
    println!("收到: {received}");
}

通道何时关闭

当所有发送端都被 drop 时, 通道关闭, 接收端的迭代随之结束(recv 则返回 Err).

这对应 Go 里 close(ch) 后 for range ch 退出的行为, 只是 Rust 靠"发送端的所有权/Drop"自动达成, 无需显式 close.

多生产者: 克隆发送端

mpsc 的"多生产者"通过克隆发送端实现--每个线程拿一个 tx 的克隆, 都能往同一个通道发:

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

let (tx, rx) = mpsc::channel();

for i in 0..3 {
    let tx = tx.clone(); // 每个生产者一个克隆
    thread::spawn(move || {
        tx.send(format!("线程 {i} 的消息")).unwrap();
    });
}
drop(tx); // 丢弃最初的 tx, 否则通道永不关闭(它还活着)

for msg in rx {
    println!("{msg}");
}

记得 drop 掉多余的发送端

上面 drop(tx) 不能省. 通道在所有发送端都 drop 后才关闭--克隆了 3 个发给线程, 但最初那个 tx 还在主线程手里.

若不 drop 它, for msg in rx 会因"还有一个发送端活着"而永久阻塞等待. 这是 mpsc 的常见新手坑.

两种范式怎么选

实践指导

消息传递(channel): 适合"任务分发""流水线""事件通知"等数据有明确流向的场景. 所有权清晰流转, 无锁, 不易出并发 bug--优先考虑.
共享状态(Arc/Mutex): 适合"多个线程频繁读写同一份共享状态"(如共享计数器, 缓存)的场景, 用 channel 表达反而别扭时.

两者不是对立的, 实践中常混用. 但当一个问题用 channel 能自然表达时, 优先用它--把所有权的清晰流转交给编译器, 比手动管锁省心.

快速回顾

两种范式: 共享状态(Arc/Mutex, 共享加锁数据)vs 消息传递(channel, 发消息不共享).
mpsc: 多生产者单消费者通道, channel() 返回 (tx, rx), send/recv 收发.
所有权随消息转移: send 后发送方无法再访问该数据, 编译期杜绝"发出去还在用"的竞争.
关闭: 所有发送端 drop 后通道关闭, for x in rx 自动结束; 多生产者靠 tx.clone(), 记得 drop 多余 tx.
与 Go 对比: Go 的"通过通信共享内存"是建议, Rust 靠所有权把它变成编译期强制的事实.
选择: 数据有明确流向优先用 channel; 频繁读写共享状态用 Arc/Mutex.

动手练习

下面代码为什么编译不过? 它体现了 Rust channel 的什么特性?

let (tx, rx) = mpsc::channel();
let v = vec![1, 2, 3];
tx.send(v).unwrap();
println!("{:?}", v);

参考答案

编译错误: v 的所有权在 tx.send(v) 时已转移给通道(接收方), 发送后发送方不能再访问它, 所以 println!("{:?}", v) 用了已移动的值.

这体现 Rust channel 的核心特性--所有权随消息转移, 从根本上杜绝"数据被发送方和接收方同时持有"的竞争.

下面的程序会永久阻塞, 为什么? 如何修复?

let (tx, rx) = mpsc::channel();
let tx2 = tx.clone();
thread::spawn(move || tx2.send(1).unwrap());
for x in rx { println!("{x}"); }

参考答案

永久阻塞: for x in rx 在所有发送端都 drop 后才结束, 但最初的 tx 仍在主线程作用域内活着, 通道永不关闭, 迭代收完那条消息后继续等待.

修复: 在循环前 drop 掉主线程持有的 tx:

let (tx, rx) = mpsc::channel();
let tx2 = tx.clone();
thread::spawn(move || tx2.send(1).unwrap());
drop(tx); // 丢弃多余发送端
for x in rx { println!("{x}"); } // 收到 1 后通道关闭, 正常结束

用 channel 实现: 3 个工作线程各自计算一个值并发回主线程, 主线程收集所有结果.

参考答案

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

fn main() {
    let (tx, rx) = mpsc::channel();
    for i in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(i * 10).unwrap(); // 计算并发回
        });
    }
    drop(tx); // 丢弃最初的 tx

    let results: Vec<i32> = rx.iter().collect(); // 收集直到通道关闭
    println!("{:?}", results); // 顺序不定, 如 [0, 20, 10]
}

注意 drop(tx) 必不可少, 否则 rx.iter() 不会结束. 结果顺序取决于线程调度, 不固定.

Go 也提倡"通过通信共享内存", 但为什么说这句话在 Rust 里是"编译期保证", 在 Go 里只是"建议"?

参考答案

在 Go 里, 可以把一个指针塞进 channel 发出去, 然后发送方继续通过那个指针访问数据--这就变成了"既通信又共享"且存在竞争, Go 编译器不会阻止, 只能靠运行时 race detector 抽查.

而在 Rust 里, send 会转移所有权: 发送后发送方在编译期就无法再访问该数据, "发出去还自己留着用"根本通不过编译. 所以"发送即放手, 不再共享"在 Rust 里是类型系统强制的事实, 而非靠程序员自觉遵守的建议.