rust-async-runtime

使用方法

复制安装命令，在自己的智能体环境安装前先检查源码包。
从原始文档核对所需工具、凭据和支持的宿主。
说明输入材料与期望交付物，并对照来源检查输出。

npx skills add https://github.com/luckyegg168/rust-skill --skill rust-async-runtime

开始前需了解

安装不包含模型订阅、API 额度或访问文件的权限。
仓库星数和安装量反映活跃情况，不等于独立质量或安全测试。

仓库原始内容

按源仓库内容呈现，保留标题、案例、代码、表格、链接以及原文引用的演示图片。

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 類型

rust

// 多執行緒 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 基本原理

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

rust

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

blocking in async 問題

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

rust

// ❌ 錯誤：阻塞整個 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 策略

rust

// 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! 多路複用

rust

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 正確用法

rust

// ❌ 持有 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

rust

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!

rust

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

rust

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 依賴模板

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）以減少編譯時間。

參考來源

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

使用方法

开始前需了解

Rust Async Runtime

適用場景

核心知識

Runtime 選擇

tokio Runtime 類型

async/await 基本原理

blocking in async 問題

Channel 通訊模式

Task Spawn 策略

select! 多路複用

async Mutex 正確用法

程式碼範例

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

Intermediate: channel 通訊 + select!

Advanced: graceful shutdown + async stream + semaphore

常見錯誤對照表

Cargo.toml 依賴模板

參考來源

按任务相关推荐

更多 Skills

rust-crate-selection

rust-design-patterns

rust-ffi-interop

rust-async-runtime

使用方法

开始前需了解

Rust Async Runtime

適用場景

核心知識

Runtime 選擇

tokio Runtime 類型

async/await 基本原理

blocking in async 問題

Channel 通訊模式

Task Spawn 策略

select! 多路複用

async Mutex 正確用法

程式碼範例

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

Intermediate: channel 通訊 + select!

Advanced: graceful shutdown + async stream + semaphore

常見錯誤對照表

Cargo.toml 依賴模板

參考來源

按任务相关推荐

rust-patterns

更多 Skills

rust-crate-selection

rust-design-patterns

rust-ffi-interop