阶段四 · 并发编排与 IO

任务编排:join 与 select

一句话总结

在同一个任务内并发跑多个 Future,有两种工具:join! 等全部完成,select! 取最先完成的那个(其余取消).再加上 timeout 给操作设上限.
它们和 spawn 的区别是:不新建任务,而是在当前任务里编排多个 Future.

spawn 之外的另一种并发

第 06 篇用 spawn 实现并发--把 Future 交给运行时变成独立任务.本篇的 join! / select! 是另一条路:它们在当前任务内部同时驱动多个 Future,不创建新任务,也因此不要求 Send(回顾第 07 篇,Send 约束来自跨线程调度,而这里始终在同一任务里).

spawn join! / select!
是否新建任务 是,独立调度 否,在当前任务内编排
Send 要求 需要(可能跨线程) 不需要(同任务)
适合 互相独立,生命周期长的任务 当前任务里"等几件事"的组合

join!:等待全部完成

join! 同时推进多个 Future,等它们全部就绪后,一并返回结果元组.这是"并行做几件事,都要结果"的标准写法:

use tokio::join;

async fn load() {
    // 三个请求并发进行,总耗时 ≈ 最慢的那个,而非三者之和
    let (user, posts, stats) = join!(
        fetch_user(),
        fetch_posts(),
        fetch_stats(),
    );
    // 到这里三者都已完成
}

把 a().await; b().await;(顺序,耗时相加)换成 join!(a(), b())(并发,耗时取最大),就像从"排队办完一件再办下一件"变成"几个窗口同时办".
这是消除"假串行"的最直接手段--回顾第 06 篇:相邻 await 并不会自动并发,join! 才会.

如果其中任意一个失败就该整体放弃,用 try_join!:任一 Future 返回 Err 时立即返回该错误,不等其余.

use tokio::try_join;

let (a, b) = try_join!(fetch_a(), fetch_b())?; // 任一出错即短路返回

select!:取最先完成的一个

select! 同时等多个 Future,哪个先就绪就处理哪个,然后丢弃其余.它是实现超时,取消,多路事件响应的核心:

use tokio::select;

async fn worker(mut shutdown: tokio::sync::watch::Receiver<bool>) {
    loop {
        select! {
            data = receive_data() => {
                process(data).await;
            }
            _ = shutdown.changed() => {
                println!("收到关闭,退出");
                break;
            }
        }
    }
}

select! 像同时盯着几部电话,哪部先响就接哪部,接起一部的同时挂断其余.
"其余被丢弃"这点至关重要:那些没选中的 Future 会被 drop 掉,等于取消--这正是第 13 篇"drop 即取消"的伏笔.

select! 的取消陷阱

没被选中的分支 Future 会被取消(drop).如果某个分支做到一半被丢弃(比如读了一半数据),可能丢失中间状态.

要求"被取消也不破坏状态"的特性叫取消安全(cancellation safety).Tokio 文档会标注每个方法是否取消安全--在 select! 的循环里,务必使用取消安全的操作.这是 select! 最需要小心的地方.

timeout:给操作设上限

超时是 select! 思想的常见特化,Tokio 直接封装成 timeout:在规定时间内没完成就返回错误.

use tokio::time::{timeout, Duration};

match timeout(Duration::from_secs(3), fetch_data()).await {
    Ok(data) => println!("按时拿到:{data:?}"),
    Err(_)   => println!("超过 3 秒,放弃"), // 超时,fetch_data 被取消
}

超时触发时,被包裹的 fetch_data() 同样会被 drop 取消.这比 Go 用 context.WithTimeout + select 手动拼更简洁--但取消语义的注意事项是一致的.

如何选择

要在当前任务里同时处理多个 Future:
        │
        ├── 全都要完成,都要结果 ─────▶ join!  (任一失败即止 → try_join!)
        │
        ├── 只要最先完成的那个 ───────▶ select!
        │
        └── 给单个操作加时间上限 ─────▶ timeout

  若任务彼此独立,需各自长期运行 ──▶ 回到 spawn(第 06 篇)

快速回顾

  • 与 spawn 之别:join!/select! 在当前任务内编排多个 Future,不新建任务,无 Send 要求.
  • join!:并发推进,等全部完成,返回结果元组,消除"假串行";try_join! 任一出错即短路.
  • select!:取最先就绪的分支,丢弃(取消)其余;用于超时,关闭信号,多路事件.
  • 取消安全:select! 中未选中的分支被 drop,循环里务必用取消安全的操作,避免丢失中间状态.
  • timeout:给操作设时间上限,超时即取消被包裹的 Future,比手搓 select 简洁.

动手练习

  1. join! 计时:用 join! 并发执行三个不同时长的 sleep 任务,计时验证总耗时 ≈ 最长者而非三者之和.
  2. 串行对比:把上题改成相邻 .await,对比耗时,直观感受 join! 的并发收益.
  3. select! 关闭信号:用 select! 写一个循环:同时等"数据到达"和"关闭信号",收到关闭则退出.
  4. timeout 超时:用 timeout 给一个会睡 5 秒的操作设 2 秒上限,观察返回 Err 且操作被取消.
  5. 取消安全:查阅 Tokio 文档,找出一个被标注为"取消安全"和一个"非取消安全"的方法,说明区别.