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

```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 基本原理

1. `async fn` 回傳一個實作 `Future` trait 的匿名型別
2. `Future` 是惰性的——宣告不會執行，必須被 `.await` 或交給 executor
3. `.await` 讓出控制權給 runtime，直到 Future 完成
4. 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`）以減少編譯時間。

## 參考來源

1. [Tokio Tutorial](https://tokio.rs/tokio/tutorial) — 官方教學，涵蓋 spawning、channels、I/O
2. [Asynchronous Programming in Rust](https://rust-lang.github.io/async-book/) — Rust 官方 async book
3. [Tokio API Documentation](https://docs.rs/tokio/latest/tokio/) — tokio 1.x 完整 API 參考
4. [Alice Ryhl — Actors with Tokio](https://ryhl.io/blog/actors-with-tokio/) — channel-based actor 模式實戰
5. [Jon Gjengset — Decrusting the tokio crate](https://www.youtube.com/watch?v=o2ob8zkeq2s) — 深入 tokio 內部架構
6. [Rust Edition 2024 — async fn in trait](https://doc.rust-lang.org/edition-guide/rust-2024/) — 原生 async trait 支援說明