Rust 语言只定义了 Future、async/await 和 Waker 等异步基础设施，并没有在标准库中提供完整的异步运行时。Tokio 补齐了这一层：它提供任务调度器、网络 I/O 驱动、定时器、异步同步原语、异步文件接口以及阻塞任务隔离机制。

如果把 Rust 异步程序比作一座城市：

Future            = 等待执行的工作
Executor/Scheduler = 安排工作在哪个线程运行
I/O Driver         = 监听 socket 是否就绪
Timer Driver       = 管理定时器何时到期
Waker              = 通知某个任务可以继续
Tokio              = 将上述组件组合成运行时

本文不再重复 Future::poll 和状态机的基础推导，而是聚焦 Tokio 本身：runtime 如何启动，任务如何调度，网络 I/O 为什么不会阻塞线程，以及生产代码中应该怎样处理并发、超时、共享状态、CPU 密集任务和优雅退出。

一、Tokio 提供了什么

一个典型 Tokio 应用依赖：

[dependencies]
tokio = { version = "1", features = ["full"] }

full 适合学习和应用开发，但库作者通常应该只启用实际需要的 feature，缩短编译时间并减少依赖面。

Tokio 的主要组成包括：

组件作用
Runtime组合调度器、I/O driver 和 timer driver
Task由 runtime 调度的异步任务
tokio::netTCP、UDP、Unix Socket 等异步网络接口
tokio::timesleep、interval、timeout
tokio::syncchannel、Mutex、RwLock、Semaphore、Notify
tokio::fs文件系统异步接口
spawn_blocking将阻塞或 CPU 密集工作移出异步 worker
select!同时等待多个异步分支

需要先建立一个重要认识：

Tokio 不会让普通阻塞函数自动变成异步函数。只有与 runtime 协作、在等待时返回 Pending 的操作，才能释放 worker thread。

二、#[tokio::main] 做了什么

最常见入口是：

#[tokio::main]
async fn main() {
    println!("hello tokio");
}

它大致展开为：

fn main() {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap();

    runtime.block_on(async {
        println!("hello tokio");
    });
}

enable_all() 启用 I/O 和时间驱动。block_on 让当前同步线程进入 runtime，持续推进根 Future，直到它完成。

Tokio 支持两种主要调度器：

multi_thread：多个 worker thread，默认用于服务端程序
current_thread：单线程运行所有异步任务

可以显式配置：

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
    // ...
}

单线程 runtime 并不意味着一次只能处理一个连接。大量任务仍能在 I/O 等待点交错执行；只是任何长时间不让出执行权的任务都会阻塞全部任务。

三、Task 为什么比线程轻量

tokio::spawn 创建的是异步 Task，不是操作系统线程：

let handle = tokio::spawn(async {
    42
});

let value = handle.await.unwrap();
assert_eq!(value, 42);

每个 Task 主要保存：

  • Future 状态机；
  • 调度状态和引用计数；
  • Waker 所需信息；
  • 输出或 panic 结果。

Task 在 .await 返回 Pending 后不占用线程。线程可以继续执行另一个 ready task，因此少量 worker 可以承载大量主要等待网络的任务。

tokio::spawn 要求 Future 通常满足：

Future + Send + 'static

'static 不代表任务一定存活到程序结束，而是任务不能借用可能提前失效的栈变量。常见做法是使用 async move 转移所有权：

let request_id = String::from("req-42");

let handle = tokio::spawn(async move {
    process(request_id).await
});

如果确实需要执行非 Send Future，可以使用 LocalSet 配合 current-thread runtime，但不要把它当成规避所有权设计的默认方案。

四、调度器怎样工作

multi-thread scheduler 的核心思想是：

每个 worker 有本地任务队列
        │
        ├── 优先执行自己的 ready task
        ├── 必要时读取全局注入队列
        └── 空闲时从其他 worker 偷任务

这种方式称为 work stealing。任务被唤醒后重新进入 ready queue，worker 再次 poll 它。

概念执行过程：

Worker 0 poll Task A
  -> A 等待 socket，返回 Pending
  -> Worker 0 转去 poll Task B

I/O driver 发现 A 的 socket ready
  -> 调用 A 的 Waker
  -> A 回到 ready queue
  -> 某个 worker 再次 poll A

Tokio 还需要避免单个总是 ready 的任务无限霸占线程。协作式调度依赖任务经常遇到 .await，Tokio 的部分资源操作也带有 cooperative budget。极端计算循环应显式让出：

loop {
    do_small_piece_of_work();
    tokio::task::yield_now().await;
}

不过如果工作本质上是长时间 CPU 计算，更正确的方案通常是 spawn_blocking、Rayon 或独立计算服务，而不是不断 yield_now()。

五、I/O Driver 为什么能管理大量连接

Linux 上 Tokio 通常基于 epoll，其他平台使用对应的操作系统事件机制。以 TCP 读取为例：

Task 调用 AsyncRead::poll_read
  -> 尝试非阻塞 read
  -> 数据尚未到达，得到 WouldBlock
  -> 向 I/O driver 登记兴趣和 Waker
  -> 返回 Pending

网卡收到数据
  -> 内核将 fd 标记为 readable
  -> epoll 通知 Tokio I/O driver
  -> driver 唤醒相关 Task
  -> Task 再次 poll_read
  -> read 成功

线程没有在 read 上睡眠等待。Task 只是暂停，worker 可以处理其他连接。

Tokio I/O 类型应使用 tokio::net，不要在 async task 中直接使用阻塞式 std::net 读写。已有标准库 socket 可切换为 nonblocking 后转换为 Tokio 类型，但必须满足 API 要求。

六、Timer Driver 与取消

tokio::time::sleep 也不会为每个定时器创建线程：

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

sleep(Duration::from_millis(100)).await;

sleep Future 把 deadline 注册到 timer driver，然后返回 Pending。时间到达时，driver 唤醒 Task。

许多 Tokio Future 具有取消安全边界：当 Future 被 drop，表示调用方不再等待它。但取消不是“向任意代码注入异常”，而是停止继续 poll 并释放 Future 保存的状态。

这意味着：

  • drop 一个尚未开始写入的 Future 可能什么也没发生；
  • drop 一个已产生外部副作用的 Future 不会自动回滚；
  • select! 循环中必须确认被取消操作是否 cancel-safe；
  • 业务事务需要自己定义幂等、补偿或提交边界。

七、例子一：正确并发执行两个请求

顺序写法：

let user = fetch_user().await?;
let orders = fetch_orders().await?;

总延迟接近两次请求之和。如果两者互不依赖，可使用 join!：

use tokio::try_join;

async fn load_page() -> anyhow::Result<Page> {
    let (user, orders) = try_join!(
        fetch_user(),
        fetch_orders(),
    )?;

    Ok(Page { user, orders })
}

join! 和 try_join! 在当前 Task 中并发 poll 多个 Future，不会额外创建 Task；try_join! 遇到第一个错误会返回。

如果需要独立任务、独立生命周期或跨线程调度，再使用 spawn：

let user_task = tokio::spawn(fetch_user());
let order_task = tokio::spawn(fetch_orders());

let user = user_task.await??;
let orders = order_task.await??;

最佳实践：

  • 只为“并发”优先用 join!；
  • 需要后台执行或独立取消时使用 spawn；
  • 不要为了让代码看起来异步而无条件 spawn 每个函数；
  • 始终处理 JoinError，任务可能 panic 或被 abort。

八、例子二：超时与取消

网络调用必须有时间边界：

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

async fn load_with_timeout() -> anyhow::Result<Response> {
    let response = timeout(
        Duration::from_secs(2),
        fetch_remote_data(),
    )
    .await
    .map_err(|_| anyhow::anyhow!("request timed out"))??;

    Ok(response)
}

timeout 超时后 drop 内部 Future。是否会取消底层操作，取决于该 Future 的实现和副作用阶段。

同时等待响应和退出信号可用 select!：

use tokio::signal;

async fn run_service() -> anyhow::Result<()> {
    tokio::select! {
        result = serve() => result,
        _ = signal::ctrl_c() => {
            println!("shutdown requested");
            Ok(())
        }
    }
}

最佳实践：

  • 为外部 RPC、数据库请求和队列操作设置 deadline；
  • 区分 timeout、连接错误和业务错误；
  • 重试必须带退避、上限和幂等条件；
  • 不要在无限循环中无条件重试立即失败的请求；
  • select! 中循环使用同一个 Future 时，先确认取消安全性。

九、例子三：用 Semaphore 限制并发

一次 spawn 十万个任务虽然可能工作，但会同时占用连接、内存和下游容量。使用 Semaphore 建立背压：

use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio::task::JoinSet;

async fn process_all(items: Vec<Item>) -> anyhow::Result<()> {
    let limit = Arc::new(Semaphore::new(64));
    let mut tasks = JoinSet::new();

    for item in items {
        let permit = limit.clone().acquire_owned().await?;

        tasks.spawn(async move {
            let _permit = permit;
            process(item).await
        });
    }

    while let Some(result) = tasks.join_next().await {
        result??;
    }

    Ok(())
}

permit 在任务结束时自动释放。JoinSet 便于收集动态任务，并在对象 drop 时管理剩余任务。

更流式的场景也可以使用 StreamExt::buffer_unordered。

最佳实践：

  • 并发上限应来自下游容量，而不是 CPU 核数的机械倍数；
  • 对数据库连接池、HTTP client、文件描述符和内存同时考虑；
  • 有界 channel 和 Semaphore 是建立背压的常用工具；
  • 不要只限制 task 数量，却让每个 task 创建大量子任务。

十、例子四：共享状态与 Mutex

Tokio 同时提供异步 Mutex：

use std::{collections::HashMap, sync::Arc};
use tokio::sync::Mutex;

type Cache = Arc<Mutex<HashMap<String, String>>>;

async fn insert(cache: Cache, key: String, value: String) {
    let mut guard = cache.lock().await;
    guard.insert(key, value);
}

但“异步代码就必须使用 tokio::sync::Mutex”是误区。

选择原则：

  • 临界区很短且不会跨 .await：优先 std::sync::Mutex；
  • 必须持锁跨 .await：使用 Tokio Mutex，但重新考虑设计；
  • 高并发共享服务：优先消息传递或分片状态；
  • 读多写少：可评估 RwLock，不要假设一定更快。

危险写法：

let mut guard = state.lock().await;
call_remote_service().await;
guard.update();

远程调用期间一直持锁，会把一个网络延迟放大成全局串行瓶颈。更好的方式是缩短临界区：

let snapshot = {
    let guard = state.lock().await;
    guard.snapshot()
};

let result = call_remote_service(snapshot).await?;

{
    let mut guard = state.lock().await;
    guard.apply(result);
}

同时必须考虑两次加锁之间状态可能变化，必要时使用版本号或 compare-and-set 语义。

十一、例子五：阻塞操作必须隔离

错误示例：

async fn handler() {
    std::thread::sleep(Duration::from_secs(1));
}

这会阻塞 Tokio worker thread，而不是只暂停当前 Task。同样危险的还有：

  • 大文件的同步读写；
  • 阻塞式数据库客户端；
  • 长时间 CPU 压缩、加密和解析；
  • Mutex 竞争严重的同步调用；
  • 在 async 中执行外部命令并同步等待。

对于有限、不可避免的阻塞函数：

let output = tokio::task::spawn_blocking(move || {
    blocking_library_call(input)
})
.await??;

spawn_blocking 使用专门的 blocking thread pool，避免占住异步 worker。

但它不是无限 CPU 任务调度器。持续 CPU 密集工作应该：

  • 限制同时运行数量；
  • 使用 Rayon 等计算线程池；
  • 通过 channel 与 Tokio runtime 通信；
  • 必要时拆成独立服务。

另一个重要事实是：已经开始运行的 spawn_blocking 闭包通常不能靠 abort 强制停止。闭包应自行检查取消标志或被设计成有限时长操作。

十二、例子六：有界 Channel 与优雅退出

消息传递通常比共享可变状态更容易管理：

use tokio::sync::{mpsc, oneshot};

struct Command {
    key: String,
    reply: oneshot::Sender<anyhow::Result<String>>,
}

async fn manager(mut rx: mpsc::Receiver<Command>) {
    while let Some(command) = rx.recv().await {
        let result = handle(command.key).await;
        let _ = command.reply.send(result);
    }
}

调用侧：

let (reply_tx, reply_rx) = oneshot::channel();
command_tx.send(Command {
    key,
    reply: reply_tx,
}).await?;

let value = reply_rx.await??;

使用有界 mpsc::channel(capacity) 可以在消费者跟不上时让生产者等待，从而形成背压。无界 channel 可能把短期流量高峰变成持续内存增长。

服务退出时，可以使用 CancellationToken、广播 channel 或 watch channel 通知子任务，并用 JoinSet 等待它们完成：

use tokio_util::sync::CancellationToken;
use tokio::task::JoinSet;

async fn run() -> anyhow::Result<()> {
    let shutdown = CancellationToken::new();
    let mut tasks = JoinSet::new();

    for worker_id in 0..4 {
        let token = shutdown.child_token();
        tasks.spawn(async move {
            worker(worker_id, token).await
        });
    }

    tokio::signal::ctrl_c().await?;
    shutdown.cancel();

    while let Some(result) = tasks.join_next().await {
        result??;
    }

    Ok(())
}

最佳实践：

  • 停止接收新请求；
  • 通知后台任务退出；
  • 为 drain 设置最大等待时间；
  • flush 日志、指标和必要状态；
  • 超时后明确 abort 哪些仍可安全取消的任务。

十三、select! 的常见陷阱

select! 默认在多个就绪分支之间采用公平策略；也可以使用 biased; 指定从上到下优先检查。使用 biased 模式时，必须自己保证高频分支不会饿死 shutdown 分支。

循环中常见写法：

loop {
    tokio::select! {
        Some(message) = rx.recv() => handle(message).await,
        _ = shutdown.cancelled() => break,
    }
}

注意事项：

  • 被禁用分支的 Future 可能每轮重新创建；
  • 某个分支赢得竞争后，其他分支 Future 会被 drop；
  • 读取流或协议帧时要确认中途取消不会丢失数据；
  • 不要在某分支中执行长时间无 .await 计算；
  • 对 shutdown 分支给予足够调度机会。

十四、不要在 async 中滥用文件 I/O

普通文件不像 socket 那样总能通过 readiness API 高效异步化。Tokio 的 tokio::fs 在许多平台上会把文件操作放到 blocking pool。

因此：

  • tokio::fs 可以避免阻塞 async worker；
  • 它不保证磁盘 I/O 变成内核原生异步；
  • 大量小文件操作仍可能压满 blocking pool；
  • 高吞吐存储场景应考虑批处理、专用线程池或 io_uring 方案；
  • 不要把 tokio::fs 当作无限并发许可证。

十五、错误处理与任务管理

不要随手丢弃 JoinHandle：

let _ = tokio::spawn(do_work());

这种 detached task 的错误和 panic 很容易丢失。更好的选择：

  • 请求范围内的并发：join! / try_join!；
  • 动态任务集合：JoinSet；
  • 长期后台任务：保存 handle，并接入 shutdown；
  • 真正 fire-and-forget：至少在任务内部记录错误和关键上下文。

任务 panic 不会自动让整个进程崩溃，但会表现为 JoinError。关键后台任务意外退出时，服务通常应该触发整体 shutdown，而不是静默降级。

十六、可观测性

异步系统仅看线程堆栈通常很难定位问题。建议使用结构化日志和 tracing：

use tracing::{info, instrument};

#[instrument(skip(client), fields(request_id = %request_id))]
async fn fetch(
    client: &Client,
    request_id: &str,
) -> anyhow::Result<Response> {
    info!("sending request");
    client.send().await
}

需要关注：

  • 请求总延迟和各 .await 阶段延迟；
  • channel 深度和 Semaphore 等待时间；
  • 活跃 task 数；
  • blocking pool 饱和；
  • 超时、取消和重试次数；
  • runtime worker 是否长期满负荷；
  • 下游连接池等待时间。

Tokio Console 可以帮助观察 Task 生命周期、poll 时间、wakeup 和资源等待，特别适合排查 task 长时间不让出、锁竞争和异步死锁。

十七、Runtime 配置原则

不要一开始就随意调整 worker 数量。默认 multi-thread runtime 通常是合理起点。

需要调整时，先明确瓶颈：

  • 网络 I/O 密集：worker 不必与连接数相等；
  • CPU 密集：应隔离计算，而不是无限增加 async worker；
  • 大量阻塞调用：先修复阻塞边界；
  • tail latency 异常：检查长 poll、锁竞争和队列积压；
  • NUMA 或超大机器：可考虑拆分 runtime 或绑定专用资源，但需要基准测试。

库代码通常不应该偷偷创建全局 runtime，也不应该在已有 runtime 内再次 block_on。库优先暴露 async API，让应用决定 runtime 配置和生命周期。

十八、最佳实践清单

并发

  • 独立 Future 使用 join!，独立任务使用 spawn；
  • 使用 Semaphore、有界 channel 或流并发限制建立背压；
  • 保存和管理重要任务的 JoinHandle；
  • 避免无上限 spawn。

阻塞边界

  • async task 内不调用 thread::sleep；
  • 阻塞库放入 spawn_blocking；
  • 长时间 CPU 工作使用专用计算池；
  • 不持有锁执行慢 I/O。

超时与取消

  • 外部操作设置 deadline；
  • 重试带指数退避、随机抖动和次数上限；
  • 验证 select! 中操作的取消安全性；
  • 设计幂等键或补偿机制。

共享状态

  • 短临界区可用标准 Mutex；
  • 避免持有 guard 跨 .await；
  • 优先消息传递和状态所有权集中；
  • 高并发 map 考虑分片而非单一全局锁。

生命周期

  • 使用 CancellationToken 或 channel 广播退出；
  • shutdown 时停止接收、等待 drain、设置超时；
  • 不创建无法停止、无法观察的后台任务；
  • 将 runtime 生命周期放在应用边界管理。

十九、总结

Tokio 的核心不是“把函数前加上 async”，而是一套协作式执行系统：

Future 在 poll 中推进
遇到 I/O 或 timer 返回 Pending
I/O/Timer Driver 在资源就绪时调用 Waker
Scheduler 将 Task 放回 ready queue
多个 worker 通过本地队列和 work stealing 执行任务

高质量 Tokio 程序通常具备这些特征：

  • Task 经常在真正的等待点让出线程；
  • 阻塞和 CPU 密集工作被明确隔离；
  • 并发有上限，并对下游形成背压；
  • 每个外部操作有超时和取消策略；
  • 共享状态的锁范围很短；
  • 后台任务可观察、可停止、可等待；
  • tracing 和指标能够解释请求时间花在哪里。

理解 runtime 的执行模型后，很多“异步性能问题”会变得容易判断：程序究竟是在等待 I/O、等待锁、等待配额，还是根本阻塞了 worker thread。

参考资料

  • Tokio 官方文档与教程
  • Tokio Runtime、Task、I/O、Time 和 Sync 模块文档
  • Tokio 官方源码中的 runtime scheduler 与 I/O driver
  • Rust 标准库 Future、Context、Waker 文档