Skip to content
Published on

异步 Rust：async/await 与 Tokio 运行时

🇰🇷KO🇺🇸EN🇯🇵JA🇨🇳ZH
필사 모드
分享
Authors

引言 — Rust 的 async 为何不同

在许多语言里,异步是内置在运行时里的。JavaScript 从一开始就有事件循环在转,Go 也默认自带一个调度 goroutine 的运行时。但 Rust 不一样。Rust 在语言层面提供了 async 语法,但真正驱动它运转的运行时并没有放进标准库。 运行时是一个需要另外挑选的独立 crate(通常是 Tokio)。

这个选择乍看陌生,但是有原因的。Rust 从嵌入式设备到大规模服务器都在用,不存在一个适合所有场景的单一运行时。所以语言只提供最小的骨架(Future trait 和 async/await 语法),把调度策略交给使用者自己挑选。

本文就从这副骨架讲起。Future 到底是什么、.await 实际做了什么、Tokio 在这之上如何驱动任务,以及实务中一定会撞上的陷阱(Send 约束、阻塞代码),都会用例子逐一说明。

Future 是一个懒惰的状态机

在 Rust 里调用一个 async fn,看起来函数体应该立刻执行,但什么都不会发生。 async fn 在调用的那一刻并不会执行函数体,而只是构造并返回一个叫 Future 的值。这个 Future 是一个表示"尚未完成的计算"的值。

async fn say_hello() {
    println!("hello");
}

fn main() {
    let fut = say_hello(); // 这里并不会打印"hello"!
    // fut 只是一个 Future 值,还没有开始执行
}

这就是 Rust async 的核心性质,懒惰性(laziness)。Future 在有人轮询(poll)它之前不会前进一步。轮询就是在问"你现在能推进吗?"。Future 对这个问题有两种回答:计算完成了就返回 Poll::Ready(值),还有东西要等就返回 Poll::Pending。

Future trait 的骨架大致长这样:

use std::pin::Pin;
use std::task::{Context, Poll};

trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

编译器会把 async 块转换成一个状态机(state machine)。每个 .await 点都变成一个状态,每次被轮询时,都会从上次暂停的地方恢复,推进到下一个 .await。正是这个状态机,让函数的中间状态得以在不使用栈的情况下保存下来,这也是为什么 Rust 的 async 轻量到无需堆分配也能运行。

.await — 中途让出控制权的点

.await 是等待另一个 Future 完成的点。但"等待"并不意味着占住一个线程停在原地。.await 真正的含义是:"如果这个 Future 还是 Pending,就把控制权让给执行器,让它去做别的事。"

async fn fetch_and_process() {
    let data = fetch_data().await;     // 如果这里还在等待,就让给其他任务
    let result = process(data).await;  // 完成之后继续往下走
    println!("{result}");
}

在 fetch_data().await 这一步,如果数据还没准备好,这个函数就会在原地暂停并交出控制权。这段时间里,执行器会去运行别的任务。等数据准备好之后,执行器会再次轮询这个 Future,函数会从暂停的那个 .await 点恢复执行。正是这种协作式的让出,才使得一个线程能够处理海量的并发工作。

这里有一条重要的规则:.await 只能在 async 上下文(async fn 或 async 块)里使用。普通的同步函数无法 .await 一个 Future。那么最初的那个 Future 由谁来发起轮询呢?是运行时。

执行器与运行时 — Tokio

因为 Future 是懒惰的,只有有人发起轮询并持续驱动它,它才会真正执行。承担这个角色的是执行器(executor),而在执行器之上再加上定时器、网络 I/O 之类的配套功能,组成的就是运行时(runtime)。在 Rust 生态里,事实上的标准是 Tokio。

最简单的起点是 #[tokio::main] 宏。这个宏会包裹 main 函数,建立一个运行时,并在其上驱动你的最顶层 Future,直到它完成。

#[tokio::main]
async fn main() {
    println!("start");
    say_hello().await; // 这里才真正执行
    println!("end");
}

async fn say_hello() {
    println!("hello");
}

#[tokio::main] 实际上展开成大致如下的代码:同步的 main 创建一个运行时,再用 block_on 把 async 块驱动到完成。

fn main() {
    tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async {
            println!("start");
            say_hello().await;
            println!("end");
        });
}

Tokio 默认使用多线程调度器。它会按 CPU 核心数开出等量的工作线程,把任务分配到这些线程上执行。如果某个工作线程闲下来,就会去"偷"其他工作线程的任务,用这种工作窃取(work-stealing)方式把负载均摊开。如果需要更轻量的单线程运行时,也可以切换成 #[tokio::main(flavor = "current_thread")]。

任务与 spawn — 并发的单位

到目前为止,我们都是按顺序 .await 单个 Future。但异步真正的威力,要在同时驱动多个操作时才体现出来。这个单位就是任务(task),用 tokio::spawn 创建。

tokio::spawn 接收一个 Future,把它作为独立任务注册到运行时里。这个任务会立刻在后台开始执行,spawn 会马上把该任务的句柄(JoinHandle)返回给你。任务比操作系统线程轻量得多,即便开出数万个,负担也很小。

#[tokio::main]
async fn main() {
    let mut handles = vec![];

    for i in 0..5 {
        let handle = tokio::spawn(async move {
            // 每个任务独立地、并发地执行
            some_async_work(i).await
        });
        handles.push(handle);
    }

    // 等待所有任务结束
    for handle in handles {
        let result = handle.await.unwrap();
        println!("got: {result}");
    }
}

async fn some_async_work(i: u32) -> u32 {
    i * 2
}

被 spawn 出去的任务,即便父级不去 .await 它,也会自行推进,但要拿到结果就需要 .await 这个 JoinHandle。如果任务发生了 panic,.await 的结果会返回 Err,可以在这里检测到。

如果你好奇多个生产者、消费者以任务的形式互相通信是什么样子——比如消息队列或工作池——可以在本站的消息队列演练场里可视化地体验队列的行为。它有助于直观理解异步任务是如何通过 channel 相互传递工作的。

Send 与 'static — spawn 的约束

传给 tokio::spawn 的 Future 带着两条约束。看一眼签名就知道原因:传入的 Future 必须是 Send,而且必须是 'static 生命周期。

Send 约束先说。在多线程运行时里,任务可以在任意一个工作线程上执行,甚至可以在跨越 .await 的时候被挪到另一个线程上。要做到这一点,Future 持有的所有值都必须能在线程间安全移动,也就是必须是 Send。所以如果跨越 .await 还持有一个非 Send 的值,就会编译报错。典型的例子就是前面智能指针那篇文章里出现过的 Rc。

use std::rc::Rc;

async fn bad() {
    let data = Rc::new(5); // Rc 不是 Send
    some_async_work().await; // 错误: 跨越 .await 持有了 Rc
    println!("{}", data);
}

这种情况下的解决办法是用 Arc 代替 Rc。或者,也可以在 .await 之前就把非 Send 的值从作用域里丢掉,让它不跨越 .await 点。

'static 约束存在的原因是,任务的生命周期可能比父级作用域更长。spawn 出去的任务,可能在父级已经返回之后依然存活,所以借用父级作用域里的引用是危险的。因此 Future 必须拥有数据,而不是借用引用。实务中,这通常通过在闭包上加 move、并用 Arc 共享所需数据来解决。

异步 trait — trait 中的 async fn

有很长一段时间,Rust 里无法直接在 trait 里写 async fn,因为 async fn 返回的 Future 的大小在 trait 这一层是未知的。于是多年来一直由 async-trait 这个 crate 来填补这个空缺。它用宏把返回类型改成 Box<dyn Future>,以承担一次堆分配为代价,换来可以在 trait 里写异步方法。

use async_trait::async_trait;

#[async_trait]
trait Repository {
    async fn find(&self, id: u64) -> Option<String>;
}

struct MyRepo;

#[async_trait]
impl Repository for MyRepo {
    async fn find(&self, id: u64) -> Option<String> {
        // 实际场景中会异步查询数据库
        Some(format!("item {id}"))
    }
}

好消息是,在相对较新的 Rust 里,trait 内的 async fn 已经开始在语言层面得到支持。 现在很多情况下,不需要 async-trait 宏,也可以直接在 trait 里写 async fn。不过在把它当作 trait object(dyn)使用,或者要给返回的 Future 加上 Send 约束等一些细节上,仍然需要留意,所以根据框架的不同,async-trait 有时还会继续被使用。

select! — 多个 Future 中最先完成的那个

有时候你想同时等待多个 Future,并对最先完成的那一个做出反应。最典型的场景就是"等待一项工作,但如果超时先到就放弃"。这种情况下可以用 tokio::select! 宏。

select! 会同时轮询列出的多个 Future,一旦其中一个先完成,就执行那个分支,其余的则被取消。

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

#[tokio::main]
async fn main() {
    tokio::select! {
        result = do_work() => {
            println!("作业完成: {result}");
        }
        _ = sleep(Duration::from_secs(5)) => {
            println!("超时!放弃这项工作");
        }
    }
}

async fn do_work() -> u32 {
    sleep(Duration::from_secs(10)).await;
    42
}

上面的例子里,do_work 要花 10 秒,而超时时间是 5 秒,所以 5 秒后计时器分支先完成,那一支被执行,do_work 则被取消。这里体现出异步 Rust 的一个重要性质:一个不再被轮询的 Future 就会直接被取消。 正在进行中的 do_work 只是不再被轮询,自然而然地被丢弃。不需要另外发送取消信号。

在循环里使用 select!,就可以构建一个在单个循环里同时监视多个事件源(channel 接收、定时器、终止信号等)的事件循环。这是异步服务器核心里常见的模式。

阻塞 vs 异步 — 为什么不能混用

异步编程里最常见、也最致命的错误,就是在异步上下文里调用阻塞代码。这里的"阻塞"指的是长时间占用 CPU 的重计算,或者像 std::thread::sleep、同步文件 I/O、同步网络调用这样真正把线程停下来的操作。

为什么危险?异步任务是短暂借用一个工作线程来被轮询的,在 .await 处让出控制权,归还线程。但如果任务发起了阻塞调用,它就不会让出,而是把整个工作线程占为己有。分配到那个工作线程上的其他所有任务,在这段时间里完全无法推进,只能挨饿。最坏的情况下,几个阻塞任务就能占满所有工作线程,让整个运行时停摆。

// 反例: 在异步上下文里把整个线程阻塞住
async fn bad() {
    std::thread::sleep(std::time::Duration::from_secs(5)); // 占住工作线程 5 秒!
}

// 正例: 用异步定时器让出控制权
async fn good() {
    tokio::time::sleep(std::time::Duration::from_secs(5)).await; // 归还线程
}

总结成两条原则。第一,凡是睡眠或等待,一律使用异步版本——用 tokio::time::sleep 代替 std::thread::sleep,用异步 HTTP 客户端代替同步 HTTP。第二,长时间占用 CPU 的重计算,或者不得不同步的库,用 tokio::task::spawn_blocking 挪到专用的阻塞线程池里去。这样一来,重活就不会把 async 工作线程饿死。

async fn heavy() {
    let result = tokio::task::spawn_blocking(|| {
        // 长时间占用 CPU 的计算或同步 I/O 放在这里
        expensive_sync_computation()
    })
    .await
    .unwrap();
    println!("{result}");
}

fn expensive_sync_computation() -> u64 {
    (0..1_000_000).sum()
}

结语

Rust 的 async 起初显得陌生,但只要抓住几个核心要点,就能连成一幅一致的图景。

  • async fn 只是构造出一个懒惰的 Future,在被轮询之前什么都不做。
  • .await 是一个让出控制权的点,而不是占住一个线程。
  • 要真正驱动 Future 运转,需要一个运行时(Tokio),并用 tokio::spawn 开出并发任务。
  • 被 spawn 的 Future 必须是 Send + 'static,所以要记住跨线程移动和生命周期这两件事。
  • 用 select! 让多个 Future 竞争,停止轮询后 Future 会自然被取消。
  • 在异步上下文里要避免阻塞,把重的同步工作用 spawn_blocking 隔离出去。

把这些原则内化于心,Rust 的 async 就会成为一个能同时给你安全性和性能的强大工具。语言给出的只是骨架,但正因为这副骨架是清晰的,你才能在准确理解发生了什么的前提下写代码。

参考资料

Discuss on Twitter • View on GitHub