Async Rust与Tokio：async/await、task与channel

1. 概览

Rust内置了async/await语法，但没有内置runtime来运行Future。Tokio是最流行的runtime。它提供了调度器、定时器、异步I/O和同步工具。

本文从第一个async函数开始，到常见操作：spawn task、使用join!并行执行、使用select!设置超时、在task间共享状态、通过channel发送数据和处理blocking代码。最后讨论如何编写测试和常见错误。

2. 安装

在Cargo.toml中添加Tokio：

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

[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }

full特性启用所有模块：多线程runtime、time、sync、net、macros等。这种方式在学习时很方便。在实际项目中，你应该只启用实际使用的特性以减少编译时间，例如["rt-multi-thread", "macros", "time", "sync"]。test-util特性允许在测试中控制时间。我们将在第8节使用它。

3. 第一个async函数

3.1. #[tokio::main]与.await

use std::time::Duration;
use tokio::time::sleep;

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100)).await; // 模拟访问数据库
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let order = fetch_order(42).await;
    println!("{order}");
}

输出：

Order #42

解释：

• async fn fetch_order不会立即返回String。它返回一个impl Future<Output = String>。
• tokio::time::sleep是sleep的async版本。在等待时，thread被返回给runtime来处理其他任务。
• #[tokio::main]是一个宏。它创建一个多线程runtime，然后使用block_on在该runtime中运行main。

3.2. Future是lazy的

这是与JavaScript中Promise的重大区别：Rust中的future什么都不做，直到被.await或轮询。

async fn load_customer(id: u32) -> String {
    println!("正在加载customer {id}");
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let fut = load_customer(7); // 还没有运行函数内的任何代码
    println!("在await之前");
    let customer = fut.await;
    println!("{customer}");
}

输出：

在await之前
正在加载customer 7
Customer 7

“正在加载customer 7”这一行只在调用.await后才出现。如果你忘记.await，函数将永远不会运行。编译器会警告unused implementer of Future that must be used，所以不要忽视这个警告。

4. Spawn task

.await顺序运行futures。要让它们并发运行，使用tokio::spawn创建task。Task是由Tokio管理的轻量级“绿色线程”。

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(100 * id as u64)).await;
    format!("Order #{id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let mut handles = Vec::new();

    for id in 1..=3 {
        handles.push(tokio::spawn(async move { fetch_order(id).await }));
    }

    for handle in handles {
        let order = handle.await.expect("task被panic了");
        println!("{order}");
    }
    println!("总耗时: ~{} ms", start.elapsed().as_millis());
}

输出：

Order #1
Order #2
Order #3
总耗时: ~300 ms

如果顺序执行，总时间会是100 + 200 + 300 = 600 ms。spawn时，三个task并发运行，总时间约为300 ms。

一些需要掌握的要点：

• tokio::spawn返回JoinHandle<T>。当.await handle时，你会得到Result<T, JoinError>。如果task发生panic或被取消，结果就是Err。
• 传入的future必须满足Send + 'static，因为task可能在其他线程上运行并且生存期超过当前作用域。这就是为什么需要async move。
• Task在spawn时立即开始运行，不需要等待.await handle。

注意： Drop一个JoinHandle不会取消task。Task仍然会以detached模式继续运行。要取消它，调用handle.abort()。

当task数量不固定时，可以使用tokio::task::JoinSet。它允许按task完成的顺序获取结果，使用join_set.join_next().await。

5. 在同一task中并发执行：join!、select!、timeout

5.1. join!和try_join!

不一定总是需要spawn。tokio::join!在同一task中同时轮询多个futures并等待它们全部完成：

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

async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

async fn fetch_customer(id: u32) -> String {
    sleep(Duration::from_millis(150)).await;
    format!("Customer {id}")
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let (order, customer) = tokio::join!(fetch_order(1), fetch_customer(7));
    println!("{order} - {customer} ({} ms)", start.elapsed().as_millis());
}

输出：

Order #1 - Customer 7 (200 ms)

总时间等于最长的future，约200 ms。futures不需要'static，所以它们可以借用本地变量。

当函数返回Result时，使用tokio::try_join!。这个宏在第一个错误处停止并drop剩余的futures：

let (order, customer) = tokio::try_join!(fetch_order_checked(1), fetch_customer_checked(7))?;

这里fetch_order_checked和fetch_customer_checked是返回Result<String, AppError>的async函数。

5.2. timeout

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

match timeout(Duration::from_millis(50), fetch_order(1)).await {
    Ok(order) => println!("{order}"),
    Err(_) => println!("超过时间限制"),
}

输出：

超过时间限制

fetch_order需要200 ms，超过50 ms的限制。此时timeout返回Err(Elapsed)，内部的future被drop，即被取消。

5.3. select!

select!等待第一个分支完成，然后取消其他分支：

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

    tokio::spawn(async move {
        sleep(Duration::from_millis(100)).await;
        let _ = shutdown_tx.send(());
    });

    tokio::select! {
        order = fetch_order(1) => println!("处理完成 {order}"),
        _ = shutdown_rx => println!("收到shutdown信号，停止处理"),
    }
}

输出：

收到shutdown信号，停止处理

shutdown信号在100 ms后到达，早于fetch_order（200 ms），所以第二个分支获胜。

注意： 在Rust中取消意味着future在最近的.await点被drop。如果被取消的分支正在进行一个非原子操作，例如已经从socket读了一半的message，数据可能会丢失。在将函数放入select!循环之前，应该检查每个函数文档中的“cancel safety”部分。

6. 在task间共享数据

6.1. Arc<Mutex<T>>

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

#[tokio::main]
async fn main() {
    let inventory = Arc::new(Mutex::new(HashMap::from([("laptop", 10u32)])));
    let mut handles = Vec::new();

    for _ in 0..5 {
        let inventory = Arc::clone(&inventory);
        handles.push(tokio::spawn(async move {
            let mut stock = inventory.lock().unwrap();
            *stock.get_mut("laptop").unwrap() -= 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("剩余: {}", inventory.lock().unwrap()["laptop"]);
}

输出：

剩余: 5

这里使用std::sync::Mutex而不是tokio::sync::Mutex。当critical section短且**内部没有.await**时推荐此选择，因为标准mutex更快。

只有当必须在.await点保持lock时才需要tokio::sync::Mutex，例如在等待I/O时持有共享连接。

6.2. Channel mpsc

与其共享内存，不如在task间通过channel传递message：

use tokio::sync::mpsc;

#[derive(Debug)]
struct Order {
    id: u32,
    amount: u64,
}

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<Order>(32);

    for id in 1..=3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(Order { id, amount: id as u64 * 100 }).await.unwrap();
        });
    }
    drop(tx); // 关闭原始sender以便下面的循环能够结束

    let mut total = 0;
    while let Some(order) = rx.recv().await {
        println!("接收 {order:?}");
        total += order.amount;
    }
    println!("总计: {total}");
}

输出：

接收 Order { id: 1, amount: 100 }
接收 Order { id: 3, amount: 300 }
接收 Order { id: 2, amount: 200 }
总计: 600

接收的顺序可能在不同运行间变化。参数32是缓冲容量。当缓冲满时，send会等待，产生自然的back-pressure机制。当所有sender都被drop时，recv()返回None。如果忘记drop(tx)，程序会在while循环中永远挂起。

除了mpsc，Tokio还有：

• oneshot：发送单个值。
• broadcast：多个consumer，每个consumer接收所有message。
• watch：consumer只关心最新值。

7. 处理blocking代码

Tokio使用协作式调度。Task只在.await点让出线程。重型CPU操作或同步I/O会占用worker线程，导致其他task挨饿。

fn hash_password(password: &str) -> String {
    // 模拟CPU密集操作，例如bcrypt/argon2
    std::thread::sleep(std::time::Duration::from_millis(500));
    format!("hashed:{password}")
}

#[tokio::main]
async fn main() {
    let password = String::from("s3cret");
    let hashed = tokio::task::spawn_blocking(move || hash_password(&password))
        .await
        .expect("blocking task被panic了");
    println!("{hashed}");
}

输出：

hashed:s3cret

spawn_blocking将closure推送到专用于blocking的线程池。这样runtime的worker线程保持空闲以处理其他请求。

注意： 不要在async函数中直接调用std::thread::sleep、std::fs::read或同步HTTP客户端。使用async版本（tokio::time::sleep、tokio::fs）或在spawn_blocking中包装它们。对于大规模并行计算，Rayon通常更合适。

8. 测试async代码

宏#[tokio::test]为每个测试创建一个runtime。默认情况下这是单线程runtime。

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

pub async fn fetch_order(id: u32) -> String {
    sleep(Duration::from_millis(200)).await;
    format!("Order #{id}")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn fetch_order_returns_formatted_id() {
        assert_eq!(fetch_order(42).await, "Order #42");
    }

    #[tokio::test(start_paused = true)]
    async fn timeout_fires_when_order_service_is_slow() {
        let result = timeout(Duration::from_millis(50), fetch_order(1)).await;
        assert!(result.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn order_arrives_within_generous_timeout() {
        let result = timeout(Duration::from_secs(1), fetch_order(1)).await;
        assert_eq!(result.unwrap(), "Order #1");
    }
}

运行：

cargo test

输出：

running 3 tests
test tests::fetch_order_returns_formatted_id ... ok
test tests::timeout_fires_when_order_service_is_slow ... ok
test tests::order_arrives_within_generous_timeout ... ok

start_paused = true（需要test-util特性）冻结Tokio的时钟。当所有task都在等待定时器时，runtime自动跳到下一个检查点。因此超时测试几乎立即运行并给出稳定结果，不依赖CI机器速度。

如果需要检查多线程行为，使用#[tokio::test(flavor = "multi_thread", worker_threads = 2)]。

9. 常见错误

在.await点保持std::sync::MutexGuard。 Guard不是Send，因此future也不再是Send。此时tokio::spawn会报编译错误future cannot be sent between threads safely。修复方法是通过{ ... }块在.await之前缩小guard的作用域，或切换到tokio::sync::Mutex。

创建嵌套runtime。 在已在Tokio上运行的代码中调用Runtime::new().block_on(...)会引发panic，消息为Cannot start a runtime from within a runtime。在async上下文中，只需.await。

不必要时使用async。 读取几个文件的CLI工具或纯计算任务从Tokio中受益不多。你只需付出额外的编译时间和复杂性。当有许多并发I/O操作时async才有用：web服务器、爬虫、代理、调用多个服务的客户端。

无限spawn。 Task便宜但不免费。为百万元素列表中的每个项spawn一个task可能会耗尽下游资源，如数据库连接池。用tokio::sync::Semaphore限制并发级别或使用futures crate中的buffer_unordered。

忽视JoinError。 如果.await handle时使用let _ =，task中的panic会被吞掉。你应该记录错误或将其传播上去。

10. 结论

本文已覆盖async Rust与Tokio的核心部分：

• Future是lazy的，只在被.await时才运行。
• tokio::spawn创建并发运行的task。传入的future需要Send + 'static。
• join!和try_join!在同一task中运行多个futures。select!和timeout处理竞速和取消。
• 共享数据可通过Arc<Mutex<T>>共享，或通过mpsc和oneshot等channel传递。
• Blocking代码需要放在spawn_blocking中。
• #[tokio::test(start_paused = true)]让时间相关的测试快速且稳定。

掌握这些部分后，你可以继续学习通过tokio::net进行网络I/O，或使用Axum等框架在Tokio基础上构建web服务。