Rust 信道（channel）
在并发那一页，线程之间用 Arc<Mutex<T>> 加锁来共享数据。这样可行，但还有一种往往更简单的做法：线程之间不共享数据，而是互相发送消息。
可以把信道（channel）想象成一根有两个口的管子。一个线程从发送端（sender）把值放进去，另一个线程从接收端（receiver）按同样的顺序把值取出来。谁也不跟谁共享东西：每个值都从一个线程移动到另一个线程。Rust 在这里借用了 Go 社区的一句话：“不要通过共享内存来通信，而要通过通信来共享内存。”
标准库的信道在 std::sync::mpsc 里。这个名字的意思是 multiple producer, single consumer（多个生产者，单个消费者）：可以有很多发送端，但只有一个接收端。
use std::sync::mpsc;use std::thread;
fn main() { let (tx, rx) = mpsc::channel();
thread::spawn(move || { let message = String::from("来自工作线程的问候"); tx.send(message).unwrap(); });
let received = rx.recv().unwrap(); println!("main 收到了：{received}");}输出：
main 收到了：来自工作线程的问候一步一步来看：
mpsc::channel()返回一对值：一个Sender和一个Receiver。按惯例，它们叫做tx（transmit，发送）和rx（receive，接收）。move把tx交给新线程，和你在并发那一页把数据移动到线程里的做法一样。tx.send(value)把一个值放进信道。它返回一个 Result，因为如果接收端已经不在了，发送就会失败。rx.recv()会等待，直到有值到达。只有当所有发送端都不在了、并且信道已经空了，它才会返回Err。
注意这里不需要 join()。recv() 本身就会等消息，所以 main 不会提前结束。
Receiver 可以当作迭代器来用。对它写一个 for 循环，循环会替你不断调用 recv()；等所有发送端都被丢弃后，循环会自己结束：
use std::sync::mpsc;use std::thread;
fn main() { let (tx, rx) = mpsc::channel();
thread::spawn(move || { for i in 1..=3 { tx.send(i * 10).unwrap(); } // 线程结束时，tx 在这里被丢弃 });
for value in rx { println!("收到 {value}"); } println!("信道已关闭，全部完成");}输出：
收到 10收到 20收到 30信道已关闭，全部完成这里只有一个发送端，所以值到达的顺序和发送的顺序完全一样。“发送端都不在了，循环就停止”这条规则很重要，下一节还会用到，先记住它。
多个生产者：克隆发送端
Section titled “多个生产者：克隆发送端”想让多个线程都能发送，就用 tx.clone() 给每个线程一份自己的发送端。所有副本都往同一个信道里送值：
use std::sync::mpsc;use std::thread;
fn main() { let (tx, rx) = mpsc::channel();
for id in 1..=3 { let tx = tx.clone(); thread::spawn(move || { tx.send(format!("工作线程 {id} 报到")).unwrap(); }); } drop(tx); // 丢弃原来的发送端，否则下面的循环永远不会结束
let mut messages: Vec<String> = rx.iter().collect(); messages.sort(); for m in messages { println!("{m}"); }}输出：
工作线程 1 报到工作线程 2 报到工作线程 3 报到有两点要注意：
drop(tx)很关键。 每个线程结束时会丢弃自己的克隆，但main手里还拿着原来那个。只要还有一个发送端存在，接收端就会一直等下去，程序就会永远卡住。丢弃它就等于告诉信道：“main 不会再发送任何东西了。”- 线程结束的顺序不固定。 消息可能按工作线程 2、1、3 的顺序到达。所以这里先把它们收集起来再排序，让打印结果每次都一样。去掉
sort()多运行几次，就能看到顺序在变。
不想一直等：try_recv 和 recv_timeout
Section titled “不想一直等：try_recv 和 recv_timeout”recv() 会一直阻塞到有东西到达。有时候你并不想等。这时有两个“亲戚”可以帮忙：
try_recv()会立刻返回：要么是一个值，要么是Empty（还没有东西），要么是Disconnected（所有发送端都不在了）。recv_timeout(duration)会等，但最多只等一段时间。时间到了还没有东西，就得到Timeout。
use std::sync::mpsc;use std::thread;use std::time::Duration;
fn main() { let (tx, rx) = mpsc::channel::<i32>();
// 还什么都没发送：try_recv 会立刻返回。 println!("{:?}", rx.try_recv());
thread::spawn(move || tx.send(7).unwrap()).join().unwrap();
// 已经有一个值在等着，所以这里不用等就能拿到它。 println!("{:?}", rx.recv_timeout(Duration::from_secs(1)));
// 发送端不在了，信道也空了。 println!("{:?}", rx.try_recv());}输出：
Err(Empty)Ok(7)Err(Disconnected)我们在接收之前先 join() 了那个线程，所以值已经在信道里等着了，输出每次都一样。
如果一个循环里还有别的事要做，try_recv() 就很好用，比如游戏循环每一帧检查一次有没有新消息。recv_timeout() 适合“工作线程太慢就放弃”这种场景。
用 sync_channel 创建有界信道
Section titled “用 sync_channel 创建有界信道”普通的 channel() 没有大小限制。如果发送端比接收端快很多，消息就会在内存里越堆越多。mpsc::sync_channel(n) 创建的信道最多只能存放 n 条消息。满了以后，send() 会等待，直到接收端取走一条。这叫做背压（backpressure）：慢的消费者会自动让快的生产者慢下来。
use std::sync::mpsc;use std::thread;
fn main() { // 同一时间只能放 2 条消息。 let (tx, rx) = mpsc::sync_channel(2);
let producer = thread::spawn(move || { for i in 1..=5 { tx.send(i).unwrap(); // 缓冲区满了就会阻塞 } });
let total: i32 = rx.iter().sum(); producer.join().unwrap(); println!("总和 = {total}");}输出：
总和 = 15生产者最多只能比 main 领先两个数字。如果写成 sync_channel(0)，就完全没有缓冲区：每次 send() 都要等接收端准备好接手，就像当面把包裹交到对方手里一样。
一个小小的工作线程池
Section titled “一个小小的工作线程池”现在把学到的东西组合起来。工作线程池（worker pool）就是几个线程从同一个队列里领任务，再把结果送回来。它用到两个信道：一个放任务，一个放结果。
这里有个问题：Receiver 不能克隆（毕竟是单个消费者）。所以工作线程们通过 Arc<Mutex<_>> 共享同一个任务接收端，轮流从里面取任务：
use std::sync::mpsc;use std::sync::{Arc, Mutex};use std::thread;
fn main() { let (job_tx, job_rx) = mpsc::channel::<u64>(); let (result_tx, result_rx) = mpsc::channel::<(u64, u64)>();
// Receiver 不能克隆，所以工作线程们通过 Arc<Mutex<_>> 共享同一个。 let job_rx = Arc::new(Mutex::new(job_rx));
let mut workers = Vec::new(); for _ in 0..3 { let job_rx = Arc::clone(&job_rx); let result_tx = result_tx.clone(); workers.push(thread::spawn(move || loop { let job = job_rx.lock().unwrap().recv(); match job { Ok(n) => result_tx.send((n, n * n)).unwrap(), Err(_) => break, // 没有任务了 } })); } drop(result_tx);
for n in 1..=6 { job_tx.send(n).unwrap(); } drop(job_tx); // 告诉工作线程：不会再有新任务了
let mut results: Vec<(u64, u64)> = result_rx.iter().collect(); for w in workers { w.join().unwrap(); } results.sort(); for (n, square) in results { println!("{n} 的平方是 {square}"); }}输出：
1 的平方是 12 的平方是 43 的平方是 94 的平方是 165 的平方是 256 的平方是 36它是怎样干净利落地收尾的：
drop(job_tx)关闭了任务信道。之后每个工作线程的recv()都会返回Err，工作线程就跳出循环。- 工作线程结束时，它手里的那个
result_tx克隆也被丢弃了。原来的那个我们早就丢弃了，所以三个工作线程都结束后，结果信道就关闭了。 - 然后
result_rx.iter()停止。因为工作线程结束的顺序不固定，所以最后要对结果排序。
还有一个小细节：job_rx.lock().unwrap().recv() 是故意写在同一条 let 语句里的。锁会在这条语句结束时释放，这样当这个工作线程还在计算平方的时候，其他工作线程已经可以去领下一个任务了。
信道和 Arc<Mutex<T>> 怎么选
Section titled “信道和 Arc<Mutex<T>> 怎么选”两者都是安全的，只是适合的问题形态不同：
| 适合用信道的情况…… | 适合用 Arc<Mutex<T>> 的情况…… |
|---|---|
| 工作朝一个方向流动：生产者产出东西，消费者负责处理 | 很多线程需要读取和更新同一个值（计数器、缓存） |
| 你需要一个由任务、事件或日志行组成的队列 | 数据很大，复制或到处移动会很浪费 |
| 每个值同一时间只有一个所有者 | 线程需要查看当前状态，而不是一连串的更新 |
一个好用的经验法则：如果你的程序可以描述为“这些线程告诉那个线程某件事”，就用信道；如果是“这些线程都在修改这个东西”，就用 Mutex。很多实际程序两者都会用，就像我们的工作线程池一样。
发送之后还在用这个值。 send() 会获取所有权，和把值传给函数一样。发送之后，这个值就归接收端了：
use std::sync::mpsc;use std::thread;
fn main() { let (tx, rx) = mpsc::channel();
thread::spawn(move || { let message = String::from("hi"); tx.send(message).unwrap(); println!("sent {message}"); });
println!("{}", rx.recv().unwrap());}error[E0382]: borrow of moved value: `message` --> src/main.rs:10:25 | 8 | let message = String::from("hi"); | ------- move occurs because `message` has type `String`, which does not implement the `Copy` trait 9 | tx.send(message).unwrap(); | ------- value moved here10 | println!("sent {message}"); | ^^^^^^^ value borrowed here after move这是 Rust 在保护你：如果发送方还能修改 message，两个线程就会同时碰到同一份数据。可以先打印再发送，或者发送一个 .clone()。
在循环里把发送端移动进多个线程。 第一个线程就把 tx 拿走了，第二个线程什么也拿不到：
use std::sync::mpsc;use std::thread;
fn main() { let (tx, rx) = mpsc::channel();
for id in 1..=3 { thread::spawn(move || { tx.send(id).unwrap(); }); }
for value in rx { println!("{value}"); }}error[E0382]: use of moved value: `tx` --> src/main.rs:8:23 | 5 | let (tx, rx) = mpsc::channel(); | -- move occurs because `tx` has type `std::sync::mpsc::Sender<i32>`, which does not implement the `Copy` trait... 8 | thread::spawn(move || { | ^^^^^^^ value moved into closure here, in previous iteration of loop修复方法是在循环开头写 let tx = tx.clone();，就像“多个生产者”那个例子一样。
程序永远不结束。 这种情况没有编译错误，所以特别隐蔽。如果你在遍历接收端（for x in rx），而某个发送端还活着（常常是 main 里原来那个 tx），循环就会永远等下去。程序“就那么卡住了”的时候，检查一下是不是每个发送端都被丢弃了。
std::sync::mpsc 能应付很多场景，不过你还会遇到两个常用的替代品：
- crossbeam-channel（英文）提供多消费者信道（接收端可以克隆），还有一个
select!宏，可以同时等待多个信道。 tokio::sync::mpsc是给异步代码用的信道。如果你在用async/.await（见 async/await），就用它代替标准库的信道，因为它的recv().await不会阻塞整个线程。
mpsc::channel()给你一对(Sender, Receiver)。send()把值移动进去，recv()等待下一个值。- 克隆发送端就能有多个生产者。接收端永远只有一个。
- 遍历接收端的循环会在所有发送端都被丢弃时结束，所以别忘了
drop()原来那个。 try_recv()和recv_timeout()让你可以检查信道，而不用一直等下去。sync_channel(n)把队列限制在n条消息。- 来自多个线程的消息到达顺序不固定：需要稳定的输出时，就排序或求和。
- 数据从一个线程流向另一个线程时用信道；多个线程共享并修改同一个值时用
Arc<Mutex<T>>。
小测验
1. “mpsc”是什么的缩写？
2. 怎样让三个线程往同一个信道里发送？
3. 一个 for value in rx 循环永远不结束。最可能的原因是什么？
4. sync_channel(2) 里已经有 2 条消息时，send() 会怎样？
得分: 0 / 4
- 使用消息传递在线程间通信 — Rust 程序设计语言：官方对信道的介绍，还有更多示例。
std::sync::mpsc— 标准库文档（英文）：Sender、SyncSender和Receiver的全部方法。- 通道 — 通过例子学 Rust：一个简短的可运行示例，有好几个生产者线程。
- 将单线程 server 变为多线程 server — Rust 程序设计语言：一步一步用信道和
Arc<Mutex<Receiver>>搭建一个完整的线程池。 tokio::sync::mpsc— docs.rs（英文）：配合 Tokio 使用的异步信道，以及它和标准库信道的区别。