深入 Rust async：select!、超时、stream 和 Pin
在 Async/Await 那一页，你已经学过基础知识：async fn 返回一个惰性的 future，.await 运行它，#[tokio::main] 启动运行时，tokio::join! 和 tokio::spawn 让多件事同时进行。如果这些对你来说还很陌生，先去读那一页。这一页是在它的基础上继续往前走。
真实的异步程序还需要几样工具。再想想那个忙碌的厨房。有时候你要等每一道菜都做好（join!）。有时候你只关心最先发生的那件事：菜做好了，或者客人走了（select!）。有时候等了五分钟你就不等了（timeout）。还有时候，订单一整晚源源不断地一张张送进来（stream）。这一页会讲到所有这些，还会用大白话解释 Pin，这个你迟早会在错误信息里看到的奇怪名词。
所有示例都使用 tokio，并开启 features = ["full"]，和基础篇一样。“试一试”按钮都能用，因为 Rust Playground 内置了 tokio。
同时运行很多件事
Section titled “同时运行很多件事”你已经认识 tokio::join! 了：它等待所有 future 完成，并把结果作为元组返回。它的兄弟 tokio::try_join! 用于返回 Result 的 future。它要么给你包含所有值的 Ok，要么给你第一个 Err，而且一旦有一个失败，就不再等待其他的：
use std::time::Duration;use tokio::time::sleep;
async fn load_profile(id: u32) -> Result<String, String> { sleep(Duration::from_millis(20)).await; Ok(format!("资料 #{id}"))}
async fn load_settings(id: u32) -> Result<String, String> { sleep(Duration::from_millis(10)).await; if id == 2 { return Err(format!("用户 {id} 没有设置")); } Ok(String::from("深色模式"))}
#[tokio::main]async fn main() { for id in [1, 2] { match tokio::try_join!(load_profile(id), load_settings(id)) { Ok((profile, settings)) => println!("用户 {id}：{profile}，{settings}"), Err(e) => println!("用户 {id} 加载失败：{e}"), } }}输出：
用户 1：资料 #1，深色模式用户 2 加载失败：用户 2 没有设置join! 和 try_join! 是在当前任务内部轮流运行这些 future 的。如果是几个事先就知道的 future，这再合适不过。但如果你有一串任务，数量要到运行时才知道，就应该把它们作为独立的任务 spawn 出去。你可以像基础篇那样，把每个 JoinHandle 放进一个 Vec。不过 tokio 有个更好用的工具：JoinSet，一个装任务的袋子，每个任务一完成就把结果交给你：
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::sleep;
async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) { sleep(Duration::from_millis(ms)).await; // 假装这是一次网络请求 (site, ms)}
#[tokio::main]async fn main() { let mut set = JoinSet::new(); for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] { set.spawn(check_site(site, ms)); }
let mut results = Vec::new(); while let Some(res) = set.join_next().await { results.push(res.unwrap()); // 只有任务 panic 时才会是 Err }
results.sort(); // 完成顺序每次运行都可能不同 for (site, ms) in results { println!("{site} 用了 {ms}ms 响应"); }}输出：
crates.io 用了 20ms 响应docs.rs 用了 10ms 响应rust-lang.org 用了 30ms 响应集合空了以后，join_next() 会返回 None，所以 while let 循环会自己结束。结果按完成顺序到达，而不是 spawn 的顺序。如果几个任务用时差不多，这个顺序每次运行都可能不同，所以打印之前我们先排了序。还有一个好处：如果你丢弃了一个 JoinSet，里面还没完成的任务都会被取消，不会有任务被意外地留在后台运行。
用 select! 让 future 赛跑
Section titled “用 select! 让 future 赛跑”join! 等待所有 future，tokio::select! 只等第一个。每个分支由一个模式、一个 future 和一段代码组成：pattern = future => code。哪个 future 先完成，就运行它的那段代码，其余分支都会被取消。
在 Rust 里，“取消”是什么意思？输掉的 future 只是被丢弃了，就像其他任何离开作用域的值一样。它永远不会再运行。它的局部变量也会被丢弃，所以它们的清理代码会执行。下面这个例子用一个小小的 Drop 类型把这个过程展示出来：
use std::time::Duration;use tokio::time::sleep;
struct Download(&'static str);
impl Drop for Download { fn drop(&mut self) { println!("清理{}的下载", self.0); }}
async fn download(mirror: &'static str, ms: u64) -> &'static str { let _download = Download(mirror); sleep(Duration::from_millis(ms)).await; println!("{mirror}的下载完成了"); mirror}
#[tokio::main]async fn main() { tokio::select! { winner = download("欧洲", 50) => println!("胜出：{winner}"), winner = download("亚洲", 1000) => println!("胜出：{winner}"), } println!("main 继续执行");}输出：
欧洲的下载完成了清理欧洲的下载清理亚洲的下载胜出：欧洲main 继续执行注意少了什么：“亚洲的下载完成了”从来没有打印出来。亚洲那个 future 停在了它的 .await 处并被丢弃，它的 Download 值也立刻被清理掉了。异步 Rust 里的取消都是这样运作的：不再 poll 一个 future，然后丢弃它。不需要什么特别的“取消”调用。
select! 的一个经典用法是超时：让你的工作和一个计时器赛跑。
tokio::select! { report = build_report() => println!("{report}"), _ = tokio::time::sleep(Duration::from_secs(5)) => println!("太慢了，放弃"),}这样写完全没问题，不过还有一个专门的捷径 tokio::time::timeout，下一节会讲到。
一直工作，直到收到停止信号
Section titled “一直工作，直到收到停止信号”select! 在 loop 里才真正大显身手。下面这个 worker 从一个信道里接收任务，同时还在第二个信道上监听“请停下来”的消息。oneshot 信道只传递一条消息，正好用来当停止信号：
use std::time::Duration;use tokio::sync::{mpsc, oneshot};use tokio::time::sleep;
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { Some(job) = job_rx.recv() => { println!("正在处理：{job}"); done += 1; } _ = &mut stop_rx => { println!("收到停止信号"); break; } } } done });
for job in ["调整照片尺寸", "发送邮件", "更新搜索索引"] { job_tx.send(job.to_string()).await.unwrap(); } sleep(Duration::from_millis(100)).await; // 给 worker 留出完成的时间 stop_tx.send(()).unwrap();
let done = worker.await.unwrap(); println!("worker 一共完成了 {done} 个任务");}输出：
正在处理：调整照片尺寸正在处理：发送邮件正在处理：更新搜索索引收到停止信号worker 一共完成了 3 个任务有几个细节值得了解：
- 我们写的是
&mut stop_rx，而不是stop_rx。每一轮循环，select!都需要同一个接收端，所以我们把它借出去，而不是在第一轮就把它移动走。 - 每一轮循环，
job_rx.recv()都会创建一个新的 future，输掉的那个会被丢弃。这没关系，因为recv()是取消安全（cancel-safe）的：丢弃它永远不会丢失消息。 - 如果模式没有匹配上（这里是指所有发送端都没了，
recv()返回None），select!会关掉那个分支，继续等待其他分支。 - 如果多个分支同时就绪，
select!会随机挑一个，这样就没有哪个分支能霸占整个循环。如果想从上到下依次检查分支，在select!里第一行写上biased;。
一段话讲清取消安全。因为输掉的分支会被丢弃，它做了一半的工作也会被扔掉。对信道的 recv() 或者 sleep() 来说，这没什么害处：什么都还没被取走。但想象一个 future 要从网络 socket 读 100 个字节，已经读了 60 个时被丢弃了，那 60 个字节就没了。这样的 future 不是取消安全的。Tokio 的文档会写明每个方法是否取消安全。在循环里用 select! 时，检查一下你调用的方法；或者在循环外只创建一次 future，然后一直 poll 同一个 future（在 Pin 那一节你会看到怎么做）。
超时和计时器
Section titled “超时和计时器”tokio::time::timeout(duration, future) 可以包住任何 future。如果 future 按时完成，你会得到 Ok(value)。如果时间先用完了，future 就会被丢弃（也就是被取消，和 select! 里输掉的分支一样），你会得到 Err(Elapsed)：
use std::time::Duration;use tokio::time::{sleep, timeout};
async fn fetch_price(server_delay_ms: u64) -> u32 { sleep(Duration::from_millis(server_delay_ms)).await; 42}
#[tokio::main]async fn main() { let limit = Duration::from_millis(200);
for (server, delay) in [("快服务器", 20), ("慢服务器", 2000)] { match timeout(limit, fetch_price(delay)).await { Ok(price) => println!("{server}：价格是 {price}"), Err(e) => println!("{server}：{e}"), } }}输出：
快服务器：价格是 42慢服务器：deadline has elapseddeadline has elapsed 是 tokio 自己的错误信息，意思是“时限已到”。凡是和外部世界打交道的操作都应该加上超时：HTTP 请求、数据库查询、从 socket 读数据。一个永远不回应的服务器，不应该让你的程序永远卡住。
sleep 和 interval 的区别。sleep 只等一次。interval 给你一个稳定的节奏，就像节拍器。你在循环里调用 .tick().await，每次 tick 都会按计划到来。第一次 tick 会立即触发：
use std::time::Duration;use tokio::time::interval;
#[tokio::main]async fn main() { let mut heartbeat = interval(Duration::from_millis(50));
for beat in 1..=3 { heartbeat.tick().await; // 第一次 tick 会立即完成 println!("心跳 {beat}"); }}输出：
心跳 1心跳 2心跳 3为什么不在循环里 sleep 50ms 就好？因为那样每一轮要花“你的工作时间 + 50ms”，时间表会慢慢漂移。interval 是从每次 tick 的计划开始时间算起的，所以如果你的工作花了 10ms，它只会再等剩下的 40ms。心跳、轮询、“每分钟保存一次”这类任务用 interval；只想停顿一次就用 sleep。
Stream：异步迭代器
Section titled “Stream：异步迭代器”迭代器用 next() 一个接一个地给你值。stream（流）是它的异步版本：next().await 会在下一个值准备好时把它交给你，没有更多值时返回 None。想想不断跳动的股票价格、从数据库陆续到达的行，或者来自聊天服务器的消息。
其实你已经用过一个类似 stream 的东西了：tokio 信道的接收端。rx.recv().await 会等待下一条消息，等所有发送端都没了就返回 None：
use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8);
tokio::spawn(async move { for temp in [21, 22, 24, 23] { tx.send(temp).await.unwrap(); sleep(Duration::from_millis(10)).await; // 每 10ms 读一次数 } // tx 在这里被丢弃，信道随之关闭 });
while let Some(temp) = rx.recv().await { println!("温度：{temp}°C"); } println!("传感器停止了");}输出：
温度：21°C温度：22°C温度：24°C温度：23°C传感器停止了标准库还没有 Stream trait（它还在设计中）。整个生态用的是 futures crate 里的那个，而 tokio 的辅助 crate tokio-stream 提供了一个 StreamExt trait，里面有你在迭代器上熟悉的那些方法：next()、map()、filter()、take() 等等。把它加到你的项目里：
[dependencies]tokio = { version = "1", features = ["full"] }tokio-stream = "0.1"use tokio_stream::{self as stream, StreamExt};
#[tokio::main]async fn main() { let prices = stream::iter(vec![100, 102, 98, 105, 110]);
let mut doubled = prices.map(|p| p * 2).take(3);
while let Some(p) = doubled.next().await { println!("{p}"); }}这段代码会打印 200、204 和 196。它读起来和迭代器代码一模一样，只是 next() 后面多了 .await。要把 tokio 信道的接收端变成真正的 stream，就把它包起来：tokio_stream::wrappers::ReceiverStream::new(rx)。（futures crate 也有自己的 StreamExt，思路相同。每个文件只选其中一个，因为两个 trait 都有 next() 方法，会发生冲突。）
用大白话讲 Pin
Section titled “用大白话讲 Pin”迟早有一天，你会在某个类型或者错误信息里看到 Pin。下面只讲思路，不讲高深的理论。
Rust 编译 async fn 时，会把它变成一个隐藏的结构体，也就是一个状态机。每个在 .await 之后还要用到的局部变量，都会成为这个结构体的字段。现在想象有一个局部变量借用了另一个局部变量，比如 let name = String::from("Ana"); let r = &name;，而两者之间有一个 .await。这时结构体里就保存着一个指向自己字段的指针。如果你把这个结构体移动到内存中的另一个位置，那个指针就会指向原来那个已经空了的地方。
Pin 就是解决办法。被 pin 住的值是一个承诺：“这个值不会再在内存里移动了。”future 一旦被 pin 住，poll 它就是安全的。大多数时候你根本察觉不到，因为 .await 会替你 pin 住 future。
你会在三种情况下遇到 Pin：
- 多次 poll 同一个 future，比如
select!循环里的截止时间。 - 存储 future，放进
Vec或结构体字段里，通常写成Pin<Box<dyn Future<Output = T>>>。 - 在
async fn里递归。
先看第一种。如果你想在循环里重复使用一个 sleep，却没有 pin 住它，会发生这样的事：
let deadline = sleep(Duration::from_millis(300));
loop { tokio::select! { _ = &mut deadline => break, Some(n) = rx.recv() => println!("{n}"), }}error[E0277]: `PhantomPinned` cannot be unpinned | = note: consider using the `pin!` macro consider using `Box::pin` if you need to access the pinned value outside of the current scope = note: required for `Sleep` to implement `Unpin`编译器甚至直接告诉了你修复方法：考虑使用 pin! 宏。std::pin::pin!（或者 tokio::pin!）会把 future 原地 pin 在栈上。现在循环的每一轮共用同一个截止时间：
use std::pin::pin;use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8); tokio::spawn(async move { for order in 1..=3 { tx.send(order).await.unwrap(); sleep(Duration::from_millis(10)).await; } sleep(Duration::from_secs(5)).await; // 然后店里就安静下来了 });
let mut deadline = pin!(sleep(Duration::from_millis(300)));
loop { tokio::select! { _ = &mut deadline => { println!("打烊了"); break; } Some(order) = rx.recv() => println!("收到订单 #{order}"), } }}输出：
收到订单 #1收到订单 #2收到订单 #3打烊了如果我们把 sleep(...) 写在 select! 里面，每来一张新订单都会重新开始一个 300ms 的计时器，生意好的店就永远打不了烊。在循环外 pin 住一个截止时间，就解决了这个问题。
至于递归，编译器需要知道 future 有多大，而一个包含自己的 future 会无限大。在 async fn countdown 里调用 countdown(n - 1).await 会得到：
error[E0733]: recursion in an async fn requires boxing | = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future意思是：async fn 里的递归需要装箱，必须引入 Box::pin 这样的间接层。Box::pin 会把内层的 future 移到堆上并 pin 在那里。这样外层的 future 只保存一个指针，大小就固定了：
async fn countdown(n: u32) { if n == 0 { println!("发射！"); return; } println!("{n}..."); Box::pin(countdown(n - 1)).await;}
#[tokio::main]async fn main() { countdown(3).await;}输出：
3...2...1...发射！说实话，大多数人需要的也就这些：要多次 poll 的 future 用 pin!，存储 future 或递归用 Box::pin。如果你想了解完整的来龙去脉，“延伸阅读”里那一章 Rust Book 会一步一步讲清楚。
trait 中的 async fn 与异步闭包
Section titled “trait 中的 async fn 与异步闭包”从 Rust 1.75 开始，你可以直接在 trait 里写 async fn。从 Rust 1.85 开始，还有了异步闭包，写作 async |x| ...，以及配套的 trait AsyncFn、AsyncFnMut 和 AsyncFnOnce（也就是闭包 trait 的异步版本）。下面一个例子把两者都用上了：
use std::time::Duration;use tokio::time::sleep;
trait Notifier { async fn notify(&self, user: &str) -> String;}
struct Email;
impl Notifier for Email { async fn notify(&self, user: &str) -> String { sleep(Duration::from_millis(10)).await; // 假装在和邮件服务器通信 format!("邮件已发送给 {user}") }}
async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) { for user in users { println!("{}", job(user).await); }}
#[tokio::main]async fn main() { let email = Email; let users = ["ana", "ben"];
for_each_user(&users, async |user: &str| email.notify(user).await).await;}输出：
邮件已发送给 ana邮件已发送给 bentrait 里的 async fn 有一个坑。trait 并不保证返回的 future 是 Send 的，而 tokio::spawn 需要 Send 的 future（因为它们可能在线程之间移动）。所以通过泛型 N: Notifier 来 spawn 一次调用会失败：
fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) { tokio::spawn(async move { let msg = notifier.notify("ana").await; println!("{msg}"); });}error: future cannot be sent between threads safely |7 | let msg = notifier.notify("ana").await; | ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send` |help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify` |2 - async fn notify(&self, user: &str) -> String;2 + fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;编译器给出的建议就是通常的修复方法：把 trait 方法写成 fn ... -> impl Future<Output = String> + Send。实现者仍然可以写普通的 async fn。另外要注意，带 async fn 的 trait 目前还不能作为 dyn Notifier 这样的 trait 对象使用。如果你需要这样做，流行的 async-trait crate 会替你把 future 装箱。
阻塞运行时。在异步任务里调用 std::thread::sleep、做繁重的计算，或者用 std::fs 读取大文件，都会冻住整个工作线程，上面的其他任务都只能干等。基础篇展示过 sleep 的情况。对于 CPU 密集或者会阻塞的工作，交给 tokio::task::spawn_blocking，它会在专门为此准备的另一组线程上运行：
fn count_primes(limit: u64) -> usize { (2..limit) .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)) .count()}
#[tokio::main]async fn main() { let handle = tokio::task::spawn_blocking(|| count_primes(50_000));
println!("一边数质数，一边照常响应请求……");
let primes = handle.await.unwrap(); println!("小于 50000 的质数有 {primes} 个");}输出：
一边数质数，一边照常响应请求……小于 50000 的质数有 5133 个跨越 .await 持有 std::sync::Mutex 的锁。这段代码看起来人畜无害：
let handle = tokio::spawn(async move { let mut count = visits.lock().unwrap(); *count += 1; save_to_disk(*count).await;});error: future cannot be sent between threads safely | = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`note: future is not `Send` as this value is used across an await |14 | let mut count = visits.lock().unwrap(); | --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`15 | *count += 1;16 | save_to_disk(*count).await; | ^^^^^ await occurs here, with `mut count` maybe used later编译器是在保护你。这个任务在 .await 处休眠时仍然持有锁，其他任何想要这把锁的任务都会阻塞它所在的整个线程。修复方法是在 .await 之前释放锁，把加锁的代码放进单独的块里：
let count = { let mut count = visits.lock().unwrap(); *count += 1; *count}; // 锁在这里被释放save_to_disk(count).await;如果你确实需要在等待时一直持有锁，就用 tokio::sync::Mutex，它的 lock().await 等待时不会阻塞线程。（关于锁的更多内容，见共享状态。）
忘了写 .await。future 是惰性的，所以不带 .await 的 save_file(); 什么都不会做。编译器会用“futures do nothing unless you .await or poll them”警告你，意思是“除非你 .await 或 poll 它们，否则 future 什么都不做”。完整的解释见 Async/Await，千万别忽略这个警告。
spawn 了任务，却从不等待它。main 返回时，运行时就会关闭，所有没完成的任务都会被丢弃：
use std::time::Duration;use tokio::time::sleep;
#[tokio::main]async fn main() { tokio::spawn(async { sleep(Duration::from_millis(100)).await; println!("报告已保存"); // 永远不会打印！ });
println!("main 结束了");}输出：
main 结束了报告永远不会被保存。如果这项工作很重要，就留着 JoinHandle 并在 main 结束前 .await 它（或者把任务放进一个 JoinSet，然后把它清空）。
1. 带截止时间的比价
Section titled “1. 带截止时间的比价”四家商店分别在 20、40、300 和 60 毫秒后回复报价请求。用 JoinSet 同时询问这四家，但每家只给 150ms。打印收到的价格（按商店名排序），再打印有几家超时了。提示：把 timeout(limit, ask_shop(name, delay)) spawn 进集合。这样每个完成的任务都会给你一个 Ok 或 Err(Elapsed)，所以把商店名和这个结果一起返回。
查看答案
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::{sleep, timeout};
async fn ask_shop(delay_ms: u64) -> u32 { sleep(Duration::from_millis(delay_ms)).await; 100 + delay_ms as u32 // 一个编出来的价格}
#[tokio::main]async fn main() { let shops = [("Corner Shop", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bits & Bobs", 60)]; let limit = Duration::from_millis(150);
let mut set = JoinSet::new(); for (name, delay) in shops { set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) }); }
let mut prices = Vec::new(); let mut timed_out = 0; while let Some(res) = set.join_next().await { match res.unwrap() { (name, Ok(price)) => prices.push((name, price)), (_, Err(_)) => timed_out += 1, } }
prices.sort(); for (name, price) in prices { println!("{name}: {price}"); } println!("{timed_out} 家商店超时了");}输出：
Bits & Bobs: 160Corner Shop: 120MegaMart: 1401 家商店超时了商店名保留了英文，因为结果是按名字排序的，英文名的字母顺序在每种语言里都一样。所有商店同时等待，所以不管 SlowCo 有多慢，整个比价只需要大约 150ms。它的超时一触发，它的 future 就被丢弃了，不可能继续在后台运行。
2. 用异步递归计算文件夹大小
Section titled “2. 用异步递归计算文件夹大小”一个 Folder 有名字、一组以 KB 为单位的文件大小，以及一组子文件夹。写一个 async fn total_size(folder: &Folder) -> u64，把它自己的文件加上每个子文件夹的大小相加，对每个子文件夹调用自己。假装每个文件夹都需要先做一次简短的异步“磁盘读取”，用 tokio::task::yield_now().await 来模拟。提示：递归调用外面需要套上 Box::pin(...)，否则会得到 E0733 错误。
查看答案
struct Folder { name: &'static str, files_kb: Vec<u64>, subfolders: Vec<Folder>,}
async fn total_size(folder: &Folder) -> u64 { tokio::task::yield_now().await; // 假装在读磁盘
let mut total: u64 = folder.files_kb.iter().sum(); for sub in &folder.subfolders { total += Box::pin(total_size(sub)).await; } println!("{}: {} KB", folder.name, total); total}
#[tokio::main]async fn main() { let home = Folder { name: "home", files_kb: vec![4], subfolders: vec![ Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] }, Folder { name: "code", files_kb: vec![12, 8], subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }], }, ], };
let total = total_size(&home).await; println!("总计：{total} KB");}输出：
photos: 3072 KBtarget: 500 KBcode: 520 KBhome: 3596 KB总计：3596 KB每个文件夹都在它所有的子文件夹之后才打印，因为它得先等子文件夹的大小算出来。Box::pin 把每次嵌套调用的 future 放到堆上，所以外层的 future 始终保持固定大小。
3. 先清空队列，再停下
Section titled “3. 先清空队列，再停下”修改前面停止信号的例子，让它在发送停止信号之前不再需要 sleep(100ms)。立刻发送三个任务和停止信号，但要确保 worker 在停下之前仍然完成所有排队的任务。提示：没有额外设置时，如果两个分支都就绪了，select! 会随机挑一个。加上 biased;，并把任务分支放在最前面。
查看答案
use tokio::sync::{mpsc, oneshot};
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
for job in ["调整照片尺寸", "发送邮件", "更新搜索索引"] { job_tx.send(job.to_string()).await.unwrap(); } stop_tx.send(()).unwrap();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { biased; Some(job) = job_rx.recv() => { println!("正在处理：{job}"); done += 1; } _ = &mut stop_rx => { println!("队列已清空，停止"); break; } } } done });
let done = worker.await.unwrap(); println!("worker 一共完成了 {done} 个任务");}输出：
正在处理：调整照片尺寸正在处理：发送邮件正在处理：更新搜索索引队列已清空，停止worker 一共完成了 3 个任务有了 biased;，select! 总是先尝试任务分支，所以只有当 recv() 没有任何就绪的消息时，它才会去看停止信号。如果不加，worker 可能处理一两个任务就停了，而且这种 bug 只会偶尔冒出来。
join!等待所有 future，try_join!遇到第一个Err就停下，JoinSet运行许多 spawn 出来的任务，并在它们完成时交回结果（如果顺序重要就排个序）。select!等待第一个完成的 future，其余的会被丢弃，这就是异步 Rust 里取消的运作方式。在循环里，使用取消安全的 future；顺序重要时加上biased;。timeout会放弃太慢的 future。interval按稳定的节奏 tick，sleep只等一次。- stream 是异步迭代器。信道接收端用起来就像一个 stream，
tokio-stream提供了map、take等方法。 Pin承诺 future 不会在内存里移动。多次 poll 同一个 future 用pin!，存储 future 和异步递归用Box::pin。- 别阻塞运行时（用
spawn_blocking），别跨越.await持有std的锁，spawn 出去的任务记得 await。
小测验
1. 在 tokio::select! 里，没有最先完成的那些分支会怎样？
2. 时间用完时，tokio::time::timeout 会返回什么？
3. 为什么 async fn 调用自己时需要 Box::pin？
4. 你要在一个 tokio 程序里做 2 秒钟的 CPU 密集型工作。最合适的工具是什么？
得分: 0 / 4
- Select — Tokio 教程（英文）：详细讲解
select!：分支如何工作、取消意味着什么，以及如何在循环里使用它。 - Streams — Tokio 教程（英文）：如何用
tokio-stream使用、改造和构建 stream。 - 深入理解 async 相关的 traits — Rust 程序设计语言：循序渐进地介绍
Future、Pin、Unpin和Stream。 - std::pin — Rust 标准库（英文）：官方对 pin、
Pin、Unpin和pin!宏的解释。 - docs.rs 上的 tokio::time（英文）：
sleep、timeout、interval，以及错过的 tick 如何处理。