Async Rust: Tokio, Futures & Patterns
How async/await works in Rust, the Tokio runtime, spawning tasks, channels, timeouts and cancellation, and runtime tuning.
Futures are State Machines
A Rust async fn compiles to a state machine: each .await point is a suspension. Futures do nothing until polled — there is no runtime baked into the language, you pick one: Tokio (most common), async-std or smol.
# Cargo.toml: tokio = { version = "1", features = ["full"] }
#[tokio::main] // main 本身也是异步
async fn main() {
// 并发等待两个 future
let (a, b) = tokio::join!(fetch("x"), fetch("y"));
println!("{a} {b}");
}
async fn fetch(key: &str) -> String { /* ... */ }
// 阻塞代码必须 spawn_blocking（如同步 IO、CPU 密集）
let data = tokio::task::spawn_blocking(|| heavy_compute()).await?;Spawning Tasks
use tokio::task;
// spawn 的任务独立于当前 future
let handle = task::spawn(async {
// 'static 要求：闭包不能借用外部数据
// 需要共享数据就 move 或用 Arc
let mut n = 0;
loop {
n += 1;
tokio::time::sleep(Duration::from_secs(1)).await;
}
});
// 等待结果：JoinHandle 的 await
let v: i32 = handle.await?;
// JoinSet 管理 一组任务
let mut set = task::JoinSet::new();
for i in 0..10 {
set.spawn(async move { i * 2 });
}
while let Some(res) = set.join_next().await { /* ... */ }Tasks need 'static futures: borrow from the surrounding scope and the borrow checker rejects the spawn. Move owned data in (often wrapped in Arc) — this is the most common async beginner error.
Channels
use tokio::sync::mpsc;
// mpsc：多生产者单消费者
let (tx, mut rx) = mpsc::channel::<Order>(100);
// 生产者：send 是异步的，满了会等待
let t = tokio::spawn(async move {
tx.send(Order::new()).await.unwrap();
});
// 消费者
while let Some(o) = rx.recv().await {
process(o);
}
// 广播（一对多）与 watch（只保留最新值）
// tokio::sync::broadcast / tokio::sync::watch
// 同线程内同步通信用 oneshot（一次性）
let (tx, rx) = tokio::sync::oneshot::channel();Timeouts and Cancellation
use tokio::time::{timeout, Duration};
// 超时包装：Err(elapsed) 表示超时
match timeout(Duration::from_secs(3), fetch("x")).await {
Ok(res) => println!("{res}"),
Err(_) => println!("timeout"),
}
// select!：等待第一个完成，其余被丢弃
// （丢弃 = 未来的 poll 不再发生，是安全的取消点）
tokio::select! {
msg = rx.recv() => println!("{msg:?}"),
_ = tokio::time::sleep(Duration::from_secs(5)) => {
println!("no message in 5s");
}
}
// select! 循环：每个分支用 = 绑定再 break
loop {
tokio::select! {
Some(o) = rx.recv() => handle(o),
true = shutdown.cancelled() => break,
}
}Runtime Tuning
- Default multi-thread runtime: one worker thread per core, work-stealing queue.
- Flavor: current_thread for CLI or embedded; multi_thread for servers.
- Bump worker_threads / blocking_threads for DB-heavy or blocking-call-heavy services.
- Never block in async context — use spawn_blocking; keep it in mind when integrating sync libraries.
Future 是状态机
Rust 的 async fn 会编译成状态机：每个 .await 都是一个挂起点。Future 在被 poll 之前什么都不做——语言本身不带运行时，由你选择 Tokio（最常用）、async-std 或 smol。
# Cargo.toml: tokio = { version = "1", features = ["full"] }
#[tokio::main] // main 本身也是异步
async fn main() {
// 并发等待两个 future
let (a, b) = tokio::join!(fetch("x"), fetch("y"));
println!("{a} {b}");
}
async fn fetch(key: &str) -> String { /* ... */ }
// 阻塞代码必须 spawn_blocking（如同步 IO、CPU 密集）
let data = tokio::task::spawn_blocking(|| heavy_compute()).await?;任务派生
use tokio::task;
// spawn 的任务独立于当前 future
let handle = task::spawn(async {
// 'static 要求：闭包不能借用外部数据
// 需要共享数据就 move 或用 Arc
let mut n = 0;
loop {
n += 1;
tokio::time::sleep(Duration::from_secs(1)).await;
}
});
// 等待结果：JoinHandle 的 await
let v: i32 = handle.await?;
// JoinSet 管理 一组任务
let mut set = task::JoinSet::new();
for i in 0..10 {
set.spawn(async move { i * 2 });
}
while let Some(res) = set.join_next().await { /* ... */ }任务要求 'static 的 future：借用外层作用域会被借用检查器拒绝。把需要的数据 move 进去（常包一层 Arc）——这是异步入门最常见的错误。
通道
use tokio::sync::mpsc;
// mpsc：多生产者单消费者
let (tx, mut rx) = mpsc::channel::<Order>(100);
// 生产者：send 是异步的，满了会等待
let t = tokio::spawn(async move {
tx.send(Order::new()).await.unwrap();
});
// 消费者
while let Some(o) = rx.recv().await {
process(o);
}
// 广播（一对多）与 watch（只保留最新值）
// tokio::sync::broadcast / tokio::sync::watch
// 同线程内同步通信用 oneshot（一次性）
let (tx, rx) = tokio::sync::oneshot::channel();超时与取消
use tokio::time::{timeout, Duration};
// 超时包装：Err(elapsed) 表示超时
match timeout(Duration::from_secs(3), fetch("x")).await {
Ok(res) => println!("{res}"),
Err(_) => println!("timeout"),
}
// select!：等待第一个完成，其余被丢弃
// （丢弃 = 未来的 poll 不再发生，是安全的取消点）
tokio::select! {
msg = rx.recv() => println!("{msg:?}"),
_ = tokio::time::sleep(Duration::from_secs(5)) => {
println!("no message in 5s");
}
}
// select! 循环：每个分支用 = 绑定再 break
loop {
tokio::select! {
Some(o) = rx.recv() => handle(o),
true = shutdown.cancelled() => break,
}
}运行时调优
- 默认多线程运行时：每个核心一个 worker，任务队列按核窃取。
- current_thread 适合 CLI/嵌入式，服务器用 multi_thread。
- 数据库或阻塞调用偏多的服务可调大 worker_threads、blocking_threads。
- 异步上下文里绝不阻塞——用 spawn_blocking；集成同步库时尤其注意。