Rust 2026 经验谈 - Tokio 2026 实战

1
// 1. tokio::spawn：在 worker 线程上调度 async 任务
2
let handle: JoinHandle<Result<Data>> = tokio::spawn(async move {
3
    fetch_data(url).await
4
});
5

6
// 2. tokio::task::spawn_blocking：在独立线程池上调度同步（阻塞）代码
7
let handle: JoinHandle<Result<Data>> = tokio::task::spawn_blocking(move || {
8
    blocking_c_library_call()  // 不可以 .await
9
});
10

11
// 3. tokio::task::spawn_local：在当前线程上调度（!Send 任务）
12
let handle: JoinHandle<()> = tokio::task::spawn_local(async move {
13
    // 这个 Future 不需要 Send，只在本线程运行
14
    let not_send = Rc::new(42);
15
    println!("{}", not_send);
16
});

1
// spawn_local 的典型场景：GUI 应用
2
use tokio::task::LocalSet;
3

4
#[tokio::main]
5
async fn main() {
6
    let local = LocalSet::new();
7

8
    local.spawn_local(async move {
9
        let not_send = Rc::new(0);
10
        // 这个 Rc 不会跨线程，安全
11
        loop {
12
            do_gui_work(&not_send).await;
13
        }
14
    });
15

16
    local.await;  // 运行 LocalSet 中的所有任务
17
}

1
// Tokio 内部的 budget 逻辑（简化版）
2
fn poll_accept(listener: &TcpListener, cx: &mut Context) -> Poll<io::Result<TcpStream>> {
3
    if budget_remaining() == 0 {
4
        // 预算用完，强制让出
5
        cx.waker().wake_by_ref();  // 安排下次 poll
6
        return Poll::Pending;
7
    }
8

9
    match listener.accept() {
10
        Ok(stream) => {
11
            consume_budget();  // 消耗一个预算
12
            Poll::Ready(Ok(stream))
13
        }
14
        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
15
            register_waker(cx);
16
            Poll::Pending
17
        }
18
        Err(e) => Poll::Ready(Err(e)),
19
    }
20
}

1
use tokio::sync::mpsc;
2

3
async fn event_loop(mut rx: mpsc::Receiver<Event>, shutdown: tokio::sync::watch::Receiver<bool>) {
4
    loop {
5
        tokio::select! {
6
            Some(event) = rx.recv() => {
7
                handle_event(event).await;
8
            }
9
            _ = shutdown.changed() => {
10
                tracing::info!("shutdown signal received");
11
                break;
12
            }
13
            else => {
14
                // 所有分支都返回 Pending/关闭
15
                tracing::info!("all channels closed");
16
                break;
17
            }
18
        }
19
    }
20
}

1
// 反模式：cancel 不安全的操作
2
async fn bad_select() {
3
    let buf = b"hello world";
4

5
    tokio::select! {
6
        // 如果 timeout 先就绪，write 可能已经写了部分数据
7
        // 但不知道写了多少！下次 write 从哪里开始？
8
        result = socket.write(buf) => {
9
            process(result);
10
        }
11
        _ = tokio::time::sleep(Duration::from_secs(5)) => {
12
            // timeout：write 被取消，可能已写部分数据
13
        }
14
    }
15
}

1
// 用 oneshot 通信，而非直接 select
2
let (tx, rx) = tokio::sync::oneshot::channel();
3

4
tokio::spawn(async move {
5
    let data = socket.read(&mut buf).await;
6
    let _ = tx.send(data);
7
});
8

9
tokio::select! {
10
    result = rx => {
11
        // 读取完成
12
    }
13
    _ = tokio::time::sleep(timeout) => {
14
        // 超时，读取任务继续运行（不被取消）
15
        // 需要清理：abort 读取任务
16
    }
17
}

1
// 场景 1：服务启动——所有组件都要成功
2
async fn start_server() -> Result<()> {
3
    let (db, cache, mq) = tokio::try_join!(
4
        connect_db(),
5
        connect_cache(),
6
        connect_mq(),
7
    )?;
8
    // 任何一个连接失败都不启动
9
    Ok(())
10
}
11

12
// 场景 2：服务关闭——所有组件都要关闭，但失败只记录日志
13
async fn shutdown(db: Db, cache: Cache, mq: Mq) {
14
    let (db_result, cache_result, mq_result) = tokio::join!(
15
        db.close(),
16
        cache.close(),
17
        mq.close(),
18
    );
19
    if let Err(e) = db_result {
20
        tracing::error!("db close failed: {}", e);
21
    }
22
    // 即使 db 关闭失败，cache 和 mq 也要尝试关闭
23
}
24

25
// 场景 3：竞速——任一完成即可
26
async fn fetch_with_fallback(primary: &str, fallback: &str) -> Data {
27
    tokio::select! {
28
        data = fetch(primary) => data,
29
        data = fetch(fallback) => data,
30
    }
31
}

1
fn main() {
2
    // 先创建运行时
3
    let rt = tokio::runtime::Runtime::new().unwrap();
4

5
    // 同步初始化阶段
6
    let config = load_config();
7

8
    // 在同步代码中 spawn async 任务
9
    let handle = rt.spawn(async {
10
        async_initialization().await
11
    });
12

13
    // 阻塞等待
14
    let result = rt.block_on(handle).unwrap();
15

16
    // 进入 async 主循环
17
    rt.block_on(async {
18
        serve(result).await;
19
    });
20
}

1
fn main() {
2
    // IO 运行时
3
    let io_rt = tokio::runtime::Builder::new_multi_thread()
4
        .worker_threads(4)
5
        .enable_io()
6
        .build()
7
        .unwrap();
8

9
    // 计算运行时
10
    let compute_rt = tokio::runtime::Builder::new_multi_thread()
11
        .worker_threads(8)
12
        .build()
13
        .unwrap();
14

15
    // 在 IO 运行时中运行主循环
16
    io_rt.block_on(async {
17
        // IO 任务在本运行时
18
        let data = fetch_data().await;
19

20
        // 计算任务在计算运行时
21
        let result = compute_rt.spawn(async move {
22
            heavy_computation(data).await
23
        }).await.unwrap();
24

25
        write_result(result).await;
26
    });
27
}

1
use tokio::signal;
2
use tokio::sync::watch;
3

4
async fn serve_with_graceful_shutdown() {
5
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
6

7
    // 监听 SIGTERM / Ctrl+C
8
    tokio::spawn(async move {
9
        signal::ctrl_c().await.ok();
10
        tracing::info!("shutdown signal received");
11
        let _ = shutdown_tx.send(true);
12
    });
13

14
    // 启动服务器，传入 shutdown 信号
15
    let server = hyper::Server::bind(&addr)
16
        .serve(make_service)
17
        .with_graceful_shutdown(async {
18
            shutdown_rx.changed().await.ok();
19
        });
20

21
    if let Err(e) = server.await {
22
        tracing::error!("server error: {}", e);
23
    }
24

25
    // 服务器停止后，等待 spawned 任务完成
26
    tracing::info!("server stopped, waiting for tasks to finish...");
27
    tokio::time::sleep(Duration::from_secs(5)).await;
28
}