rust-async-runtime

name	rust-async-runtime
description	Rust 非同步執行時技能。涵蓋 tokio 1.44 runtime 架構與選擇、async/await 語法模式、 Future trait 原理、常見陷阱（blocking in async、async trait）、 channel 通訊（mpsc/oneshot/broadcast/watch）、task spawn 策略、 select! 多路複用、graceful shutdown、async stream、Mutex 在 async 中的正確用法。 觸發關鍵詞：async, await, tokio, async-std, Future, spawn, channel, mpsc, select, runtime, blocking, async trait, stream, executor

Rust Async Runtime

適用場景

• 需要高併發 I/O（HTTP 伺服器、資料庫連線、WebSocket）
• 實作非同步任務排程、背景工作、定時器
• 使用 channel 進行任務間通訊
• 處理 async stream（串流資料處理、SSE、gRPC streaming）
• 需要 graceful shutdown 的長駐服務
• 在 async 環境中正確使用鎖與同步原語

核心知識

Runtime 選擇

Runtime	特性	適用場景
tokio	最成熟、生態系最大、work-stealing 排程器	生產環境首選，絕大多數專案
async-std	API 貼近 std、較輕量	學習用途、小型專案
smol	極簡、可組合、無 macro	嵌入式或需要最小依賴的場景

結論：除非有明確理由，一律選 tokio。生態系中 95%+ 的 async crate 以 tokio 為主要支援目標。

tokio Runtime 類型

// 多執行緒 runtime（預設），適合生產環境
#[tokio::main]
async fn main() { /* ... */ }

// 等同於：
fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async { /* ... */ })
}

// 單執行緒 runtime，適合測試或低負載場景
#[tokio::main(flavor = "current_thread")]
async fn main() { /* ... */ }

選擇依據：

• multi_thread：CPU 核心 > 1、需要平行處理、生產環境
• current_thread：單元測試、嵌入式環境、確定性排程需求

async/await 基本原理

1. async fn 回傳一個實作 Future trait 的匿名型別
2. Future 是惰性的——宣告不會執行，必須被 .await 或交給 executor
3. .await 讓出控制權給 runtime，直到 Future 完成
4. Runtime（executor）負責輪詢（poll）所有 Future 直到 Poll::Ready

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

blocking in async 問題

絕對不要在 async 上下文中執行同步阻塞操作：

// ❌ 錯誤：阻塞整個 executor 執行緒
async fn bad() {
    std::thread::sleep(Duration::from_secs(5)); // 阻塞！
    std::fs::read_to_string("large.txt").unwrap(); // 阻塞！
}

// ✅ 正確：使用 tokio 非同步版本
async fn good() {
    tokio::time::sleep(Duration::from_secs(5)).await;
    tokio::fs::read_to_string("large.txt").await.unwrap();
}

// ✅ 正確：無法避免同步操作時，用 spawn_blocking
async fn also_good() {
    let result = tokio::task::spawn_blocking(|| {
        heavy_cpu_computation()
    }).await.unwrap();
}

Channel 通訊模式

Channel 類型	模式	用途
mpsc	多生產者 → 單消費者	任務佇列、事件匯集
oneshot	單次發送 → 單次接收	請求-回應、任務結果回傳
broadcast	單生產者 → 多消費者	事件廣播、配置更新通知
watch	單生產者 → 多消費者（僅最新值）	狀態監控、設定熱更新

Task Spawn 策略

// 1. tokio::spawn — 在 runtime 中啟動新任務（需要 'static + Send）
let handle = tokio::spawn(async move {
    do_work().await
});
let result = handle.await?;

// 2. tokio::spawn_blocking — 在專用執行緒池中執行同步阻塞程式碼
let result = tokio::task::spawn_blocking(move || {
    compute_heavy_stuff()
}).await?;

// 3. tokio::task::spawn_local — 不要求 Send，但僅限 LocalSet
let local = tokio::task::LocalSet::new();
local.run_until(async {
    tokio::task::spawn_local(async {
        // 可以使用非 Send 型別
    }).await.unwrap();
}).await;

select! 多路複用

use tokio::select;

// 同時等待多個 Future，第一個完成的分支被執行，其餘被取消
select! {
    val = async_operation_a() => {
        println!("A 先完成: {val}");
    }
    val = async_operation_b() => {
        println!("B 先完成: {val}");
    }
    // 可選的 else 分支：所有分支都被 disable 時執行
    else => {
        println!("所有分支都不可用");
    }
}

注意：被取消的分支對應的 Future 會被 drop，確保你的 Future 是取消安全的（cancel-safe）。 mpsc::Receiver::recv() 是取消安全的，但 read_exact() 不是。

async Mutex 正確用法

// ❌ 持有 std::sync::Mutex 跨 .await 點會導致死鎖
let guard = std_mutex.lock().unwrap();
some_async_op().await; // guard 仍被持有！
drop(guard);

// ✅ 方案一：縮小臨界區，不跨 .await
let data = {
    let guard = std_mutex.lock().unwrap();
    guard.clone()
}; // guard 在此已 drop
some_async_op().await;

// ✅ 方案二：需要跨 .await 持有鎖時，使用 tokio::sync::Mutex
let guard = tokio_mutex.lock().await;
some_async_op().await;
drop(guard);

經驗法則：不跨 .await 用 std::sync::Mutex（更快）；跨 .await 用 tokio::sync::Mutex。

程式碼範例

Basic: async/await + tokio::spawn + JoinSet

use std::time::Duration;
use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    // 啟動單一任務
    let handle = tokio::spawn(async {
        tokio::time::sleep(Duration::from_millis(100)).await;
        42
    });
    let result = handle.await.expect("任務 panic");
    println!("單一任務結果: {result}");

    // 使用 JoinSet 管理多個併發任務
    let mut set = JoinSet::new();
    for i in 0..5 {
        set.spawn(async move {
            tokio::time::sleep(Duration::from_millis(50 * i)).await;
            i * 10
        });
    }
    while let Some(res) = set.join_next().await {
        println!("任務完成: {}", res.expect("任務 panic"));
    }
}

Intermediate: channel 通訊 + select!

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

/// 工作請求：附帶 oneshot 回傳通道
struct Request {
    payload: String,
    respond_to: oneshot::Sender<String>,
}

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

    // 工作者：接收請求並回傳結果
    let worker = tokio::spawn(async move {
        while let Some(req) = rx.recv().await {
            let response = format!("已處理: {}", req.payload);
            let _ = req.respond_to.send(response);
        }
    });

    // 發送三個請求
    for i in 0..3 {
        let (resp_tx, resp_rx) = oneshot::channel();
        tx.send(Request {
            payload: format!("任務-{i}"),
            respond_to: resp_tx,
        }).await.expect("channel 已關閉");

        // 使用 select! 加上超時
        select! {
            result = resp_rx => {
                println!("{}", result.expect("工作者已斷線"));
            }
            _ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {
                println!("請求 {i} 超時");
            }
        }
    }
    drop(tx); // 關閉 channel，讓工作者結束
    worker.await.expect("工作者 panic");
}

Advanced: graceful shutdown + async stream + semaphore

use std::time::Duration;
use tokio::sync::{mpsc, Semaphore};
use tokio::signal;
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::StreamExt;
use std::sync::Arc;

#[tokio::main]
async fn main() {
    let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1);
    let semaphore = Arc::new(Semaphore::new(3)); // 最多 3 個併發

    // 資料來源 stream
    let (data_tx, data_rx) = mpsc::channel(64);
    let producer = tokio::spawn(async move {
        for i in 0u64.. {
            if data_tx.send(i).await.is_err() { break; }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
    });

    // 消費者：帶限流 + stream 處理
    let sem = semaphore.clone();
    let consumer = tokio::spawn(async move {
        let mut stream = ReceiverStream::new(data_rx)
            .map(|n| n * 2)
            .take(20); // 最多處理 20 筆
        while let Some(value) = stream.next().await {
            let permit = sem.clone().acquire_owned().await.expect("semaphore 已關閉");
            tokio::spawn(async move {
                println!("處理: {value}");
                tokio::time::sleep(Duration::from_millis(200)).await;
                drop(permit); // 釋放許可
            });
        }
    });

    // 等待 Ctrl+C 或消費者完成
    tokio::select! {
        _ = signal::ctrl_c() => {
            println!("\n收到 Ctrl+C，開始優雅關閉...");
        }
        _ = consumer => {
            println!("消費者處理完畢");
        }
    }
    drop(shutdown_tx);
    producer.abort();

    // 等待所有 semaphore 許可歸還（確認所有進行中任務完成）
    let _ = semaphore.acquire_many(3).await;
    println!("所有任務已完成，程式結束");
}

常見錯誤對照表

錯誤訊息	原因	修復方式
future cannot be sent between threads safely / X is not Send	tokio::spawn 要求 Future 為 Send，但持有了非 Send 型別（如 Rc、MutexGuard）	改用 Arc 取代 Rc；縮小 MutexGuard 作用域使其不跨 .await；或使用 spawn_local
Cannot start a runtime from within a runtime	在 async 區塊內呼叫 block_on() 導致巢狀 runtime	移除內層 block_on()，改用 .await；若整合同步庫，用 spawn_blocking + 獨立 runtime
blocking annotated I/O must be called from the context of the Tokio runtime	在 runtime 外部呼叫 tokio I/O 函式	確保程式碼在 #[tokio::main] 或 Runtime::block_on 範圍內執行
the trait bound ... async_trait is not satisfied / async fn in trait 問題	Rust edition < 2024 不支援原生 async fn in trait，或物件安全限制	Edition 2024 原生支援 async fn in trait；需要 dyn Trait 時使用 #[trait_variant::make(SendTrait: Send)] 或 async-trait crate
死鎖：程式卡住不動	持有 std::sync::Mutex 鎖跨 .await 點	縮小臨界區避免跨 .await，或改用 tokio::sync::Mutex
capacity of channel is full / 發送端卡住	mpsc channel 已滿，背壓未處理	增大 buffer、加速消費端、用 try_send 非阻塞處理、或使用 unbounded_channel（小心 OOM）
JoinError: task panicked	tokio::spawn 的任務發生 panic	在 spawn 的 async 區塊內加 catch_unwind，或在 .await 時用 match 處理 JoinError

Cargo.toml 依賴模板

[package]
name = "my-async-app"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1.44", features = ["full"] }
tokio-stream = "0.1"
tokio-util = "0.7"            # CancellationToken、codec 等工具
futures = "0.3"                # FutureExt、StreamExt 擴充 trait
async-trait = "0.1"            # edition < 2024 的 async fn in trait 支援
pin-project-lite = "0.2"       # 安全的 pin projection

# 可選
reqwest = { version = "0.12", features = ["json"] }  # HTTP 客戶端
tower = "0.5"                  # Service 抽象與 middleware
tracing = "0.1"                # 結構化日誌
tracing-subscriber = "0.3"

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

注意：生產環境可只啟用所需 feature（如 rt-multi-thread, macros, net, time）以減少編譯時間。

參考來源

1. Tokio Tutorial — 官方教學，涵蓋 spawning、channels、I/O
2. Asynchronous Programming in Rust — Rust 官方 async book
3. Tokio API Documentation — tokio 1.x 完整 API 參考
4. Alice Ryhl — Actors with Tokio — channel-based actor 模式實戰
5. Jon Gjengset — Decrusting the tokio crate — 深入 tokio 內部架構
6. Rust Edition 2024 — async fn in trait — 原生 async trait 支援說明