Tokio 运行时

Tokio 提供任务调度、异步 I/O、计时器及同步工具。执行器轮询 Future，驱动等待操作系统事件并唤醒任务；用户代码要通过适合异步环境的操作主动让出执行权。

Tokio 任务是轻量异步任务，可以在多个线程间调度，不应写成“是真并发，所以不是协程”。任务如何保存状态与运行在几个 OS 线程上，是不同维度。

两种主要调度方式

current_thread 在调用 block_on 的线程上驱动任务，仍可并发处理大量等待；multi_thread 使用多个 worker，并具有工作窃取等机制。默认 worker 数与可用 CPU 等配置有关，可被覆盖，不是保证一个物理核心永久绑定一个 worker。

I/O 驱动的平台实现和具体队列算法属于版本相关细节，理解概念时不必把每种 socket、文件、定时器都硬套进同一个系统调用。部分文件操作会借助阻塞池，而非直接变成内核 readiness 通知。Runtime

current_thread 的运行时在未被 block_on 驱动时，关联任务不会凭空继续执行。多线程运行时的 block_on 主 Future 也不是简单地等于某一个 worker 上的普通 spawn 任务。

一个完整的有界生产/消费程序

Cargo.toml：

[package]
name = "tokio-demo"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] }

src/main.rs：

use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (tx, mut rx) = mpsc::channel(2);
    let producer = tokio::spawn(async move {
        for value in 1_u32..=4 {
            if tx.send(value).await.is_err() {
                return; // 消费者已结束，不再生产
            }
        }
    });

    let mut total = 0;
    while let Some(value) = rx.recv().await {
        total += value;
    }
    producer.await?;
    assert_eq!(total, 10);
    println!("{total}");
    Ok(())
}

容量 2 限制排队项，send 在队列满时等待；生产者结束并释放最后一个 sender 后，消费者排空缓冲再得到 None。producer.await 显式回收任务结果，避免把 panic 等失败当成“任务自然结束”。

bounded channel 不会限制生产前已经创建的无限任务，设计时要让背压覆盖到真正消耗资源的入口。

队列有界，还要看排队项之外的内存

假设队列容量 100，单条消息最多 1 MiB，仅消息负载就可能接近 100 MiB；此外还有生产者正在构造的值、正在处理的值和任务状态。容量按项计数，不自动限制每项字节数，也不限制上游已创建的所有 Future。

如果先无限 spawn，再在每个任务内部等待信号量，资源密集区的并发可能受限，但等待任务仍可无限积累。要控制任务数量，应在创建任务之前取得许可，或等已有任务完成再接收/创建下一项工作；结果也要及时回收。Semaphore

spawn 的 Send 与生命周期

tokio::spawn 通常要求 Future 与其输出满足 Send + 'static，因为任务可独立存活并可能跨 worker 移动。即使用 current_thread 调度器，这个 API 的类型要求也不会自动放宽。

需要 !Send 局部任务时，可按 Tokio 的局部任务机制使用 LocalSet 与 spawn_local，而不是 unsafe 把 Rc 包装成 Send。临时 Rc 或同步锁 guard 不跨挂起点存活时，又可能不阻碍整个 Future 的 Send；清楚的词法作用域通常最易表达这一点。

使用局部任务处理 Rc

下面独立程序使用本篇同一 Cargo 依赖。LocalSet 保证其中 spawn_local 的任务在局部执行环境运行，但 spawn_local 仍需要任务拥有的捕获满足 'static；它并不是普通借用作用域线程的替代品。

use std::{cell::RefCell, rc::Rc};
use tokio::task::LocalSet;
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let local = LocalSet::new();
    local.run_until(async {
        let values = Rc::new(RefCell::new(vec![1]));
        let shared = Rc::clone(&values);
        let task = tokio::task::spawn_local(async move {
            tokio::task::yield_now().await;
            shared.borrow_mut().push(2);
        });
        task.await.unwrap();
        assert_eq!(&*values.borrow(), &[1, 2]);
    }).await;
}

Rc 在这个任务中跨 await 存活，所以不适合 tokio::spawn，但适合明确的局部任务模型。RefCell 借用仍应在合理范围结束，LocalSet 不会消除单线程重入冲突。LocalSet

JoinHandle 的 Drop 不等于任务取消

丢弃 Tokio JoinHandle 通常使任务脱离等待，任务仍可继续执行。abort 请求取消任务，在合适的调度边界生效；等待 handle 可以观察取消结束。一个永不让出的长 poll 不会因为另一个任务调用 abort 就立即被抢占。JoinHandle

这与丢弃一个尚未 spawn、由当前任务直接拥有的 Future 不同。资源清理和后台工作退出需要明确的所有权与协作协议，必要时使用关闭信号并等待完成。

超时等待 JoinHandle 后，明确请求终止并回收

use std::{sync::{Arc, atomic::{AtomicBool, Ordering}}, time::Duration};
use tokio::{sync::oneshot, time::timeout};
struct MarkDropped(Arc<AtomicBool>);
impl Drop for MarkDropped {
    fn drop(&mut self) { self.0.store(true, Ordering::SeqCst); }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let dropped = Arc::new(AtomicBool::new(false));
    let flag = Arc::clone(&dropped);
    let (ready_tx, ready_rx) = oneshot::channel();
    let mut task = tokio::spawn(async move {
        let resource = MarkDropped(flag);
        ready_tx.send(()).unwrap();
        std::future::pending::<()>().await;
        drop(resource);
    });
    ready_rx.await.unwrap(); // 确认资源已构造，任务会停在 pending。
    assert!(timeout(Duration::from_millis(1), &mut task).await.is_err());
    assert!(!dropped.load(Ordering::SeqCst));
    task.abort();
    assert!(task.await.unwrap_err().is_cancelled());
    assert!(dropped.load(Ordering::SeqCst));
}

timeout 借用 &mut task，超时后仍保留 JoinHandle，因而可以决定继续等待、协作关闭或 abort。若把 handle 按值交给 timeout，超时会丢弃等待对象中的 handle，后台任务却通常继续存在。上例的 Pending 永远不完成，取消结果是确定的；一般任务在 abort 生效前可能已经正常完成，处理代码应允许这种竞态。

1 毫秒这里只用于触发超时分支，不是计时精度承诺。timeout 无法抢占一次始终不返回的 poll；耗时计算仍可能越过截止时间继续运行。它限制等待协议，不是操作系统级强制终止。timeout

丢掉等待者，后台任务会停吗？

相同的 drop 语法，效果取决于被丢弃的对象拥有什么。直接 Future 拥有操作状态，JoinHandle 只是 spawned 任务的等待和控制入口。先试丢弃 handle，再重置并请求 abort；对照图中任务生命的变化，便能区分“停止等待”“请求取消”和“取消已经完成”。

select! 的取消边界看操作数所有权

select! 在当前任务中轮询多个分支，并非自动把每个分支 spawn 到不同线程。选出分支后，其他局部 Future 被丢弃；若分支只是借用了外部固定的 Future，它的底层对象仍可能由外部继续持有。

若某分支拥有一个 JoinHandle，丢弃它只会脱离任务，不自动停止该任务。若丢弃的是已经消费部分输入的 read_exact Future，进度又可能丢失。重复 select 循环必须核对各操作的取消安全性。select!

默认分支轮询顺序与 biased 模式有不同公平性考虑，biased 也不能保证业务公平。分支包含阻塞代码时会阻塞整个 select 的推进。

JoinSet 将任务创建与结果回收放在同一个循环

use tokio::task::{JoinError, JoinSet};
fn accept(result: Result<Result<u32, &'static str>, JoinError>,
          values: &mut Vec<u32>, rejected: &mut usize) {
    match result {
        Ok(Ok(value)) => values.push(value),
        Ok(Err("invalid input")) => *rejected += 1,
        other => panic!("未预期的任务结果：{other:?}"),
    }
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
    let mut tasks = JoinSet::new();
    let mut values = Vec::new();
    let mut rejected = 0;
    for value in 0u32..5 {
        if tasks.len() >= 2 {
            accept(tasks.join_next().await.unwrap(), &mut values, &mut rejected);
        }
        tasks.spawn(async move {
            tokio::task::yield_now().await;
            if value == 3 { Err("invalid input") } else { Ok(value * 2) }
        });
        assert!(tasks.len() <= 2);
    }
    while let Some(result) = tasks.join_next().await {
        accept(result, &mut values, &mut rejected);
    }
    values.sort_unstable();
    assert_eq!(values, [0, 2, 4, 8]);
    assert_eq!(rejected, 1);
}

这里最多保留两个尚未回收的任务。JoinSet 按完成顺序交出结果，不保证输入顺序，所以示例在验证时排序；业务若需要原顺序，可以携带序号并重排，同时考虑重排缓冲上限。输出 Vec 在这个有限例子中安全可控，面对无限输入则应改为逐项交付结果。

返回值有两层失败：外层 JoinError 表示任务取消或 panic，内层 Err 是业务主动返回的失败。一次 .await? 只传播外层，不能据此断言业务成功。JoinSet

JoinSet 被丢弃会请求中止其任务，这与丢弃单个 JoinHandle 不同；若需要确认所有析构完成，应使用并等待相应关闭/回收流程。单纯发出中止请求不等于已经完成资源回收。

阻塞工作与 CPU 工作

std::thread::sleep、阻塞库调用和长计算不应直接长时间占住异步 worker。有限阻塞任务可以用 spawn_blocking，但其已开始运行的闭包通常无法被 abort 取消；运行时关闭也要考虑它的结束行为。spawn_blocking

阻塞池有自己的线程与队列策略，不是自动为无限 CPU 工作提供合理并发上限。大量 CPU 计算应明确限流或使用合适的专用执行池，并以目标负载验证。

锁不要只按“async 里就必须异步锁”选择

短小、低争用且不跨 await 的临界区可以使用普通 Mutex。持锁跨 await 容易阻塞其他任务或引入死锁；std MutexGuard 还可能使 Future 不满足 Send。

Tokio 的异步 Mutex 允许异步等待锁，并可在需要时跨 await 持 guard，但这不免除业务死锁和长时间独占问题。共享 I/O 资源有时更适合由一个任务拥有，用消息传递发请求。Tokio 共享状态

诊断延迟时区分：排队等待、调度未获运行、CPU 执行、I/O 等待与锁等待。仅看到请求慢，无法确定是 worker 数太少还是下游能力不足。

关闭顺序与可观察性

有界消费者若等到所有 sender 都消失才退出，就必须知道每个 sender 克隆由谁持有。留在长寿命结构中的多余克隆可能让 recv 永远等不到 None。通过 Receiver::close 主动停止新发送后，仍应继续接收并处理已缓冲消息；先前取得的许可还需要按通道契约处理，不能将 close 理解为瞬间清空一切。mpsc Receiver::close

服务关闭可以依次停止入口、发出协作信号、处理已接收工作、回收任务，再结束运行时。对每步记录待处理数量、最老任务年龄与剩余预算，才能分辨是生产者没停、某个任务没让出，还是下游仍在等待。

手动构造 Runtime 时，还要启用所需的时间或 I/O 驱动；“创建成功”不意味着任何依赖驱动的 Future 都可运行。运行时句柄的进入上下文与实际持续驱动运行时也不同，尤其在 current_thread 环境中应保持明确的 block_on 所有权。