Async Rust：从 Future 到生产实践

版权与署名说明（非官方译本） 本文档为 microsoft/RustTraining 中 async-book 的中文翻译与改编版本。 译文主要由 AI 生成，未经过人工校对与修订。 原作版权归 Microsoft Corporation 所有；文档许可为 CC-BY-4.0。 本译本为社区非官方版本，不代表 Microsoft 官方立场。

讲师介绍

• Microsoft SCHIE（Silicon and Cloud Hardware Infrastructure Engineering，硅与云硬件基础设施工程）团队 Principal Firmware Architect
• 资深行业专家，专长于安全、系统编程（固件、操作系统、虚拟机监控程序）、CPU 与平台架构，以及 C++ 系统开发
• 2017 年起在 Rust 上编程（@AWS EC2），此后一直热爱这门语言

Rust 异步编程深度指南。与大多数从 tokio::main 入手、对内部机制一笔带过的 async 教程不同，本指南从第一性原理出发——Future Trait、轮询（polling）、状态机——逐步推进到真实世界的模式、运行时选型与生产陷阱。

适合谁读

• 能写同步 Rust，但觉得 async 令人困惑的 Rust 开发者
• 来自 C#、Go、Python 或 JavaScript、熟悉 async/await 但不了解 Rust 模型的开发者
• 曾被 Future is not Send、Pin<Box<dyn Future>>，或「程序为什么挂住了？」困扰的任何人

前置知识

你应熟悉：

• 所有权（Ownership）、借用（Borrowing）与生命周期（Lifetime）
• Trait 与泛型（包括 impl Trait）
• 使用 Result<T, E> 与 ? 运算符
• 基本多线程（std::thread::spawn、Arc、Mutex）

无需先前的 async Rust 经验。

如何使用本书

首次阅读请按顺序通读。 第一至三部分相互依赖。每章标注：

每章包含：

• 顶部的 「你将学到」 块
• 面向视觉学习者的 Mermaid 图
• 带隐藏解答的 内联练习
• 总结核心思想的 要点回顾
• 指向相关章节的 交叉引用

进度建议

总预估时间：22–30 小时

如何完成练习

每个内容章节都有内联练习。毕业项目（第 17 章）将所有内容整合为单一项目。为获得最大学习效果：

1. 展开解答前先尝试练习 — 挣扎才是学习发生的地方
2. 亲手敲代码，不要复制粘贴 — 肌肉记忆对 Rust 语法很重要
3. 运行每个示例 — 使用 cargo new async-exercises，边学边测

目录

第一部分：异步如何工作

• 1. Rust 中异步为何不同 🟢 — 根本差异：Rust 没有内置运行时
• 2. Future Trait 🟡 — poll()、Waker，以及使一切运转的契约
• 3. Poll 如何工作 🟡 — 轮询状态机与最小执行器
• 4. Pin 与 Unpin 🔴 — 自引用结构体为何需要固定（pinning）
• 5. 状态机揭秘 🟢 — 编译器从 async fn 实际生成了什么

第二部分：生态系统

• 6. 手动构建 Future 🟡 — 从零实现 TimerFuture、Join、Select
• 7. 执行器与运行时 🟡 — tokio、smol、async-std、embassy — 如何选择
• 8. Tokio 深入解析 🟡 — 运行时风格、spawn、channel、同步原语
• 9. 何时不该用 Tokio 🟡 — LocalSet、FuturesUnordered、与运行时无关的设计
• 10. 异步 Trait 🟡 — RPITIT、dyn 分发、trait_variant、异步闭包

第三部分：生产级异步

• 11. Stream 与 AsyncIterator 🟡 — 异步迭代、AsyncRead/Write、stream 组合子
• 12. 常见陷阱 🔴 — 9 个生产 bug 及如何避免
• 13. 生产模式 🔴 — 优雅关闭、背压、Tower 中间件
• 14. 异步是优化，而非架构 🔴 — 同步核心 / 异步外壳、函数着色代价

附录

• 总结与参考卡 — 快速查阅表与决策树
• 毕业项目：异步聊天服务器 — 构建完整 async 应用

1. 为何 Rust 中的异步与众不同 🟢

你将学到：

• 为何 Rust 没有内置异步运行时（以及这对你的意义）
• 三个关键特性：惰性执行、无运行时、零成本抽象（zero-cost abstraction）
• 何时该用异步（以及何时它反而更慢）
• Rust 的模型与 C#、Go、Python、JavaScript 的对比

根本差异

大多数带有 async/await 的语言都会把底层机制藏起来。C# 有 CLR 线程池。JavaScript 有事件循环。Go 在运行时中内置了 goroutine 和调度器。Python 有 asyncio。

Rust 什么都没有。

没有内置运行时、没有线程池、没有事件循环。async 关键字是一种零成本编译策略——它把你的函数变换成实现 Future Trait（特征）的状态机。必须由其他人（executor，执行器）来驱动该状态机向前推进。

Rust 异步的三个关键特性

graph LR
    subgraph "C# / JS / Go"
        EAGER["急切执行（Eager Execution）<br/>Task 立即开始执行"]
        BUILTIN["内置运行时（Built-in Runtime）<br/>包含线程池"]
        GC["GC 管理（GC-Managed）<br/>无需关心生命周期"]
    end

    subgraph "Rust（以及 Python*）"
        LAZY["惰性执行（Lazy Execution）<br/>在 poll/await 之前什么都不发生"]
        BYOB["自带运行时（Bring Your Own Runtime）<br/>由你选择执行器"]
        OWNED["适用所有权（Ownership Applies）<br/>生命周期、Send、Sync 都很重要"]
    end

    EAGER -. "相反" .-> LAZY
    BUILTIN -. "相反" .-> BYOB
    GC -. "相反" .-> OWNED

    style LAZY fill:#e8f5e8,color:#000
    style BYOB fill:#e8f5e8,color:#000
    style OWNED fill:#e8f5e8,color:#000
    style EAGER fill:#e3f2fd,color:#000
    style BUILTIN fill:#e3f2fd,color:#000
    style GC fill:#e3f2fd,color:#000

* Python 协程与 Rust future 一样是惰性的——在 await 或被调度之前不会执行。不过 Python 仍使用 GC，没有所有权/生命周期方面的顾虑。

无内置运行时

// This compiles but does NOTHING:
async fn fetch_data() -> String {
    "hello".to_string()
}

fn main() {
    let future = fetch_data(); // Creates the Future, but doesn't execute it
    // future is just a struct sitting on the stack
    // No output, no side effects, nothing happens
    drop(future); // Silently dropped — work was never started
}

与 C# 中 Task 会急切启动形成对比：

// C# — this immediately starts executing:
async Task<string> FetchData() => "hello";

var task = FetchData(); // Already running!
var result = await task; // Just waits for completion

惰性 Future 与急切 Task

这是最重要的思维转变：

// To actually RUN a future, you need an executor:
#[tokio::main]
async fn main() {
    let result = fetch_data().await; // NOW it executes
    println!("{result}");
}

何时使用异步（以及何时不用）

graph TD
    START["什么类型的工作？"]

    IO["I/O 密集型？<br/>（网络、文件、数据库）"]
    CPU["CPU 密集型？<br/>（计算、解析）"]
    MANY["大量并发连接？<br/>（100+）"]
    FEW["少量并发任务？<br/>（<10）"]

    USE_ASYNC["✅ 使用 async/await"]
    USE_THREADS["✅ 使用 std::thread 或 rayon"]
    USE_SPAWN_BLOCKING["✅ 使用 spawn_blocking()"]
    MAYBE_SYNC["考虑同步代码<br/>（更简单、开销更小）"]

    START -->|网络、文件、数据库| IO
    START -->|计算| CPU
    IO -->|是，很多| MANY
    IO -->|只有少量| FEW
    MANY --> USE_ASYNC
    FEW --> MAYBE_SYNC
    CPU -->|并行化| USE_THREADS
    CPU -->|在异步上下文中| USE_SPAWN_BLOCKING

    style USE_ASYNC fill:#c8e6c9,color:#000
    style USE_THREADS fill:#c8e6c9,color:#000
    style USE_SPAWN_BLOCKING fill:#c8e6c9,color:#000
    style MAYBE_SYNC fill:#fff3e0,color:#000

经验法则：异步用于 I/O 并发（在等待时同时处理多件事），而非 CPU 并行（让一件事更快）。若有 10,000 个网络连接，异步表现优异。若在做数值计算，请用 rayon 或 OS 线程。

异步何时会更慢

异步并非免费。对于低并发工作负载，同步代码可能优于异步：

基准测试建议：若并发 I/O 操作少于约 10 个，在决定采用异步之前先做性能分析。在现代 Linux 上，每个连接简单 std::thread::spawn 也能轻松扩展到数百个线程。

练习：你会在何时使用异步？

要点回顾 — 为何异步与众不同

• Rust future 是惰性的——在执行器 poll 之前什么都不做
• 没有内置运行时——由你选择（或构建）自己的运行时
• 异步是一种零成本编译策略，生成状态机
• 异步在 I/O 密集型并发 上表现突出；CPU 密集型工作请用线程或 rayon

另见： 第 2 章 — Future Trait 了解让这一切运转的 trait，第 7 章 — 执行器与运行时 了解如何选择运行时

2. Future Trait 🟡

你将学到：

• Future Trait（特征）：Output、poll()、Context、Waker
• waker 如何告诉执行器「请再次 poll 我」
• 契约：从不调用 wake() = 程序会静默挂起
• 手写实现一个真正的 future（Delay）

Future 的解剖结构

异步 Rust 中的一切最终都实现这个 trait：

#![allow(unused)]
fn main() {
pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),   // The future has completed with value T
    Pending,    // The future is not ready yet — call me back later
}
}

就这些。Future 是任何可以被 poll（轮询）的东西——被问「完成了吗？」——并回答「是的，这是结果」或「还没好，准备好时我会唤醒你」。

Output、poll()、Context、Waker

sequenceDiagram
    participant E as 执行器（Executor）
    participant F as Future（Task）
    participant OS as 操作系统<br/>（如 epoll/kqueue）
    participant R as Reactor（运行时）

    E->>F: 调用 poll(cx)
    Note right of F: Future 尝试执行操作
    F->>OS: 系统调用（如读取 TCP socket）
    OS-->>F: 返回错误：未就绪
    
    F->>R: 注册：(Waker)
    F-->>E: 返回 Poll::Pending
    Note left of E: Task 移出<br/>运行队列

    E->>E: （执行器运行其他 task 或休眠）
    R->>OS: epoll_wait() / 向 OS 轮询事件

    Note right of OS: （稍后）新数据到达
    OS-->>R: 唤醒 Reactor：数据现已就绪
    
    R->>R: Reactor 找到 Waker
    R->>E: 调用 Waker::wake()
    Note right of E: Task 被推回<br/>执行器的运行队列

    E->>F: 再次调用 poll(cx)
    Note right of F: Future 再次尝试操作
    F->>OS: 系统调用（如读取 TCP socket）
    OS-->>F: 成功：返回数据缓冲区
    F-->>E: 返回 Poll::Ready(Data)

逐项拆解：

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

// A future that returns 42 immediately
struct Ready42;

impl Future for Ready42 {
    type Output = i32; // What the future eventually produces

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<i32> {
        Poll::Ready(42) // Always ready — no waiting
    }
}
}

各组成部分：

• Output — future 完成时产生的值类型
• poll() — 由执行器调用以检查进度；返回 Ready(value) 或 Pending
• Pin<&mut Self> — 确保 future 在内存中不会被移动（原因见第 4 章）
• Context — 携带 Waker，以便 future 在可以推进时通知执行器

Waker 契约

Waker 是回调机制。当 future 返回 Pending 时，它必须安排稍后调用 waker.wake()——否则执行器永远不会再次 poll 它，程序就会挂起。

#![allow(unused)]
fn main() {
use std::task::{Context, Poll, Waker};
use std::pin::Pin;
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

/// A future that completes after a delay (toy implementation)
struct Delay {
    completed: Arc<Mutex<bool>>,
    waker_stored: Arc<Mutex<Option<Waker>>>,
    duration: Duration,
    started: bool,
}

impl Delay {
    fn new(duration: Duration) -> Self {
        Delay {
            completed: Arc::new(Mutex::new(false)),
            waker_stored: Arc::new(Mutex::new(None)),
            duration,
            started: false,
        }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        // Check if already completed before storing waker
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        // Store the waker - executor may pass a new one on each poll
        *self.waker_stored.lock().unwrap() = Some(cx.waker().clone());

        // Start the background timer on first poll
        if !self.started {
            self.started = true;
            let completed = Arc::clone(&self.completed);
            let waker = Arc::clone(&self.waker_stored);
            let duration = self.duration;

            thread::spawn(move || {
                thread::sleep(duration);
                *completed.lock().unwrap() = true;

                // CRITICAL: wake the executor so it polls us again
                if let Some(w) = waker.lock().unwrap().take() {
                    w.wake(); // "Hey executor, I'm ready — poll me again!"
                }
            });
        }

        // Double-check completion after storing waker (handles race condition)
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        Poll::Pending // Not done yet
    }
}
}

关键洞见：在 C# 中，TaskScheduler 会自动处理唤醒。 在 Rust 中，你（或你使用的 I/O 库）负责调用 waker.wake()。忘了这一步，程序就会静默挂起。

练习：实现 CountdownFuture

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct CountdownFuture {
    count: u32,
}

impl CountdownFuture {
    fn new(start: u32) -> Self {
        CountdownFuture { count: start }
    }
}

impl Future for CountdownFuture {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.count == 0 {
            println!("Liftoff!");
            Poll::Ready("Liftoff!")
        } else {
            println!("{}...", self.count);
            self.count -= 1;
            cx.waker().wake_by_ref(); // Schedule re-poll immediately
            Poll::Pending
        }
    }
}
}

要点回顾 — Future Trait

• Future::poll() 返回 Poll::Ready(value) 或 Poll::Pending
• future 在返回 Pending 之前必须注册 Waker——执行器用它来判断何时重新 poll
• Pin<&mut Self> 保证 future 在内存中不会被移动（自引用状态机需要——见第 4 章）
• 异步 Rust 中的一切——async fn、.await、组合子——都建立在这一个 trait 之上

另见： 第 3 章 — Poll 如何工作 了解执行器循环，第 6 章 — 手写 Future 了解更复杂的实现

3. Poll 如何工作 🟡

你将学到：

• 执行器的 poll 循环：poll → pending → wake → 再次 poll
• 如何从零构建最小执行器
• 虚假唤醒（spurious wake）规则及其重要性
• 实用函数：poll_fn() 和 yield_now()

轮询状态机

执行器运行一个循环：poll future，若为 Pending 则停放它直到 waker 触发，然后再次 poll。这与 OS 线程根本不同——后者由内核负责调度。

stateDiagram-v2
    [*] --> Idle : Future 已创建
    Idle --> Polling : 执行器调用 poll()
    Polling --> Complete : Ready(value)
    Polling --> Waiting : Pending
    Waiting --> Polling : 调用 waker.wake()
    Complete --> [*] : 返回值

重要： 在 Waiting 状态下，future 必须已向 I/O 源注册了 waker。未注册 = 永远挂起。

最小执行器

为揭开执行器的面纱，我们来构建最简单的一个：

use std::future::Future;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::pin::Pin;

/// The simplest possible executor: busy-loop poll until Ready
fn block_on<F: Future>(mut future: F) -> F::Output {
    // Pin the future on the stack
    // SAFETY: `future` is never moved after this point — we only
    // access it through the pinned reference until it completes.
    let mut future = unsafe { Pin::new_unchecked(&mut future) };

    // Create a no-op waker (just keeps polling — inefficient but simple)
    fn noop_raw_waker() -> RawWaker {
        fn no_op(_: *const ()) {}
        fn clone(_: *const ()) -> RawWaker { noop_raw_waker() }
        let vtable = &RawWakerVTable::new(clone, no_op, no_op, no_op);
        RawWaker::new(std::ptr::null(), vtable)
    }

    // SAFETY: noop_raw_waker() returns a valid RawWaker with a correct vtable.
    let waker = unsafe { Waker::from_raw(noop_raw_waker()) };
    let mut cx = Context::from_waker(&waker);

    // Busy-loop until the future completes
    loop {
        match future.as_mut().poll(&mut cx) {
            Poll::Ready(value) => return value,
            Poll::Pending => {
                // A real executor would park the thread here
                // and wait for waker.wake() — we just spin
                std::thread::yield_now();
            }
        }
    }
}

// Usage:
fn main() {
    let result = block_on(async {
        println!("Hello from our mini executor!");
        42
    });
    println!("Got: {result}");
}

不要在生产环境使用！ 它会忙循环，浪费 CPU。真正的执行器 （tokio、smol）使用 epoll/kqueue/io_uring 休眠直到 I/O 就绪。 但这展示了核心思想：执行器就是一个调用 poll() 的循环。

唤醒通知

真正的执行器是事件驱动的。当所有 future 都是 Pending 时，执行器休眠。waker 是一种中断机制：

#![allow(unused)]
fn main() {
// Conceptual model of a real executor's main loop:
fn executor_loop(tasks: &mut TaskQueue) {
    loop {
        // 1. Poll all tasks that have been woken
        while let Some(task) = tasks.get_woken_task() {
            match task.poll() {
                Poll::Ready(result) => task.complete(result),
                Poll::Pending => { /* task stays in queue, waiting for wake */ }
            }
        }

        // 2. Sleep until something wakes us up (epoll_wait, kevent, etc.)
        //    This is where mio/polling does the heavy lifting
        tasks.wait_for_events(); // blocks until an I/O event or waker fires
    }
}
}

虚假唤醒

future 可能在其 I/O 尚未就绪时就被 poll。这称为虚假唤醒（spurious wake）。future 必须正确处理这种情况：

#![allow(unused)]
fn main() {
impl Future for MyFuture {
    type Output = Data;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Data> {
        // ✅ CORRECT: Always re-check the actual condition
        if let Some(data) = self.try_read_data() {
            Poll::Ready(data)
        } else {
            // Re-register the waker (it might have changed!)
            self.register_waker(cx.waker());
            Poll::Pending
        }

        // ❌ WRONG: Assuming poll means data is ready
        // let data = self.read_data(); // might block or panic
        // Poll::Ready(data)
    }
}
}

实现 poll() 的规则：

1. 绝不阻塞 — 若未就绪，立即返回 Pending
2. 始终重新注册 waker — 两次 poll 之间它可能已改变
3. 处理虚假唤醒 — 检查实际条件，不要假设已就绪
4. 不要在 Ready 之后 poll — 行为是未定义的（可能 panic、返回 Pending 或重复 Ready）。只有 FusedFuture 保证完成后 poll 是安全的

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Poll, Waker};

struct FlagFuture {
    flag: Arc<AtomicBool>,
    waker_slot: Arc<Mutex<Option<Waker>>>,
}

impl FlagFuture {
    fn new(flag: Arc<AtomicBool>, waker_slot: Arc<Mutex<Option<Waker>>>) -> Self {
        FlagFuture { flag, waker_slot }
    }
}

impl Future for FlagFuture {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Always re-check the actual condition — never trust the wake alone
        if self.flag.load(Ordering::Acquire) {
            return Poll::Ready(());
        }

        // Store/update the waker so we get notified
        let mut slot = self.waker_slot.lock().unwrap();
        *slot = Some(cx.waker().clone());

        // Re-check after storing the waker to avoid a race:
        // the flag could have been set between our first check
        // and storing the waker
        if self.flag.load(Ordering::Acquire) {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }
}

// The setter side (e.g., another thread or task):
fn set_flag(flag: &AtomicBool, waker_slot: &Mutex<Option<Waker>>) {
    flag.store(true, Ordering::Release);
    if let Some(waker) = waker_slot.lock().unwrap().take() {
        waker.wake();
    }
}

// Equivalent using poll_fn:
// async fn wait_for_flag(flag: Arc<AtomicBool>, waker_slot: Arc<Mutex<Option<Waker>>>) {
//     std::future::poll_fn(|cx| {
//         if flag.load(Ordering::Acquire) {
//             return Poll::Ready(());
//         }
//         *waker_slot.lock().unwrap() = Some(cx.waker().clone());
//         if flag.load(Ordering::Acquire) { Poll::Ready(()) } else { Poll::Pending }
//     }).await
// }
}

实用工具：poll_fn 和 yield_now

标准库和 tokio 提供的两个工具，可避免编写完整的 Future 实现：

#![allow(unused)]
fn main() {
use std::future::poll_fn;
use std::task::Poll;

// poll_fn: create a one-off future from a closure
let value = poll_fn(|cx| {
    // Do something with cx.waker(), return Ready or Pending
    Poll::Ready(42)
}).await;

// Real-world use: bridge a callback-based API into async
async fn read_when_ready(source: &MySource) -> Data {
    poll_fn(|cx| source.poll_read(cx)).await
}
}

#![allow(unused)]
fn main() {
// yield_now: voluntarily yield control to the executor
// Useful in CPU-heavy async loops to avoid starving other tasks
async fn cpu_heavy_work(items: &[Item]) {
    for (i, item) in items.iter().enumerate() {
        process(item); // CPU work

        // Every 100 items, yield to let other tasks run
        if i % 100 == 0 {
            tokio::task::yield_now().await;
        }
    }
}
}

何时使用 yield_now()：若你的异步函数在循环中做 CPU 工作且没有任何 .await 点，它会独占执行器线程。定期插入 yield_now().await 以实现协作式多任务。

要点回顾 — Poll 如何工作

• 执行器反复对已唤醒的 future 调用 poll()
• future 必须处理虚假唤醒——始终重新检查实际条件
• poll_fn() 让你用闭包创建临时 future
• yield_now() 是 CPU 密集型异步代码的协作式调度逃生口

另见： 第 2 章 — Future Trait 了解 trait 定义，第 5 章 — 状态机揭秘 了解编译器生成了什么

4. Pin 与 Unpin 🔴

你将学到：

• 为何自引用结构体在内存中移动时会失效
• Pin<P> 保证什么以及它如何防止移动
• 三种实用固定模式：Box::pin()、tokio::pin!()、Pin::new()
• 何时 Unpin 提供逃生口

Pin 为何存在

这是异步 Rust 中最令人困惑的概念。我们一步步建立直觉。

问题：自引用结构体

当编译器把 async fn 变换为状态机时，该状态机可能包含对自身字段的引用。这就形成了自引用结构体（self-referential struct）——在内存中移动它会使其内部引用失效。

#![allow(unused)]
fn main() {
// What the compiler generates (simplified) for:
// async fn example() {
//     let data = vec![1, 2, 3];
//     let reference = &data;       // Points to data above
//     use_ref(reference).await;
// }

// Becomes something like:
enum ExampleStateMachine {
    State0 {
        data: Vec<i32>,
        // reference: &Vec<i32>,  // PROBLEM: points to `data` above
        //                        // If this struct moves, the pointer is dangling!
    },
    State1 {
        data: Vec<i32>,
        reference: *const Vec<i32>, // Internal pointer to data field
    },
    Complete,
}
}

graph LR
    subgraph "移动前（有效）"
        A["data: [1,2,3]<br/>地址 0x1000"]
        B["reference: 0x1000<br/>（指向 data）"]
        B -->|"有效"| A
    end

    subgraph "移动后（无效）"
        C["data: [1,2,3]<br/>地址 0x2000"]
        D["reference: 0x1000<br/>（仍指向旧位置！）"]
        D -->|"悬空！"| E["💥 0x1000<br/>（已释放/垃圾）"]
    end

    style E fill:#ffcdd2,color:#000
    style D fill:#ffcdd2,color:#000
    style B fill:#c8e6c9,color:#000

自引用结构体

这不是学术问题。每个在 .await 点之间持有引用的 async fn 都会创建自引用状态机：

#![allow(unused)]
fn main() {
async fn problematic() {
    let data = String::from("hello");
    let slice = &data[..]; // slice borrows data
    
    some_io().await; // <-- .await point: state machine stores both data AND slice
    
    println!("{slice}"); // uses the reference after await
}
// The generated state machine has `data: String` and `slice: &str`
// where slice points INTO data. Moving the state machine = dangling pointer.
}

Pin 实践

Pin<P> 是一个包装器，防止指针背后的值被移动：

#![allow(unused)]
fn main() {
use std::pin::Pin;

let mut data = String::from("hello");

// Pin it — now it can't be moved
let pinned: Pin<&mut String> = Pin::new(&mut data);

// Can still use it:
println!("{}", pinned.as_ref().get_ref()); // "hello"

// But we can't get &mut String back (which would allow mem::swap):
// let mutable: &mut String = Pin::into_inner(pinned); // Only if String: Unpin
// String IS Unpin, so this actually works for String.
// But for self-referential state machines (which are !Unpin), it's blocked.
}

在实际代码中，你主要在三个地方遇到 Pin：

#![allow(unused)]
fn main() {
// 1. poll() signature — all futures are polled through Pin
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Output>;

// 2. Box::pin() — heap-allocate and pin a future
let future: Pin<Box<dyn Future<Output = i32>>> = Box::pin(async { 42 });

// 3. tokio::pin!() — pin a future on the stack
tokio::pin!(my_future);
// Now my_future: Pin<&mut impl Future>
}

Unpin 逃生口

Rust 中大多数类型都是 Unpin——它们不包含自引用，因此固定是空操作。只有编译器生成的状态机（来自 async fn）是 !Unpin。

#![allow(unused)]
fn main() {
// These are all Unpin — pinning them does nothing special:
// i32, String, Vec<T>, HashMap<K,V>, Box<T>, &T, &mut T

// These are !Unpin — they MUST be pinned before polling:
// The state machines generated by `async fn` and `async {}`

// Practical implication:
// If you write a Future by hand and it has NO self-references,
// implement Unpin to make it easier to work with:
impl Unpin for MySimpleFuture {} // "I'm safe to move, trust me"
}

快速参考

#![allow(unused)]
fn main() {
// Snippet A
let fut = async { 42 };
let pinned = Box::pin(fut);
let moved = pinned; // Move the Box
let result = moved.await;

// Snippet B
let fut = async { 42 };
tokio::pin!(fut);
let moved = fut; // Move the pinned future
let result = moved.await;

// Snippet C
use std::pin::Pin;
let mut fut = async { 42 };
let pinned = Pin::new(&mut fut);
}

#![allow(unused)]
fn main() {
let fut = async { 42 };
tokio::pin!(fut);
let moved = fut;        // Moves the Pin<&mut> wrapper — OK
// fut.await;           // ❌ Error: fut was moved
let result = moved.await; // ✅ Use moved instead
}

#![allow(unused)]
fn main() {
let fut = async { 42 };
let pinned = Box::pin(fut); // Heap-pin — works with !Unpin
}

要点回顾 — Pin 与 Unpin

• Pin<P> 是一个包装器，防止被指向的值被移动——自引用状态机所必需
• Box::pin() 是在堆上固定 future 的安全、简便默认方式
• tokio::pin!() 在栈上固定——你可以移动 Pin<&mut> 包装器，但底层 future 保持不动
• Unpin 是自动 trait 的退出机制：实现 Unpin 的类型即使被固定也可以移动（大多数类型是 Unpin；async 块不是）

另见： 第 2 章 — Future Trait 了解 poll 中的 Pin<&mut Self>，第 5 章 — 状态机揭秘 了解为何异步状态机是自引用的

5. 状态机揭秘 🟢

你将学到：

• 编译器如何将 async fn 变换为枚举状态机
• 并排对比：源代码 vs 生成的状态
• 为何 async fn 中的大型栈分配会撑大 future 尺寸
• drop 优化：值在不再需要时立即释放

编译器实际生成了什么

当你写 async fn 时，编译器把你的顺序式代码变换为基于枚举的状态机。理解这一变换是理解异步 Rust 性能特征及其诸多怪异行为的关键。

并排对比：async fn 与状态机

#![allow(unused)]
fn main() {
// What you write:
async fn fetch_two_pages() -> String {
    let page1 = http_get("https://example.com/a").await;
    let page2 = http_get("https://example.com/b").await;
    format!("{page1}\n{page2}")
}
}

编译器在概念上生成类似下面的代码：

#![allow(unused)]
fn main() {
enum FetchTwoPagesStateMachine {
    // State 0: About to call http_get for page1
    Start,

    // State 1: Waiting for page1, holding the future
    WaitingPage1 {
        fut1: HttpGetFuture,
    },

    // State 2: Got page1, waiting for page2
    WaitingPage2 {
        page1: String,
        fut2: HttpGetFuture,
    },

    // Terminal state
    Complete,
}

impl Future for FetchTwoPagesStateMachine {
    type Output = String;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<String> {
        loop {
            match self.as_mut().get_mut() {
                Self::Start => {
                    let fut1 = http_get("https://example.com/a");
                    *self.as_mut().get_mut() = Self::WaitingPage1 { fut1 };
                }
                Self::WaitingPage1 { fut1 } => {
                    let page1 = match Pin::new(fut1).poll(cx) {
                        Poll::Ready(v) => v,
                        Poll::Pending => return Poll::Pending,
                    };
                    let fut2 = http_get("https://example.com/b");
                    *self.as_mut().get_mut() = Self::WaitingPage2 { page1, fut2 };
                }
                Self::WaitingPage2 { page1, fut2 } => {
                    let page2 = match Pin::new(fut2).poll(cx) {
                        Poll::Ready(v) => v,
                        Poll::Pending => return Poll::Pending,
                    };
                    let result = format!("{page1}\n{page2}");
                    *self.as_mut().get_mut() = Self::Complete;
                    return Poll::Ready(result);
                }
                Self::Complete => panic!("polled after completion"),
            }
        }
    }
}
}

注意：这种脱糖（desugaring）是概念性的。真实编译器输出使用 unsafe pin 投影——此处展示的 get_mut() 调用需要 Unpin，但异步状态机是 !Unpin。目的是说明 状态转换，而非产出可编译代码。

stateDiagram-v2
    [*] --> Start
    Start --> WaitingPage1: 创建 http_get future #1
    WaitingPage1 --> WaitingPage1: poll() → Pending
    WaitingPage1 --> WaitingPage2: poll() → Ready(page1)
    WaitingPage2 --> WaitingPage2: poll() → Pending
    WaitingPage2 --> Complete: poll() → Ready(page2)
    Complete --> [*]: 返回 format!("{page1}\\n{page2}")

各状态内容：

• WaitingPage1 — 存储 fut1: HttpGetFuture（page2 尚未分配）
• WaitingPage2 — 存储 page1: String、fut2: HttpGetFuture（fut1 已被 drop）

为何这对性能很重要

零成本：状态机是栈分配的枚举。每个 future 无需堆分配、无垃圾回收器、无装箱——除非你显式使用 Box::pin()。

尺寸：枚举的大小是其所有变体中的最大值。每个 .await 点创建一个新变体。这意味着：

#![allow(unused)]
fn main() {
async fn small() {
    let a: u8 = 0;
    yield_now().await;
    let b: u8 = 0;
    yield_now().await;
}
// Size ≈ max(size_of(u8), size_of(u8)) + discriminant + future sizes
//      ≈ small!

async fn big() {
    let buf: [u8; 1_000_000] = [0; 1_000_000]; // 1MB on the stack!
    some_io().await;
    process(&buf);
}
// Size ≈ 1MB + inner future sizes
// ⚠️ Don't stack-allocate huge buffers in async functions!
// Use Vec<u8> or Box<[u8]> instead.
}

drop 优化：状态机转换时，会 drop 不再需要的值。在上例中，从 WaitingPage1 转换到 WaitingPage2 时 fut1 被 drop——编译器会自动插入 drop。

实用规则：async fn 中的大型栈分配会撑大 future 的 尺寸。若在异步代码中出现栈溢出，检查是否有大型数组或 深度嵌套的 future。必要时用 Box::pin() 在堆上分配子 future。

练习：预测状态机

#![allow(unused)]
fn main() {
async fn pipeline(url: &str) -> Result<usize, Error> {
    let response = fetch(url).await?;
    let body = response.text().await?;
    let parsed = parse(body).await?;
    Ok(parsed.len())
}
}

要点回顾 — 状态机揭秘

• async fn 编译为枚举，每个 .await 点对应一个变体
• future 的尺寸 = 所有变体尺寸的最大值——大型栈值会撑大它
• 编译器在状态转换时自动插入 drop
• future 尺寸成为问题时，使用 Box::pin() 或堆分配

另见： 第 4 章 — Pin 与 Unpin 了解生成的枚举为何需要固定，第 6 章 — 手写 Future 了解如何自己构建这些状态机

6. 手动构建 Future 🟡

你将学到：

• 用基于线程的唤醒实现 TimerFuture
• 构建 Join 组合子：并发运行两个 future
• 构建 Select 组合子：让两个 future 竞速
• 组合子如何组合——层层嵌套的 future

一个简单的 Timer Future

现在让我们从零开始构建真正有用的 future。这能巩固第 2–5 章的理论。

TimerFuture：完整示例

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};

pub struct TimerFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

struct SharedState {
    completed: bool,
    waker: Option<Waker>,
}

impl TimerFuture {
    pub fn new(duration: Duration) -> Self {
        let shared_state = Arc::new(Mutex::new(SharedState {
            completed: false,
            waker: None,
        }));

        // Spawn a thread that sets completed=true after the duration
        let thread_shared_state = Arc::clone(&shared_state);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = thread_shared_state.lock().unwrap();
            state.completed = true;
            if let Some(waker) = state.waker.take() {
                waker.wake(); // Notify the executor
            }
        });

        TimerFuture { shared_state }
    }
}

impl Future for TimerFuture {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared_state.lock().unwrap();
        if state.completed {
            Poll::Ready(())
        } else {
            // Store the waker so the timer thread can wake us
            // IMPORTANT: Always update the waker — the executor may
            // have changed it between polls
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

// Usage:
// async fn example() {
//     println!("Starting timer...");
//     TimerFuture::new(Duration::from_secs(2)).await;
//     println!("Timer done!");
// }
//
// ⚠️ This spawns an OS thread per timer — fine for learning, but in
// production use `tokio::time::sleep` which is backed by a shared
// timer wheel and requires zero extra threads.
}

Join：并发运行两个 Future

Join 会轮询两个 future，并在两者都完成时结束。这就是 tokio::join! 的内部实现方式：

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

/// Polls two futures concurrently, returns both results as a tuple
pub struct Join<A, B>
where
    A: Future,
    B: Future,
{
    a: MaybeDone<A>,
    b: MaybeDone<B>,
}

enum MaybeDone<F: Future> {
    Pending(F),
    Done(F::Output),
    Taken, // Output has been taken
}

// MaybeDone<F> stores F::Output, which the compiler can't prove
// is Unpin even when F: Unpin. Since we only use Join with Unpin
// futures and never pin-project into fields, implementing Unpin
// by hand is safe and lets us call self.get_mut() in poll().
impl<A: Future + Unpin, B: Future + Unpin> Unpin for Join<A, B> {}

impl<A, B> Join<A, B>
where
    A: Future,
    B: Future,
{
    pub fn new(a: A, b: B) -> Self {
        Join {
            a: MaybeDone::Pending(a),
            b: MaybeDone::Pending(b),
        }
    }
}

impl<A, B> Future for Join<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    type Output = (A::Output, B::Output);

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();

        // Poll A if not done
        if let MaybeDone::Pending(ref mut fut) = this.a {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.a = MaybeDone::Done(val);
            }
        }

        // Poll B if not done
        if let MaybeDone::Pending(ref mut fut) = this.b {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.b = MaybeDone::Done(val);
            }
        }

        // Both done?
        match (&this.a, &this.b) {
            (MaybeDone::Done(_), MaybeDone::Done(_)) => {
                // Take both outputs
                let a_val = match std::mem::replace(&mut this.a, MaybeDone::Taken) {
                    MaybeDone::Done(v) => v,
                    _ => unreachable!(),
                };
                let b_val = match std::mem::replace(&mut this.b, MaybeDone::Taken) {
                    MaybeDone::Done(v) => v,
                    _ => unreachable!(),
                };
                Poll::Ready((a_val, b_val))
            }
            _ => Poll::Pending, // At least one is still pending
        }
    }
}

// Usage (async blocks are !Unpin, so wrap them with Box::pin):
// let (page1, page2) = Join::new(
//     Box::pin(http_get("https://example.com/a")),
//     Box::pin(http_get("https://example.com/b")),
// ).await;
// Both requests run concurrently!
}

关键洞见：此处的「并发」指在同一线程上交错执行。 Join 不会创建线程——它在同一次 poll() 调用中轮询两个 future。 这是协作式并发（cooperative concurrency），而非并行（parallelism）。

graph LR
    subgraph "Future 组合子"
        direction TB
        TIMER["TimerFuture<br/>单个 future，延迟后唤醒"]
        JOIN["Join&lt;A, B&gt;<br/>等待两者都完成"]
        SELECT["Select&lt;A, B&gt;<br/>等待先完成者"]
        RETRY["RetryFuture<br/>失败时重新创建"]
    end

    TIMER --> JOIN
    TIMER --> SELECT
    SELECT --> RETRY

    style TIMER fill:#d4efdf,stroke:#27ae60,color:#000
    style JOIN fill:#e8f4f8,stroke:#2980b9,color:#000
    style SELECT fill:#fef9e7,stroke:#f39c12,color:#000
    style RETRY fill:#fadbd8,stroke:#e74c3c,color:#000

Select：让两个 Future 竞速

Select 在任一 future 先完成时结束（另一个会被丢弃）：

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

pub enum Either<A, B> {
    Left(A),
    Right(B),
}

/// Returns whichever future completes first; drops the other
pub struct Select<A, B> {
    a: A,
    b: B,
}

impl<A, B> Select<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    pub fn new(a: A, b: B) -> Self {
        Select { a, b }
    }
}

impl<A, B> Future for Select<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    type Output = Either<A::Output, B::Output>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Poll A first
        if let Poll::Ready(val) = Pin::new(&mut self.a).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

        // Then poll B
        if let Poll::Ready(val) = Pin::new(&mut self.b).poll(cx) {
            return Poll::Ready(Either::Right(val));
        }

        Poll::Pending
    }
}

// Usage with timeout:
// match Select::new(http_get(url), TimerFuture::new(timeout)).await {
//     Either::Left(response) => println!("Got response: {}", response),
//     Either::Right(()) => println!("Request timed out!"),
// }
}

公平性说明：我们的 Select 总是先轮询 A——若两者都已就绪，A 总是获胜。Tokio 的 select! 宏会随机化轮询顺序以保证公平性。

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

pub struct RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    factory: F,
    current: Option<Pin<Box<Fut>>>,
    remaining: usize,
    last_error: Option<E>,
}

impl<F, Fut, T, E> RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    pub fn new(max_attempts: usize, factory: F) -> Self {
        let current = Some(Box::pin((factory)()));
        RetryFuture {
            factory,
            current,
            remaining: max_attempts.saturating_sub(1),
            last_error: None,
        }
    }
}

impl<F, Fut, T, E> Future for RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut + Unpin,
    Fut: Future<Output = Result<T, E>>,
    E: Unpin,
{
    type Output = Result<T, E>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Pin<Box<Fut>> is always Unpin, so the struct is Unpin when F and E are.
        // This lets us safely use get_mut() without any unsafe code.
        loop {
            if let Some(ref mut fut) = self.current {
                match fut.as_mut().poll(cx) {
                    Poll::Ready(Ok(val)) => return Poll::Ready(Ok(val)),
                    Poll::Ready(Err(e)) => {
                        self.last_error = Some(e);
                        if self.remaining > 0 {
                            self.remaining -= 1;
                            self.current = Some(Box::pin((self.factory)()));
                            // Loop to poll the new future immediately
                        } else {
                            return Poll::Ready(Err(self.last_error.take().unwrap()));
                        }
                    }
                    Poll::Pending => return Poll::Pending,
                }
            } else {
                return Poll::Ready(Err(self.last_error.take().unwrap()));
            }
        }
    }
}

// Usage:
// let result = RetryFuture::new(3, || async {
//     http_get("https://flaky-server.com/api").await
// }).await;
}

要点回顾 — 手动构建 Future

• 一个 future 需要三样东西：状态、poll() 实现，以及 waker 注册
• Join 轮询两个子 future；Select 返回先完成的那一个
• 组合子本身也是包装其他 future 的 future——层层嵌套
• 手动构建 future 能加深理解，但生产环境请用 tokio::join!/select!

另见： 第 2 章 — Future Trait 了解 Trait 定义，第 8 章 — Tokio 深入 了解生产级等价实现

7. 执行器与运行时 🟡

你将学到：

• 执行器（executor）做什么：高效地 poll 与休眠
• 六大主要运行时：mio、io_uring、tokio、async-std、smol、embassy
• 选择合适运行时的决策树
• 为何与运行时无关（runtime-agnostic）的库设计很重要

执行器做什么

执行器有两项职责：

1. 在 future 可以推进时轮询 future
2. 当没有 future 就绪时高效休眠（使用操作系统 I/O 通知 API）

graph TB
    subgraph Executor["执行器（如 tokio）"]
        QUEUE["任务队列"]
        POLLER["I/O 轮询器<br/>（epoll/kqueue/io_uring）"]
        THREADS["工作线程池"]
    end

    subgraph Tasks
        T1["任务 1<br/>（HTTP 请求）"]
        T2["任务 2<br/>（数据库查询）"]
        T3["任务 3<br/>（文件读取）"]
    end

    subgraph OS["操作系统"]
        NET["网络栈"]
        DISK["磁盘 I/O"]
    end

    T1 --> QUEUE
    T2 --> QUEUE
    T3 --> QUEUE
    QUEUE --> THREADS
    THREADS -->|"poll()"| T1
    THREADS -->|"poll()"| T2
    THREADS -->|"poll()"| T3
    POLLER <-->|"注册/通知"| NET
    POLLER <-->|"注册/通知"| DISK
    POLLER -->|"唤醒任务"| QUEUE

    style Executor fill:#e3f2fd,color:#000
    style OS fill:#f3e5f5,color:#000

mio：基础层

mio（Metal I/O）不是执行器——它是最底层的跨平台 I/O 通知库。它封装了 epoll（Linux）、kqueue（macOS/BSD）和 IOCP（Windows）。

#![allow(unused)]
fn main() {
// Conceptual mio usage (simplified):
use mio::{Events, Interest, Poll, Token};
use mio::net::TcpListener;

let mut poll = Poll::new()?;
let mut events = Events::with_capacity(128);

let mut server = TcpListener::bind("0.0.0.0:8080")?;
poll.registry().register(&mut server, Token(0), Interest::READABLE)?;

// Event loop — blocks until something happens
loop {
    poll.poll(&mut events, None)?; // Sleeps until I/O event
    for event in events.iter() {
        match event.token() {
            Token(0) => { /* server has a new connection */ }
            _ => { /* other I/O ready */ }
        }
    }
}
}

大多数开发者不会直接碰 mio——tokio 和 smol 都构建在它之上。

io_uring：基于完成通知的未来

Linux 的 io_uring（内核 5.1+）代表了与 mio/epoll 所用的就绪式（readiness-based）I/O 模型的根本转变：

Readiness-based (epoll / mio / tokio):
  1. Ask: "Is this socket readable?"     → epoll_wait()
  2. Kernel: "Yes, it's ready"           → EPOLLIN event
  3. App:   read(fd, buf)                → might still block briefly!

Completion-based (io_uring):
  1. Submit: "Read from this socket into this buffer"  → SQE
  2. Kernel: does the read asynchronously
  3. App:   gets completed result with data            → CQE

graph LR
    subgraph "就绪模型（epoll）"
        A1["应用：可读了吗？"] --> K1["内核：是的"]
        K1 --> A2["应用：现在 read()"]
        A2 --> K2["内核：这是数据"]
    end

    subgraph "完成模型（io_uring）"
        B1["应用：帮我读这个"] --> K3["内核：处理中..."]
        K3 --> B2["应用：得到结果 + 数据"]
    end

    style B1 fill:#c8e6c9,color:#000
    style B2 fill:#c8e6c9,color:#000

所有权挑战：io_uring 要求内核在操作完成前拥有缓冲区。这与 Rust 标准 AsyncRead Trait 借用缓冲区的方式冲突。因此 tokio-uring 使用不同的 I/O Trait：

#![allow(unused)]
fn main() {
// Standard tokio (readiness-based) — borrows the buffer:
let n = stream.read(&mut buf).await?;  // buf is borrowed

// tokio-uring (completion-based) — takes ownership of the buffer:
let (result, buf) = stream.read(buf).await;  // buf is moved in, returned back
let n = result?;
}

// Cargo.toml: tokio-uring = "0.5"
// NOTE: Linux-only, requires kernel 5.1+

fn main() {
    tokio_uring::start(async {
        let file = tokio_uring::fs::File::open("data.bin").await.unwrap();
        let buf = vec![0u8; 4096];
        let (result, buf) = file.read_at(buf, 0).await;
        let bytes_read = result.unwrap();
        println!("Read {} bytes: {:?}", bytes_read, &buf[..bytes_read]);
    });
}

何时使用 io_uring：高吞吐文件 I/O 或网络，且系统调用开销是瓶颈时（数据库、存储引擎、服务 10 万+ 连接的代理）。对大多数应用，带 epoll 的标准 tokio 是正确选择。

tokio：开箱即用的运行时

Rust 生态中占主导地位的异步运行时。Axum、Hyper、Tonic 及大多数生产级 Rust 服务器都在使用它。

// Cargo.toml:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

#[tokio::main]
async fn main() {
    // Spawns a multi-threaded runtime with work-stealing scheduler
    let handle = tokio::spawn(async {
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        "done"
    });

    let result = handle.await.unwrap();
    println!("{result}");
}

tokio 功能：定时器、I/O、TCP/UDP、Unix socket、信号处理、同步原语（Mutex、RwLock、Semaphore、channel）、fs、进程、tracing 集成。

async-std：标准库镜像

用异步版本镜像 std API。不如 tokio 流行，但对初学者更简单。

// Cargo.toml:
// [dependencies]
// async-std = { version = "1", features = ["attributes"] }

#[async_std::main]
async fn main() {
    use async_std::fs;
    let content = fs::read_to_string("hello.txt").await.unwrap();
    println!("{content}");
}

smol：极简运行时

小巧、零依赖的异步运行时。适合希望支持异步又不想拉入 tokio 的库。

// Cargo.toml:
// [dependencies]
// smol = "2"

fn main() {
    smol::block_on(async {
        let result = smol::unblock(|| {
            // Runs blocking code on a thread pool
            std::fs::read_to_string("hello.txt")
        }).await.unwrap();
        println!("{result}");
    });
}

embassy：嵌入式异步（no_std）

面向嵌入式系统的异步运行时。无需堆分配，不要求 std。

// Runs on microcontrollers (e.g., STM32, nRF52, RP2040)
#[embassy_executor::main]
async fn main(spawner: embassy_executor::Spawner) {
    // Blink an LED with async/await — no RTOS needed!
    let mut led = Output::new(p.PA5, Level::Low, Speed::Low);
    loop {
        led.set_high();
        Timer::after(Duration::from_millis(500)).await;
        led.set_low();
        Timer::after(Duration::from_millis(500)).await;
    }
}

运行时决策树

graph TD
    START["选择运行时"]

    Q1{"在构建<br/>网络服务器？"}
    Q2{"需要 tokio 生态<br/>（Axum、Tonic、Hyper）？"}
    Q3{"在写库？"}
    Q4{"嵌入式 /<br/>no_std？"}
    Q5{"想要最少<br/>依赖？"}

    TOKIO["🟢 tokio<br/>生态最好，最流行"]
    SMOL["🔵 smol<br/>极简，无生态锁定"]
    EMBASSY["🟠 embassy<br/>嵌入式优先，无 alloc"]
    ASYNC_STD["🟣 async-std<br/>类 std API，适合学习"]
    AGNOSTIC["🔵 与运行时无关<br/>仅用 futures crate"]

    START --> Q1
    Q1 -->|是| Q2
    Q1 -->|否| Q3
    Q2 -->|是| TOKIO
    Q2 -->|否| Q5
    Q3 -->|是| AGNOSTIC
    Q3 -->|否| Q4
    Q4 -->|是| EMBASSY
    Q4 -->|否| Q5
    Q5 -->|是| SMOL
    Q5 -->|否| ASYNC_STD

    style TOKIO fill:#c8e6c9,color:#000
    style SMOL fill:#bbdefb,color:#000
    style EMBASSY fill:#ffe0b2,color:#000
    style ASYNC_STD fill:#e1bee7,color:#000
    style AGNOSTIC fill:#bbdefb,color:#000

运行时对比表

// ----- tokio version -----
// Cargo.toml: tokio = { version = "1", features = ["full"] }
#[tokio::main]
async fn main() {
    let (url_result, file_result) = tokio::join!(
        async {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            "Response from URL"
        },
        async {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            "Contents of file"
        },
    );
    println!("URL: {url_result}, File: {file_result}");
}

// ----- smol version -----
// Cargo.toml: smol = "2", futures-lite = "2"
fn main() {
    smol::block_on(async {
        let (url_result, file_result) = futures_lite::future::zip(
            async {
                smol::Timer::after(std::time::Duration::from_millis(100)).await;
                "Response from URL"
            },
            async {
                smol::Timer::after(std::time::Duration::from_millis(50)).await;
                "Contents of file"
            },
        ).await;
        println!("URL: {url_result}, File: {file_result}");
    });
}

// ----- async-std version -----
// Cargo.toml: async-std = { version = "1", features = ["attributes"] }
#[async_std::main]
async fn main() {
    let (url_result, file_result) = futures::future::join(
        async {
            async_std::task::sleep(std::time::Duration::from_millis(100)).await;
            "Response from URL"
        },
        async {
            async_std::task::sleep(std::time::Duration::from_millis(50)).await;
            "Contents of file"
        },
    ).await;
    println!("URL: {url_result}, File: {file_result}");
}

要点回顾 — 执行器与运行时

• 执行器的职责：在 waker 唤醒时 poll future，并用 OS I/O API 高效休眠
• tokio 是服务器默认选择；smol 适合最小体积；embassy 面向嵌入式
• 业务逻辑应依赖 std::future::Future，而非特定运行时
• io_uring（Linux 5.1+）是高性能 I/O 的未来，但生态仍在成熟中

另见： 第 8 章 — Tokio 深入 了解 tokio 细节，第 9 章 — 何时不该用 Tokio 了解替代方案

8. Tokio 深入 🟡

你将学到：

• 运行时风格：multi-thread 与 current-thread，以及各自适用场景
• tokio::spawn、'static 要求与 JoinHandle
• 任务取消语义（drop 即取消）
• 同步原语：Mutex、RwLock、Semaphore，以及四种 channel 类型

运行时风格：Multi-Thread 与 Current-Thread

Tokio 提供两种运行时配置：

// Multi-threaded (default with #[tokio::main])
// Uses a work-stealing thread pool — tasks can move between threads
#[tokio::main]
async fn main() {
    // N worker threads (default = number of CPU cores)
    // Tasks are Send + 'static
}

// Current-thread — everything runs on one thread
#[tokio::main(flavor = "current_thread")]
async fn main() {
    // Single-threaded — tasks don't need to be Send
    // Lighter weight, good for simple tools or WASM
}

// Manual runtime construction:
let rt = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(4)
    .enable_all()
    .build()
    .unwrap();

rt.block_on(async {
    println!("Running on custom runtime");
});

graph TB
    subgraph "多线程（默认）"
        MT_Q1["线程 1<br/>任务 A、任务 D"]
        MT_Q2["线程 2<br/>任务 B"]
        MT_Q3["线程 3<br/>任务 C、任务 E"]
        STEAL["工作窃取：<br/>空闲线程从繁忙线程偷任务"]
        MT_Q1 <--> STEAL
        MT_Q2 <--> STEAL
        MT_Q3 <--> STEAL
    end

    subgraph "Current-Thread"
        ST_Q["单线程<br/>任务 A → 任务 B → 任务 C → 任务 D"]
    end

    style MT_Q1 fill:#c8e6c9,color:#000
    style MT_Q2 fill:#c8e6c9,color:#000
    style MT_Q3 fill:#c8e6c9,color:#000
    style ST_Q fill:#bbdefb,color:#000

tokio::spawn 与 ’static 要求

tokio::spawn 将 future 放入运行时的任务队列。因为它可能在任意工作线程、任意时刻运行，future 必须是 Send + 'static：

#![allow(unused)]
fn main() {
use tokio::task;

async fn example() {
    let data = String::from("hello");

    // ✅ Works: move ownership into the task
    let handle = task::spawn(async move {
        println!("{data}");
        data.len()
    });

    let len = handle.await.unwrap();
    println!("Length: {len}");
}

async fn problem() {
    let data = String::from("hello");

    // ❌ FAILS: data is borrowed, not 'static
    // task::spawn(async {
    //     println!("{data}"); // borrows `data` — not 'static
    // });

    // ❌ FAILS: Rc is not Send
    // let rc = std::rc::Rc::new(42);
    // task::spawn(async move {
    //     println!("{rc}"); // Rc is !Send — can't cross thread boundary
    // });
}
}

为何需要 'static？ 被 spawn 的任务独立运行——它可能比创建它的作用域活得更久。编译器无法证明引用仍然有效，因此要求拥有数据。

为何需要 Send？ 任务可能在挂起时与恢复时处于不同线程。跨越 .await 点持有的所有数据必须可在线程间安全传递。

#![allow(unused)]
fn main() {
// Common pattern: clone shared data into the task
let shared = Arc::new(config);

for i in 0..10 {
    let shared = Arc::clone(&shared); // Clone the Arc, not the data
    tokio::spawn(async move {
        process_item(i, &shared).await;
    });
}
}

JoinHandle 与任务取消

#![allow(unused)]
fn main() {
use tokio::task::JoinHandle;
use tokio::time::{sleep, Duration};

async fn cancellation_example() {
    let handle: JoinHandle<String> = tokio::spawn(async {
        sleep(Duration::from_secs(10)).await;
        "completed".to_string()
    });

    // Cancel the task by dropping the handle? NO — task keeps running!
    // drop(handle); // Task continues in the background

    // To actually cancel, call abort():
    handle.abort();

    // Awaiting an aborted task returns JoinError
    match handle.await {
        Ok(val) => println!("Got: {val}"),
        Err(e) if e.is_cancelled() => println!("Task was cancelled"),
        Err(e) => println!("Task panicked: {e}"),
    }
}
}

重要：在 tokio 中，drop JoinHandle 不会取消任务。 任务会变成分离状态并继续运行。必须显式调用 .abort() 才能取消。这与直接 drop Future 不同—— 后者会取消/丢弃底层计算。

Tokio 同步原语

Tokio 提供异步感知的同步原语。核心原则：不要在 .await 点之间使用 std::sync::Mutex。

#![allow(unused)]
fn main() {
use tokio::sync::{Mutex, RwLock, Semaphore, mpsc, oneshot, broadcast, watch};

// --- Mutex ---
// Async mutex: the lock() method is async and won't block the thread
let data = Arc::new(Mutex::new(vec![1, 2, 3]));
{
    let mut guard = data.lock().await; // Non-blocking lock
    guard.push(4);
} // Guard dropped here — lock released

// --- Channels ---
// mpsc: Multiple producer, single consumer
let (tx, mut rx) = mpsc::channel::<String>(100); // Bounded buffer

tokio::spawn(async move {
    tx.send("hello".into()).await.unwrap();
});

let msg = rx.recv().await.unwrap();

// oneshot: Single value, single consumer
let (tx, rx) = oneshot::channel::<i32>();
tx.send(42).unwrap(); // No await needed — either sends or fails
let val = rx.await.unwrap();

// broadcast: Multiple producers, multiple consumers (all get every message)
let (tx, _) = broadcast::channel::<String>(100);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

// watch: Single value, multiple consumers (only latest value)
let (tx, rx) = watch::channel(0u64);
tx.send(42).unwrap();
println!("Latest: {}", *rx.borrow());
}

说明： 以下 channel 示例为简洁起见使用 .unwrap()。 生产环境应妥善处理发送/接收错误——.send() 失败表示 接收端已 drop，.recv() 失败表示 channel 已关闭。

graph LR
    subgraph "Channel 类型"
        direction TB
        MPSC["mpsc<br/>N→1<br/>有界队列"]
        ONESHOT["oneshot<br/>1→1<br/>单值"]
        BROADCAST["broadcast<br/>N→N<br/>所有接收者都收到"]
        WATCH["watch<br/>1→N<br/>仅最新值"]
    end

    P1["生产者 1"] --> MPSC
    P2["生产者 2"] --> MPSC
    MPSC --> C1["消费者"]

    P3["生产者"] --> ONESHOT
    ONESHOT --> C2["消费者"]

    P4["生产者"] --> BROADCAST
    BROADCAST --> C3["消费者 1"]
    BROADCAST --> C4["消费者 2"]

    P5["生产者"] --> WATCH
    WATCH --> C5["消费者 1"]
    WATCH --> C6["消费者 2"]

案例研究：为通知服务选择正确的 Channel

你在构建通知服务，需求如下：

• 多个 API handler 产生事件
• 单个后台任务批量发送
• 配置监视器在运行时更新速率限制
• 关闭信号必须到达所有组件

各场景用哪种 channel？

graph LR
    subgraph "通知服务"
        direction TB
        API1["API Handler 1"] -->|mpsc| BATCH["Batcher"]
        API2["API Handler 2"] -->|mpsc| BATCH
        CONFIG["配置监视器"] -->|watch| RATE["速率限制器"]
        CTRL["Ctrl+C"] -->|broadcast| API1
        CTRL -->|broadcast| BATCH
        CTRL -->|broadcast| RATE
    end

    style API1 fill:#d4efdf,stroke:#27ae60,color:#000
    style API2 fill:#d4efdf,stroke:#27ae60,color:#000
    style BATCH fill:#e8f4f8,stroke:#2980b9,color:#000
    style CONFIG fill:#fef9e7,stroke:#f39c12,color:#000
    style RATE fill:#fef9e7,stroke:#f39c12,color:#000
    style CTRL fill:#fadbd8,stroke:#e74c3c,color:#000

#![allow(unused)]
fn main() {
use std::future::Future;
use std::sync::Arc;
use tokio::sync::Semaphore;

async fn run_with_limit<F, Fut, T>(tasks: Vec<F>, limit: usize) -> Vec<T>
where
    F: FnOnce() -> Fut + Send + 'static,
    Fut: Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    let semaphore = Arc::new(Semaphore::new(limit));
    let mut handles = Vec::new();

    for task in tasks {
        let permit = Arc::clone(&semaphore);
        let handle = tokio::spawn(async move {
            let _permit = permit.acquire().await.unwrap();
            // Permit is held while task runs, then dropped
            task().await
        });
        handles.push(handle);
    }

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    results
}

// Usage:
// let tasks: Vec<_> = urls.into_iter().map(|url| {
//     move || async move { fetch(url).await }
// }).collect();
// let results = run_with_limit(tasks, 10).await; // Max 10 concurrent
}

要点回顾 — Tokio 深入

• 服务器用 multi_thread（默认）；CLI 工具、测试或 !Send 类型用 current_thread
• tokio::spawn 要求 'static future——用 Arc 或 channel 共享数据
• drop JoinHandle 不会取消任务——需显式调用 .abort()
• 按需求选同步原语：共享状态用 Mutex，并发上限用 Semaphore，通信用 mpsc/oneshot/broadcast/watch

另见： 第 9 章 — 何时不该用 Tokio 了解 spawn 的替代方案，第 12 章 — 常见陷阱 了解跨 await 持有 MutexGuard 的 bug

9. 何时不该用 Tokio 🟡

你将学到：

• 'static 问题：tokio::spawn 如何迫使你到处使用 Arc
• 面向 !Send future 的 LocalSet
• 便于借用的并发：FuturesUnordered（无需 spawn）
• 可管理的任务组：JoinSet
• 编写与运行时无关的库

graph TD
    START["需要并发 future？"] --> STATIC{"future 能否是 'static？"}
    STATIC -->|是| SEND{"future 是否 Send？"}
    STATIC -->|否| FU["FuturesUnordered<br/>在当前任务上运行"]
    SEND -->|是| SPAWN["tokio::spawn<br/>多线程"]
    SEND -->|否| LOCAL["LocalSet<br/>单线程"]
    SPAWN --> MANAGE{"需要跟踪/取消任务？"}
    MANAGE -->|是| JOINSET["JoinSet / TaskTracker"]
    MANAGE -->|否| HANDLE["JoinHandle"]

    style START fill:#f5f5f5,stroke:#333,color:#000
    style FU fill:#d4efdf,stroke:#27ae60,color:#000
    style SPAWN fill:#e8f4f8,stroke:#2980b9,color:#000
    style LOCAL fill:#fef9e7,stroke:#f39c12,color:#000
    style JOINSET fill:#e8daef,stroke:#8e44ad,color:#000
    style HANDLE fill:#e8f4f8,stroke:#2980b9,color:#000

’static Future 问题

Tokio 的 spawn 要求 'static future。这意味着你不能在 spawn 的任务里借用局部数据：

#![allow(unused)]
fn main() {
async fn process_items(items: &[String]) {
    // ❌ Can't do this — items is borrowed, not 'static
    // for item in items {
    //     tokio::spawn(async {
    //         process(item).await;
    //     });
    // }

    // 😐 Workaround 1: Clone everything
    for item in items {
        let item = item.clone();
        tokio::spawn(async move {
            process(&item).await;
        });
    }

    // 😐 Workaround 2: Use Arc
    let items = Arc::new(items.to_vec());
    for i in 0..items.len() {
        let items = Arc::clone(&items);
        tokio::spawn(async move {
            process(&items[i]).await;
        });
    }
}
}

这很烦人！在 Go 里你可以 go func() { use(item) } 用闭包就行。在 Rust 里，所有权系统迫使你思考谁拥有什么、能活多久。

tokio::spawn 的替代方案

并非每个问题都需要 spawn。下面三个工具各自解决不同约束：

#![allow(unused)]
fn main() {
// 1. FuturesUnordered — avoids 'static entirely (no spawn!)
use futures::stream::{FuturesUnordered, StreamExt};

async fn process_items(items: &[String]) {
    let futures: FuturesUnordered<_> = items
        .iter()
        .map(|item| async move {
            // ✅ Can borrow item — no spawn, no 'static needed!
            process(item).await
        })
        .collect();

    // Drive all futures to completion
    futures.for_each(|result| async move {
        println!("Result: {result:?}");
    }).await;
}

// 2. tokio::task::LocalSet — run !Send futures on current thread
//    ⚠️  Still requires 'static — solves Send, not 'static
use tokio::task::LocalSet;

let local_set = LocalSet::new();
local_set.run_until(async {
    tokio::task::spawn_local(async {
        // Can use Rc, Cell, and other !Send types here
        let rc = std::rc::Rc::new(42);
        println!("{rc}");
    }).await.unwrap();
}).await;

// 3. tokio JoinSet (tokio 1.21+) — managed set of spawned tasks
//    ⚠️  Still requires 'static + Send — solves task *management*,
//    not the 'static problem. Useful for tracking, aborting, and
//    joining a dynamic group of tasks.
use tokio::task::JoinSet;

async fn with_joinset() {
    let mut set = JoinSet::new();

    for i in 0..10 {
        // i is Copy and moved into the closure — already 'static.
        // You'd still need Arc or clone for borrowed data.
        set.spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            i * 2
        });
    }

    while let Some(result) = set.join_next().await {
        println!("Task completed: {:?}", result.unwrap());
    }
}
}

哪种工具解决哪种问题？

遇到的约束	工具	避免 'static？	避免 Send？
无法让 future 成为 'static	FuturesUnordered	✅ 是	✅ 是
future 是 'static 但 !Send	LocalSet	❌ 否	✅ 是
需要跟踪 / 取消 spawn 的任务	JoinSet	❌ 否	❌ 否

库的轻量运行时

如果你在写库——不要把用户绑死在 tokio 上：

#![allow(unused)]
fn main() {
// ❌ BAD: Library forces tokio on users
pub async fn my_lib_function() {
    tokio::time::sleep(Duration::from_secs(1)).await;
    // Now your users MUST use tokio
}

// ✅ GOOD: Library is runtime-agnostic
pub async fn my_lib_function() {
    // Use only types from std::future and futures crate
    do_computation().await;
}

// ✅ GOOD: Accept a generic future for I/O operations
pub async fn fetch_with_retry<F, Fut, T, E>(
    operation: F,
    max_retries: usize,
) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    for attempt in 0..max_retries {
        match operation().await {
            Ok(val) => return Ok(val),
            Err(e) if attempt == max_retries - 1 => return Err(e),
            Err(_) => continue,
        }
    }
    unreachable!()
}
}

经验法则：库应依赖 futures crate，而非 tokio。 应用应依赖 tokio（或所选运行时）。 这样生态才能可组合。

#![allow(unused)]
fn main() {
use futures::stream::{FuturesUnordered, StreamExt};
use tokio::time::{sleep, Duration};

// Version 1: tokio::spawn — requires 'static, must clone
async fn lengths_with_spawn(items: &[String]) -> Vec<usize> {
    let mut handles = Vec::new();
    for item in items {
        let owned = item.clone(); // Must clone — spawn requires 'static
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(10)).await;
            owned.len()
        }));
    }

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    results
}

// Version 2: FuturesUnordered — borrows data, no clone needed
async fn lengths_without_spawn(items: &[String]) -> Vec<usize> {
    let futures: FuturesUnordered<_> = items
        .iter()
        .map(|item| async move {
            sleep(Duration::from_millis(10)).await;
            item.len() // ✅ Borrows item — no clone!
        })
        .collect();

    futures.collect().await
}

#[tokio::test]
async fn test_both_versions() {
    let items = vec!["hello".into(), "world".into(), "rust".into()];

    let v1 = lengths_with_spawn(&items).await;
    // Note: v1 preserves insertion order (sequential join)

    let mut v2 = lengths_without_spawn(&items).await;
    v2.sort(); // FuturesUnordered returns in completion order

    assert_eq!(v1, vec![5, 5, 4]);
    assert_eq!(v2, vec![4, 5, 5]);
}
}

要点回顾 — 何时不该用 Tokio

• FuturesUnordered 在当前任务上并发运行 future——无 'static 要求
• LocalSet 让 !Send future 在单线程执行器上运行
• JoinSet（tokio 1.21+）提供可管理任务组与自动清理
• 库：仅依赖 std::future::Future + futures crate，不要直接依赖 tokio

另见： 第 8 章 — Tokio 深入 了解何时 spawn 合适，第 11 章 — Stream 了解 buffer_unordered() 作为另一种并发限制手段

10. 异步 Trait 🟡

你将学到：

• 为何 Trait 中的异步方法稳定化花了多年
• RPITIT：原生异步 Trait 方法（Rust 1.75+）
• dyn 分发挑战与通过 trait_variant 的 Send 约束
• 异步闭包（Rust 1.85+）：async Fn() 与 async FnOnce()

graph TD
    subgraph "异步 Trait 方案"
        direction TB
        RPITIT["RPITIT（Rust 1.75+）<br/>trait 中 async fn<br/>仅静态分发"]
        VARIANT["trait_variant<br/>自动生成 Send 变体<br/>仅静态分发"]
        BOXED["Box&lt;dyn Future&gt;<br/>手动装箱<br/>到处可用"]
        CLOSURE["异步闭包（1.85+）<br/>async Fn() / async FnOnce()<br/>回调与中间件"]
    end

    RPITIT -->|"需要 Send？"| VARIANT
    RPITIT -->|"需要 dyn？"| BOXED
    CLOSURE -->|"替代"| BOXED

    style RPITIT fill:#d4efdf,stroke:#27ae60,color:#000
    style VARIANT fill:#e8f4f8,stroke:#2980b9,color:#000
    style BOXED fill:#fef9e7,stroke:#f39c12,color:#000
    style CLOSURE fill:#e8daef,stroke:#8e44ad,color:#000

历史：为何花了这么久

Trait 中的异步方法是 Rust 多年来最被请求的功能。问题在于：

#![allow(unused)]
fn main() {
// This didn't compile until Rust 1.75 (Dec 2023):
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
}
// Why? Because async fn returns `impl Future<Output = T>`,
// and `impl Trait` in trait return position wasn't supported.
}

根本挑战：当 Trait 方法返回 impl Future 时，每个实现返回不同的具体类型。编译器需要知道返回类型的大小，但 Trait 方法是动态分发的。

RPITIT：Trait 返回位置的 Impl Trait

自 Rust 1.75 起，静态分发下可以直接这样写：

#![allow(unused)]
fn main() {
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
    // Desugars to:
    // fn get(&self, key: &str) -> impl Future<Output = Option<String>>;
}

struct InMemoryStore {
    data: std::collections::HashMap<String, String>,
}

impl DataStore for InMemoryStore {
    async fn get(&self, key: &str) -> Option<String> {
        self.data.get(key).cloned()
    }
}

// ✅ Works with generics (static dispatch):
async fn lookup<S: DataStore>(store: &S, key: &str) {
    if let Some(val) = store.get(key).await {
        println!("{key} = {val}");
    }
}
}

dyn 分发与 Send 约束

限制：不能直接使用 dyn DataStore，因为编译器不知道返回的 future 的大小：

#![allow(unused)]
fn main() {
// ❌ Doesn't work:
// async fn lookup_dyn(store: &dyn DataStore, key: &str) { ... }
// Error: the trait `DataStore` is not dyn-compatible because method `get`
//        is `async`

// ✅ Workaround: Return a boxed future
trait DynDataStore {
    fn get(&self, key: &str) -> Pin<Box<dyn Future<Output = Option<String>> + Send + '_>>;
}
}

Send 问题：在多线程运行时中，spawn 的任务必须是 Send。但异步 Trait 方法不会自动加上 Send 约束：

#![allow(unused)]
fn main() {
trait Worker {
    async fn run(self); // Future might or might not be Send
}

struct MyWorker;

impl Worker for MyWorker {
    async fn run(self) {
        // If this uses !Send types, the future is !Send
        let rc = std::rc::Rc::new(42);
        some_work().await;
        println!("{rc}");
    }
}

// ❌ This fails because the future is !Send (Rc is !Send):
// tokio::spawn(worker.run()); // Requires Send + 'static
//
// Note: We use `self` (owned) here because tokio::spawn also
// requires 'static — a future borrowing &self can't be 'static.
// Even without Rc, `async fn run(&self)` wouldn't be spawnable.
}

trait_variant crate

trait_variant crate（来自 Rust 异步工作组）会自动生成 Send 变体：

#![allow(unused)]
fn main() {
// Cargo.toml: trait-variant = "0.1"

#[trait_variant::make(SendDataStore: Send)]
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// Now you have two traits:
// - DataStore: no Send bound on the futures
// - SendDataStore: all futures are Send
// Both have the same methods, implementors implement DataStore
// and get SendDataStore for free if their futures are Send.

// Use SendDataStore when you need to spawn tasks:
async fn spawn_lookup<S: SendDataStore + 'static>(store: Arc<S>) {
    tokio::spawn(async move {
        store.get("key").await;
    });
}

// ⚠️ Note: trait_variant does NOT enable dyn dispatch.
// The generated trait still uses `impl Future`, so `dyn SendDataStore`
// is not object-safe. For dyn dispatch, you still need manual boxing
// (see the Box::pin approach above) or the `async-trait` crate.
}

速查：异步 Trait

建议：新代码（Rust 1.75+）使用原生异步 Trait。需要为 spawn 任务加 Send 约束时用 trait_variant。需要 dyn 分发时用手动 Box::pin 或 async-trait crate。原生 方案在静态分发下零成本。

异步闭包（Rust 1.85+）

自 Rust 1.85 起，异步闭包（async closures）已稳定——捕获环境并返回 future 的闭包：

#![allow(unused)]
fn main() {
// Before 1.85: awkward workaround
let urls = vec!["https://a.com", "https://b.com"];
let fetchers: Vec<_> = urls.iter().map(|url| {
    let url = url.to_string();
    // Returns a non-async closure that returns an async block
    move || async move { reqwest::get(&url).await }
}).collect();

// After 1.85: async closures just work
let fetchers: Vec<_> = urls.iter().map(|url| {
    async move || { reqwest::get(url).await }
    // ↑ This is an async closure — captures url, returns a Future
}).collect();
}

异步闭包实现新的 AsyncFn、AsyncFnMut、AsyncFnOnce Trait，分别对应 Fn、FnMut、FnOnce：

#![allow(unused)]
fn main() {
// Generic function accepting an async closure
async fn retry<F>(max: usize, f: F) -> Result<String, Error>
where
    F: AsyncFn() -> Result<String, Error>,
{
    for _ in 0..max {
        if let Ok(val) = f().await {
            return Ok(val);
        }
    }
    f().await
}
}

迁移提示：若代码使用 Fn() -> impl Future<Output = T>， 可考虑改为 AsyncFn() -> T 以获得更清晰的签名。

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::time::{sleep, Duration};

trait Cache {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// --- In-memory implementation ---
struct MemoryCache {
    store: Mutex<HashMap<String, String>>,
}

impl MemoryCache {
    fn new() -> Self {
        MemoryCache {
            store: Mutex::new(HashMap::new()),
        }
    }
}

impl Cache for MemoryCache {
    async fn get(&self, key: &str) -> Option<String> {
        self.store.lock().await.get(key).cloned()
    }

    async fn set(&self, key: &str, value: String) {
        self.store.lock().await.insert(key.to_string(), value);
    }
}

// --- Simulated Redis implementation ---
struct RedisCache {
    store: Mutex<HashMap<String, String>>,
    latency: Duration,
}

impl RedisCache {
    fn new(latency_ms: u64) -> Self {
        RedisCache {
            store: Mutex::new(HashMap::new()),
            latency: Duration::from_millis(latency_ms),
        }
    }
}

impl Cache for RedisCache {
    async fn get(&self, key: &str) -> Option<String> {
        sleep(self.latency).await; // Simulate network round-trip
        self.store.lock().await.get(key).cloned()
    }

    async fn set(&self, key: &str, value: String) {
        sleep(self.latency).await;
        self.store.lock().await.insert(key.to_string(), value);
    }
}

// --- Generic function working with any Cache ---
async fn cache_demo<C: Cache>(cache: &C, label: &str) {
    cache.set("greeting", "Hello, async!".into()).await;
    let val = cache.get("greeting").await;
    println!("[{label}] greeting = {val:?}");
}

#[tokio::main]
async fn main() {
    let mem = MemoryCache::new();
    cache_demo(&mem, "memory").await;

    let redis = RedisCache::new(50);
    cache_demo(&redis, "redis").await;
}

要点回顾 — 异步 Trait

• 自 Rust 1.75 起，可在 Trait 中直接写 async fn（无需 #[async_trait] crate）
• trait_variant::make 自动为 spawn 任务生成 Send 变体（仅静态分发）
• 异步闭包（async Fn()）在 1.85 稳定——用于回调与中间件
• 性能敏感代码优先静态分发（<S: Service>）而非 dyn

另见： 第 13 章 — 生产模式 了解 Tower 的 Service Trait，第 6 章 — 手动构建 Future 了解手动 Trait 实现

11. Stream 与 AsyncIterator 🟡

你将学到：

• Stream Trait（特征）：异步迭代多个值
• 创建 Stream：stream::iter、async_stream、unfold
• Stream 组合子：map、filter、buffer_unordered、fold
• 异步 I/O Trait：AsyncRead、AsyncWrite、AsyncBufRead

Stream Trait 概览

Stream 之于 Iterator，就像 Future 之于单个值——它异步地产生多个值：

#![allow(unused)]
fn main() {
// std::iter::Iterator (synchronous, multiple values)
trait Iterator {
    type Item;
    fn next(&mut self) -> Option<Self::Item>;
}

// futures::Stream (async, multiple values)
trait Stream {
    type Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}
}

graph LR
    subgraph "同步"
        VAL["值<br/>(T)"]
        ITER["迭代器<br/>(多个 T)"]
    end

    subgraph "异步"
        FUT["Future<br/>(异步 T)"]
        STREAM["Stream<br/>(异步多个 T)"]
    end

    VAL -->|"变为异步"| FUT
    ITER -->|"变为异步"| STREAM
    VAL -->|"变为多个"| ITER
    FUT -->|"变为多个"| STREAM

    style VAL fill:#e3f2fd,color:#000
    style ITER fill:#e3f2fd,color:#000
    style FUT fill:#c8e6c9,color:#000
    style STREAM fill:#c8e6c9,color:#000

创建 Stream

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};
use tokio::time::{interval, Duration};
use tokio_stream::wrappers::IntervalStream;

// 1. From an iterator
let s = stream::iter(vec![1, 2, 3]);

// 2. From an async generator (using async_stream crate)
// Cargo.toml: async-stream = "0.3"
use async_stream::stream;

fn countdown(from: u32) -> impl futures::Stream<Item = u32> {
    stream! {
        for i in (0..=from).rev() {
            tokio::time::sleep(Duration::from_millis(500)).await;
            yield i;
        }
    }
}

// 3. From a tokio interval
let tick_stream = IntervalStream::new(interval(Duration::from_secs(1)));

// 4. From a channel receiver (tokio_stream::wrappers)
let (tx, rx) = tokio::sync::mpsc::channel::<String>(100);
let rx_stream = tokio_stream::wrappers::ReceiverStream::new(rx);

// 5. From unfold (generate from async state)
let s = stream::unfold(0u32, |state| async move {
    if state >= 5 {
        None // Stream ends
    } else {
        let next = state + 1;
        Some((state, next)) // yield `state`, new state is `next`
    }
});
}

消费 Stream

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

async fn stream_examples() {
    let s = stream::iter(vec![1, 2, 3, 4, 5]);

    // for_each — process each item
    s.for_each(|x| async move {
        println!("{x}");
    }).await;

    // map + collect
    let doubled: Vec<i32> = stream::iter(vec![1, 2, 3])
        .map(|x| x * 2)
        .collect()
        .await;

    // filter
    let evens: Vec<i32> = stream::iter(1..=10)
        .filter(|x| futures::future::ready(x % 2 == 0))
        .collect()
        .await;

    // buffer_unordered — process N items concurrently
    let results: Vec<_> = stream::iter(vec!["url1", "url2", "url3"])
        .map(|url| async move {
            // Simulate HTTP fetch
            tokio::time::sleep(Duration::from_millis(100)).await;
            format!("response from {url}")
        })
        .buffer_unordered(10) // Up to 10 concurrent fetches
        .collect()
        .await;

    // take, skip, zip, chain — just like Iterator
    let first_three: Vec<i32> = stream::iter(1..=100)
        .take(3)
        .collect()
        .await;
}
}

与 C# IAsyncEnumerable 对比

#![allow(unused)]
fn main() {
// Rust: Stream of database rows
// NOTE: try_stream! (not stream!) is required when using ? inside the body.
// stream! doesn't propagate errors — try_stream! yields Err(e) and ends.
fn get_users(db: &Database) -> impl Stream<Item = Result<User, DbError>> + '_ {
    try_stream! {
        let mut cursor = db.query("SELECT * FROM users").await?;
        while let Some(row) = cursor.next().await {
            yield User::from_row(row?);
        }
    }
}

// Consume:
let mut users = pin!(get_users(&db));
while let Some(result) = users.next().await {
    match result {
        Ok(user) => println!("{}", user.name),
        Err(e) => eprintln!("Error: {e}"),
    }
}
}

// C# equivalent:
async IAsyncEnumerable<User> GetUsers() {
    await using var reader = await db.QueryAsync("SELECT * FROM users");
    while (await reader.ReadAsync()) {
        yield return User.FromRow(reader);
    }
}

// Consume:
await foreach (var user in GetUsers()) {
    Console.WriteLine(user.Name);
}

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

#[derive(Debug)]
struct Stats {
    count: usize,
    min: f64,
    max: f64,
    sum: f64,
}

impl Stats {
    fn average(&self) -> f64 {
        if self.count == 0 { 0.0 } else { self.sum / self.count as f64 }
    }
}

async fn compute_stats<S: futures::Stream<Item = f64>>(stream: S) -> Stats {
    stream
        .fold(
            Stats { count: 0, min: f64::INFINITY, max: f64::NEG_INFINITY, sum: 0.0 },
            |mut acc, value| async move {
                acc.count += 1;
                acc.min = acc.min.min(value);
                acc.max = acc.max.max(value);
                acc.sum += value;
                acc
            },
        )
        .await
}

#[tokio::test]
async fn test_stats() {
    let readings = stream::iter(vec![23.5, 24.1, 22.8, 25.0, 23.9]);
    let stats = compute_stats(readings).await;

    assert_eq!(stats.count, 5);
    assert!((stats.min - 22.8).abs() < f64::EPSILON);
    assert!((stats.max - 25.0).abs() < f64::EPSILON);
    assert!((stats.average() - 23.86).abs() < 0.01);
}
}

异步 I/O Trait：AsyncRead、AsyncWrite、AsyncBufRead

正如 std::io::Read/Write 是同步 I/O 的基础，它们的异步对应物是异步 I/O 的基础。这些 Trait 由 tokio::io 提供（或与运行时无关的代码使用 futures::io）：

#![allow(unused)]
fn main() {
// tokio::io — the async versions of std::io traits

/// Read bytes from a source asynchronously
pub trait AsyncRead {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,  // Tokio's safe wrapper around uninitialized memory
    ) -> Poll<io::Result<()>>;
}

/// Write bytes to a sink asynchronously
pub trait AsyncWrite {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>>;

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
}

/// Buffered reading with line support
pub trait AsyncBufRead: AsyncRead {
    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>>;
    fn consume(self: Pin<&mut Self>, amt: usize);
}
}

实践中，你很少直接调用这些 poll_* 方法。相反，应使用扩展 Trait AsyncReadExt 和 AsyncWriteExt，它们提供支持 .await 的辅助方法：

#![allow(unused)]
fn main() {
use tokio::io::{AsyncReadExt, AsyncWriteExt, AsyncBufReadExt};
use tokio::net::TcpStream;
use tokio::io::BufReader;

async fn io_examples() -> tokio::io::Result<()> {
    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;

    // AsyncWriteExt: write_all, write_u32, write_buf, etc.
    stream.write_all(b"GET / HTTP/1.0\r\n\r\n").await?;

    // AsyncReadExt: read, read_exact, read_to_end, read_to_string
    let mut response = Vec::new();
    stream.read_to_end(&mut response).await?;

    // AsyncBufReadExt: read_line, lines(), split()
    let file = tokio::fs::File::open("config.txt").await?;
    let reader = BufReader::new(file);
    let mut lines = reader.lines();
    while let Some(line) = lines.next_line().await? {
        println!("{line}");
    }

    Ok(())
}
}

实现自定义异步 I/O——在原始 TCP 之上封装协议：

#![allow(unused)]
fn main() {
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use std::pin::Pin;
use std::task::{Context, Poll};

/// A length-prefixed protocol: [u32 length][payload bytes]
struct FramedStream<T> {
    inner: T,
}

impl<T: AsyncRead + AsyncReadExt + Unpin> FramedStream<T> {
    /// Read one complete frame
    async fn read_frame(&mut self) -> tokio::io::Result<Vec<u8>>
    {
        // Read the 4-byte length prefix
        let len = self.inner.read_u32().await? as usize;

        // Read exactly that many bytes
        let mut payload = vec![0u8; len];
        self.inner.read_exact(&mut payload).await?;
        Ok(payload)
    }
}

impl<T: AsyncWrite + AsyncWriteExt + Unpin> FramedStream<T> {
    /// Write one complete frame
    async fn write_frame(&mut self, data: &[u8]) -> tokio::io::Result<()>
    {
        self.inner.write_u32(data.len() as u32).await?;
        self.inner.write_all(data).await?;
        self.inner.flush().await?;
        Ok(())
    }
}
}

tokio 与 futures I/O Trait 对比：它们相似但不完全相同——tokio 的 AsyncRead 使用 ReadBuf（安全处理未初始化内存），而 futures::AsyncRead 使用 &mut [u8]。使用 tokio_util::compat 在两者之间转换。

复制工具：tokio::io::copy(&mut reader, &mut writer) 是 std::io::copy 的异步等价物——适用于代理服务器或文件传输。tokio::io::copy_bidirectional 并发地向两个方向复制。

#![allow(unused)]
fn main() {
use tokio::io::AsyncBufReadExt;

async fn count_non_empty_lines<R: tokio::io::AsyncBufRead + Unpin>(
    reader: R,
) -> tokio::io::Result<usize> {
    let mut lines = reader.lines();
    let mut count = 0;
    while let Some(line) = lines.next_line().await? {
        if !line.is_empty() {
            count += 1;
        }
    }
    Ok(count)
}

// Works with any AsyncBufRead:
// let file = tokio::io::BufReader::new(tokio::fs::File::open("data.txt").await?);
// let count = count_non_empty_lines(file).await?;
//
// let tcp = tokio::io::BufReader::new(TcpStream::connect("...").await?);
// let count = count_non_empty_lines(tcp).await?;
}

要点回顾 — Stream 与 AsyncIterator

• Stream 是 Iterator 的异步等价物——产生 Poll::Ready(Some(item)) 或 Poll::Ready(None)
• .buffer_unordered(N) 并发处理 N 个 Stream 元素——Stream 的关键并发工具
• async_stream::stream! 是创建自定义 Stream 的最简单方式（使用 yield）
• AsyncRead/AsyncBufRead 使 I/O 代码可在文件、套接字和管道之间通用复用

另见： 第 9 章 — 何时 Tokio 并非合适选择 了解 FuturesUnordered（相关模式），第 13 章 — 生产模式 了解有界 channel 的背压

12. 常见陷阱 🔴

你将学到：

• 9 种常见异步 Rust bug 及各自的修复方法
• 为何阻塞执行器（executor）是头号错误（以及 spawn_blocking 如何修复）
• 取消（cancellation）风险：Future 在 .await 中途被丢弃时会发生什么
• 调试：tokio-console、tracing、#[instrument]
• 测试：#[tokio::test]、time::pause()、基于 Trait 的 mock

阻塞执行器

异步 Rust 的头号错误：在异步执行器线程上运行阻塞代码。这会饿死其他任务。

#![allow(unused)]
fn main() {
// ❌ WRONG: Blocks the entire executor thread
async fn bad_handler() -> String {
    let data = std::fs::read_to_string("big_file.txt").unwrap(); // BLOCKS!
    process(&data)
}

// ✅ CORRECT: Offload blocking work to a dedicated thread pool
async fn good_handler() -> String {
    let data = tokio::task::spawn_blocking(|| {
        std::fs::read_to_string("big_file.txt").unwrap()
    }).await.unwrap();
    process(&data)
}

// ✅ ALSO CORRECT: Use tokio's async fs
async fn also_good_handler() -> String {
    let data = tokio::fs::read_to_string("big_file.txt").await.unwrap();
    process(&data)
}
}

graph TB
    subgraph "❌ 在执行器上阻塞调用"
        T1_BAD["线程 1: std::fs::read()<br/>🔴 阻塞 500ms"]
        T2_BAD["线程 2: 处理请求<br/>🟢 独自工作"]
        TASKS_BAD["100 个待处理任务<br/>⏳ 被饿死"]
        T1_BAD -->|"无法 poll"| TASKS_BAD
    end

    subgraph "✅ spawn_blocking"
        T1_GOOD["线程 1: poll Future<br/>🟢 可用"]
        T2_GOOD["线程 2: poll Future<br/>🟢 可用"]
        BT["阻塞池线程:<br/>std::fs::read()<br/>🔵 独立线程池"]
        TASKS_GOOD["100 个任务<br/>✅ 全部推进"]
        T1_GOOD -->|"poll"| TASKS_GOOD
        T2_GOOD -->|"poll"| TASKS_GOOD
    end

std::thread::sleep 与 tokio::time::sleep

#![allow(unused)]
fn main() {
// ❌ WRONG: Blocks the executor thread for 5 seconds
async fn bad_delay() {
    std::thread::sleep(Duration::from_secs(5)); // Thread can't poll anything else!
}

// ✅ CORRECT: Yields to the executor, other tasks can run
async fn good_delay() {
    tokio::time::sleep(Duration::from_secs(5)).await; // Non-blocking!
}
}

在 .await 期间持有 MutexGuard

#![allow(unused)]
fn main() {
use std::sync::Mutex; // std Mutex — NOT async-aware

// ⚠️ RISKY: MutexGuard held across .await
async fn bad_mutex(data: &Mutex<Vec<String>>) {
    let mut guard = data.lock().unwrap();
    guard.push("item".into());
    some_io().await; // Guard is held here — blocks other threads from locking!
    guard.push("another".into());
}
// NOTE: This compiles! std::sync::MutexGuard is !Send, but the compiler only
// enforces Send on the Future when you pass it to something that requires it
// (e.g., tokio::spawn). Calling bad_mutex(...).await directly compiles fine.
// However, tokio::spawn(bad_mutex(data)) will fail with a Send bound error.
}

为何这通常是问题——但并非总是：

在 .await 期间持有 std::sync::Mutex 会在 I/O 持续时间内阻塞 OS 线程，使执行器无法在该线程上 poll 其他任务。对于短临界区这是浪费；对于长 I/O 则是性能陷阱。

然而，有些合法场景下你 必须 在 .await 期间持有锁——就像数据库事务在读取与提交之间持有锁一样。丢弃并重新获取锁会引入 TOCTOU（time-of-check to time-of-use，检查时间与使用时间）竞态：另一个任务可以在你的两个临界区之间修改数据。正确修复取决于具体用例：

#![allow(unused)]
fn main() {
// OPTION 1: Scope the guard — works when operations are independent
async fn scoped_mutex(data: &Mutex<Vec<String>>) {
    {
        let mut guard = data.lock().unwrap();
        guard.push("item".into());
    } // Guard dropped here
    some_io().await; // Lock is released — other tasks can proceed
    {
        let mut guard = data.lock().unwrap();
        guard.push("another".into());
    }
}
// ⚠️ Careful: another task can lock + modify the Vec between the two sections.
//    This is fine if the two pushes are independent, but wrong if "another"
//    depends on state set by "item".

// OPTION 2: Use tokio::sync::Mutex — holds lock across .await without
//           blocking the OS thread. Best when you need transactional
//           read-modify-write across an await point.
use tokio::sync::Mutex as AsyncMutex;

async fn async_mutex(data: &AsyncMutex<Vec<String>>) {
    let mut guard = data.lock().await; // Async lock — doesn't block the thread
    guard.push("item".into());
    some_io().await; // OK — tokio Mutex guard is Send
    guard.push("another".into());
    // Guard held the whole time — no TOCTOU race, no thread blocked.
}
}

何时使用哪种 Mutex：

• std::sync::Mutex：短临界区，内部无 .await
• tokio::sync::Mutex：需要在 .await 点持有锁时（事务语义、避免 TOCTOU）
• parking_lot::Mutex：std 的直接替代，更快更小，仍不可 .await

经验法则：不要盲目在 .await 两侧拆分临界区。问自己两半是否真正独立。若不是——若后半依赖前半的状态——使用 tokio::sync::Mutex 或重新设计数据流。

取消风险

丢弃 Future 会取消它——但这可能使事物处于不一致状态：

#![allow(unused)]
fn main() {
// ❌ DANGEROUS: Resource leak on cancellation
async fn transfer(from: &Account, to: &Account, amount: u64) {
    from.debit(amount).await;  // If cancelled HERE...
    to.credit(amount).await;   // ...money vanishes!
}

// ✅ SAFE: Make operations atomic or use compensation
async fn safe_transfer(from: &Account, to: &Account, amount: u64) -> Result<(), Error> {
    // Use a database transaction (all-or-nothing)
    let tx = db.begin_transaction().await?;
    tx.debit(from, amount).await?;
    tx.credit(to, amount).await?;
    tx.commit().await?; // Only commits if everything succeeded
    Ok(())
}

// ✅ ALSO SAFE: Use tokio::select! with cancellation awareness
tokio::select! {
    result = transfer(from, to, amount) => {
        // Transfer completed
    }
    _ = shutdown_signal() => {
        // Don't cancel mid-transfer — let it finish
        // Or: roll back explicitly
    }
}
}

没有异步 Drop

Rust 的 Drop Trait 是同步的——你 不能 在 drop() 内 .await。这是常见的困惑来源：

#![allow(unused)]
fn main() {
struct DbConnection { /* ... */ }

impl Drop for DbConnection {
    fn drop(&mut self) {
        // ❌ Can't do this — drop() is sync!
        // self.connection.shutdown().await;

        // ✅ Workaround 1: Spawn a cleanup task (fire-and-forget)
        let conn = self.connection.take();
        tokio::spawn(async move {
            let _ = conn.shutdown().await;
        });

        // ✅ Workaround 2: Use a synchronous close
        // self.connection.blocking_close();
    }
}
}

最佳实践：提供显式的 async fn close(self) 方法，并文档说明调用者应使用它。仅将 Drop 作为安全网，而非主要清理路径。

select! 公平性与饥饿

#![allow(unused)]
fn main() {
use tokio::sync::mpsc;

// ❌ UNFAIR: busy_stream always wins, slow_stream starves
async fn unfair(mut fast: mpsc::Receiver<i32>, mut slow: mpsc::Receiver<i32>) {
    loop {
        tokio::select! {
            Some(v) = fast.recv() => println!("fast: {v}"),
            Some(v) = slow.recv() => println!("slow: {v}"),
            // If both are ready, tokio randomly picks one.
            // But if `fast` is ALWAYS ready, `slow` rarely gets polled.
        }
    }
}

// ✅ FAIR: Use biased select or drain in batches
async fn fair(mut fast: mpsc::Receiver<i32>, mut slow: mpsc::Receiver<i32>) {
    loop {
        tokio::select! {
            biased; // Always check in order — explicit priority

            Some(v) = slow.recv() => println!("slow: {v}"),  // Priority!
            Some(v) = fast.recv() => println!("fast: {v}"),
        }
    }
}
}

意外的顺序执行

#![allow(unused)]
fn main() {
// ❌ SEQUENTIAL: Takes 2 seconds total
async fn slow() {
    let a = fetch("url_a").await; // 1 second
    let b = fetch("url_b").await; // 1 second (waits for a to finish first!)
}

// ✅ CONCURRENT: Takes 1 second total
async fn fast() {
    let (a, b) = tokio::join!(
        fetch("url_a"), // Both start immediately
        fetch("url_b"),
    );
}

// ✅ ALSO CONCURRENT: Using let + join
async fn also_fast() {
    let fut_a = fetch("url_a"); // Create future (lazy — not started yet)
    let fut_b = fetch("url_b"); // Create future
    let (a, b) = tokio::join!(fut_a, fut_b); // NOW both run concurrently
}
}

陷阱：let a = fetch(url).await; let b = fetch(url).await; 是顺序执行！ 第二个 .await 要等第一个完成后才开始。并发请用 join! 或 spawn。

案例研究：调试挂起的生产服务

真实场景：服务前 10 分钟正常处理请求，随后停止响应。日志无错误。CPU 为 0%。

诊断步骤：

1. 附加 tokio-console——显示 200+ 个任务卡在 Pending 状态
2. 查看任务详情——全部等待同一个 Mutex::lock().await
3. 根因——一个任务在 .await 期间持有 std::sync::MutexGuard 并 panic，毒化了 mutex。其他任务在 lock().unwrap() 时全部失败

修复：

预防清单：

• 若 guard 跨越任何 .await，使用 tokio::sync::Mutex
• 为异步函数添加 #[tracing::instrument] 以跟踪 span
• 在预发环境运行 tokio-console，及早发现挂起任务
• 添加健康检查端点，验证任务响应性

#![allow(unused)]
fn main() {
use std::sync::Mutex;

async fn process_requests(urls: Vec<String>) -> Vec<String> {
    let results = Mutex::new(Vec::new());
    
    for url in &urls {
        let response = reqwest::get(url).await.unwrap().text().await.unwrap();
        std::thread::sleep(std::time::Duration::from_millis(100)); // Rate limit
        let mut guard = results.lock().unwrap();
        guard.push(response);
        expensive_parse(&guard).await; // Parse all results so far
    }
    
    results.into_inner().unwrap()
}
}

#![allow(unused)]
fn main() {
use tokio::sync::Mutex;
use std::sync::Arc;
use futures::stream::{self, StreamExt};

async fn process_requests(urls: Vec<String>) -> Vec<String> {
    // Fix 4: Process URLs concurrently with buffer_unordered
    let results: Vec<String> = stream::iter(urls)
        .map(|url| async move {
            let response = reqwest::get(&url).await.unwrap().text().await.unwrap();
            // Fix 2: Use tokio::time::sleep instead of std::thread::sleep
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            response
        })
        .buffer_unordered(10) // Up to 10 concurrent requests
        .collect()
        .await;

    // Fix 3: Parse after collecting — no mutex needed at all!
    for result in &results {
        expensive_parse(result).await;
    }

    results
}
}

调试异步代码

异步堆栈跟踪出了名的晦涩——显示的是执行器的 poll 循环，而非你的逻辑调用链。以下是必备调试工具。

tokio-console：实时任务检查器

tokio-console 提供类似 htop 的视图，展示每个已 spawn 的任务：状态、poll 时长、waker 活动与资源使用。

# Cargo.toml
[dependencies]
console-subscriber = "0.4"
tokio = { version = "1", features = ["full", "tracing"] }

#[tokio::main]
async fn main() {
    console_subscriber::init(); // Replaces the default tracing subscriber
    // ... rest of your application
}

然后在另一个终端：

$ RUSTFLAGS="--cfg tokio_unstable" cargo run   # Required compile-time flag
$ tokio-console                                # Connects to 127.0.0.1:6669

tracing + #[instrument]：异步的结构化日志

tracing crate 理解 Future 生命周期（lifetime）。Span 在 .await 点保持打开，即使 OS 线程已切换，也能给出逻辑调用栈：

#![allow(unused)]
fn main() {
use tracing::{info, instrument};

#[instrument(skip(db_pool), fields(user_id = %user_id))]
async fn handle_request(user_id: u64, db_pool: &Pool) -> Result<Response> {
    info!("looking up user");
    let user = db_pool.get_user(user_id).await?;  // span stays open across .await
    info!(email = %user.email, "found user");
    let orders = fetch_orders(user_id).await?;     // still the same span
    Ok(build_response(user, orders))
}
}

输出（使用 tracing_subscriber::fmt::json()）：

{"timestamp":"...","level":"INFO","span":{"name":"handle_request","user_id":"42"},"message":"looking up user"}
{"timestamp":"...","level":"INFO","span":{"name":"handle_request","user_id":"42"},"fields":{"email":"a@b.com"},"message":"found user"}

调试清单

提示：启用 RUSTFLAGS="--cfg tokio_unstable" 以在 tokio-console 中获取任务级指标。这是编译期标志，非运行时标志。

测试异步代码

异步代码带来独特的测试挑战——你需要运行时、时间控制，以及测试并发行为的策略。

基础异步测试 使用 #[tokio::test]：

#![allow(unused)]
fn main() {
// Cargo.toml
// [dev-dependencies]
// tokio = { version = "1", features = ["full", "test-util"] }

#[tokio::test]
async fn test_basic_async() {
    let result = fetch_data().await;
    assert_eq!(result, "expected");
}

// Single-threaded test (useful for !Send types):
#[tokio::test(flavor = "current_thread")]
async fn test_single_threaded() {
    let rc = std::rc::Rc::new(42);
    let val = async { *rc }.await;
    assert_eq!(val, 42);
}

// Multi-threaded with explicit worker count:
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_behavior() {
    // Tests race conditions with real concurrency
    let counter = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
    let c1 = counter.clone();
    let c2 = counter.clone();
    let (a, b) = tokio::join!(
        tokio::spawn(async move { c1.fetch_add(1, std::sync::atomic::Ordering::SeqCst) }),
        tokio::spawn(async move { c2.fetch_add(1, std::sync::atomic::Ordering::SeqCst) }),
    );
    a.unwrap();
    b.unwrap();
    assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 2);
}
}

时间操控——测试超时而无需真实等待：

#![allow(unused)]
fn main() {
use tokio::time::{self, Duration, Instant};

#[tokio::test]
async fn test_timeout_behavior() {
    // Pause time — sleep() advances instantly, no real wall-clock delay
    time::pause();

    let start = Instant::now();
    time::sleep(Duration::from_secs(3600)).await; // "waits" 1 hour — takes 0ms
    assert!(start.elapsed() >= Duration::from_secs(3600));
    // Test ran in milliseconds, not an hour!
}

#[tokio::test]
async fn test_retry_timing() {
    time::pause();

    // Test that our retry logic waits the expected durations
    let start = Instant::now();
    let result = retry_with_backoff(|| async {
        Err::<(), _>("simulated failure")
    }, 3, Duration::from_secs(1))
    .await;

    assert!(result.is_err());
    // 1s + 2s + 4s = 7s of backoff (exponential)
    assert!(start.elapsed() >= Duration::from_secs(7));
}

#[tokio::test]
async fn test_deadline_exceeded() {
    time::pause();

    let result = tokio::time::timeout(
        Duration::from_secs(5),
        async {
            // Simulate slow operation
            time::sleep(Duration::from_secs(10)).await;
            "done"
        }
    ).await;

    assert!(result.is_err()); // Timed out
}
}

Mock 异步依赖——使用 trait 对象或泛型：

#![allow(unused)]
fn main() {
// Define a trait for the dependency:
trait Storage {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// Production implementation:
struct RedisStorage { /* ... */ }
impl Storage for RedisStorage {
    async fn get(&self, key: &str) -> Option<String> {
        // Real Redis call
        todo!()
    }
    async fn set(&self, key: &str, value: String) {
        todo!()
    }
}

// Test mock:
struct MockStorage {
    data: std::sync::Mutex<std::collections::HashMap<String, String>>,
}

impl MockStorage {
    fn new() -> Self {
        MockStorage { data: std::sync::Mutex::new(std::collections::HashMap::new()) }
    }
}

impl Storage for MockStorage {
    async fn get(&self, key: &str) -> Option<String> {
        self.data.lock().unwrap().get(key).cloned()
    }
    async fn set(&self, key: &str, value: String) {
        self.data.lock().unwrap().insert(key.to_string(), value);
    }
}

// Tested function is generic over Storage:
async fn cache_lookup<S: Storage>(store: &S, key: &str) -> String {
    match store.get(key).await {
        Some(val) => val,
        None => {
            let val = "computed".to_string();
            store.set(key, val.clone()).await;
            val
        }
    }
}

#[tokio::test]
async fn test_cache_miss_then_hit() {
    let mock = MockStorage::new();

    // First call: miss → computes and stores
    let val = cache_lookup(&mock, "key1").await;
    assert_eq!(val, "computed");

    // Second call: hit → returns stored value
    let val = cache_lookup(&mock, "key1").await;
    assert_eq!(val, "computed");
    assert!(mock.data.lock().unwrap().contains_key("key1"));
}
}

测试 channel 与任务通信：

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_producer_consumer() {
    let (tx, mut rx) = tokio::sync::mpsc::channel(10);

    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
        }
        // tx dropped here — channel closes
    });

    let mut received = Vec::new();
    while let Some(val) = rx.recv().await {
        received.push(val);
    }

    assert_eq!(received, vec![0, 1, 2, 3, 4]);
}
}

要点回顾 — 常见陷阱

• 永远不要阻塞执行器——CPU/同步工作使用 spawn_blocking
• 永远不要在 .await 期间持有 MutexGuard——严格限定锁作用域或使用 tokio::sync::Mutex
• 取消会立即丢弃 Future——对部分操作使用「可安全取消」模式
• 使用 tokio-console 和 #[tracing::instrument] 调试异步代码
• 用 #[tokio::test] 和 time::pause() 进行确定性时序测试

另见： 第 8 章 — Tokio 深入 了解同步原语，第 13 章 — 生产模式 了解优雅关闭与结构化并发

13. 生产模式 🔴

你将学到：

• 使用 watch channel 和 select! 实现优雅关闭
• 背压（backpressure）：有界 channel 防止 OOM
• 结构化并发（structured concurrency）：JoinSet 与 TaskTracker
• 超时、重试与指数退避
• 错误处理：thiserror 与 anyhow、双重 ? 模式
• Tower：axum、tonic、hyper 使用的中间件模式

优雅关闭

生产服务器必须干净关闭——完成进行中的请求、刷新缓冲区、关闭连接：

use tokio::signal;
use tokio::sync::watch;

async fn main_server() {
    // Create a shutdown signal channel
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Spawn the server
    let server_handle = tokio::spawn(run_server(shutdown_rx.clone()));

    // Wait for Ctrl+C
    signal::ctrl_c().await.expect("Failed to listen for Ctrl+C");
    println!("Shutdown signal received, finishing in-flight requests...");

    // Notify all tasks to shut down
    // NOTE: .unwrap() is used for brevity. Production code should handle
    // the case where all receivers have been dropped.
    shutdown_tx.send(true).unwrap();

    // Wait for server to finish (with timeout)
    match tokio::time::timeout(
        std::time::Duration::from_secs(30),
        server_handle,
    ).await {
        Ok(Ok(())) => println!("Server shut down gracefully"),
        Ok(Err(e)) => eprintln!("Server error: {e}"),
        Err(_) => eprintln!("Server shutdown timed out — forcing exit"),
    }
}

async fn run_server(mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            // Accept new connections
            conn = accept_connection() => {
                let shutdown = shutdown.clone();
                tokio::spawn(handle_connection(conn, shutdown));
            }
            // Shutdown signal
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    println!("Stopping accepting new connections");
                    break;
                }
            }
        }
    }
    // In-flight connections will finish on their own
    // because they have their own shutdown_rx clone
}

async fn handle_connection(conn: Connection, mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            request = conn.next_request() => {
                // Process the request fully — don't abandon mid-request
                process_request(request).await;
            }
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    // Finish current request, then exit
                    break;
                }
            }
        }
    }
}

sequenceDiagram
    participant OS as OS 信号
    participant Main as 主任务
    participant WCH as watch Channel
    participant W1 as Worker 1
    participant W2 as Worker 2

    OS->>Main: SIGINT (Ctrl+C)
    Main->>WCH: send(true)
    WCH-->>W1: changed()
    WCH-->>W2: changed()

    Note over W1: 完成当前请求
    Note over W2: 完成当前请求

    W1-->>Main: 任务完成
    W2-->>Main: 任务完成
    Main->>Main: 所有 worker 完成 → 退出

有界 Channel 的背压

无界 channel 在生产者快于消费者时可能导致 OOM。生产环境应始终使用有界 channel：

#![allow(unused)]
fn main() {
use tokio::sync::mpsc;

async fn backpressure_example() {
    // Bounded channel: max 100 items buffered
    let (tx, mut rx) = mpsc::channel::<WorkItem>(100);

    // Producer: slows down naturally when buffer is full
    let producer = tokio::spawn(async move {
        for i in 0..1_000_000 {
            // send() is async — waits if buffer is full
            // This creates natural backpressure!
            tx.send(WorkItem { id: i }).await.unwrap();
        }
    });

    // Consumer: processes items at its own pace
    let consumer = tokio::spawn(async move {
        while let Some(item) = rx.recv().await {
            process(item).await; // Slow processing is OK — producer waits
        }
    });

    let _ = tokio::join!(producer, consumer);
}

// Compare with unbounded — DANGEROUS:
// let (tx, rx) = mpsc::unbounded_channel(); // No backpressure!
// Producer can fill memory indefinitely
}

结构化并发：JoinSet 与 TaskTracker

JoinSet 将相关任务分组，并确保它们全部完成：

#![allow(unused)]
fn main() {
use tokio::task::JoinSet;
use tokio::time::{sleep, Duration};

async fn structured_concurrency() {
    let mut set = JoinSet::new();

    // Spawn a batch of tasks
    for url in get_urls() {
        set.spawn(async move {
            fetch_and_process(url).await
        });
    }

    // Collect all results (order not guaranteed)
    let mut results = Vec::new();
    while let Some(result) = set.join_next().await {
        match result {
            Ok(Ok(data)) => results.push(data),
            Ok(Err(e)) => eprintln!("Task error: {e}"),
            Err(e) => eprintln!("Task panicked: {e}"),
        }
    }

    // ALL tasks are done here — no dangling background work
    println!("Processed {} items", results.len());
}

// TaskTracker (tokio-util 0.7.9+) — wait for all spawned tasks
use tokio_util::task::TaskTracker;

async fn with_tracker() {
    let tracker = TaskTracker::new();

    for i in 0..10 {
        tracker.spawn(async move {
            sleep(Duration::from_millis(100 * i)).await;
            println!("Task {i} done");
        });
    }

    tracker.close(); // No more tasks will be added
    tracker.wait().await; // Wait for ALL tracked tasks
    println!("All tasks finished");
}
}

超时与重试

#![allow(unused)]
fn main() {
use tokio::time::{timeout, sleep, Duration};

// Simple timeout
async fn with_timeout() -> Result<Response, Error> {
    match timeout(Duration::from_secs(5), fetch_data()).await {
        Ok(Ok(response)) => Ok(response),
        Ok(Err(e)) => Err(Error::Fetch(e)),
        Err(_) => Err(Error::Timeout),
    }
}

// Exponential backoff retry
async fn retry_with_backoff<F, Fut, T, E>(
    max_attempts: u32,
    base_delay_ms: u64,
    operation: F,
) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
    E: std::fmt::Display,
{
    let mut delay = Duration::from_millis(base_delay_ms);

    for attempt in 1..=max_attempts {
        match operation().await {
            Ok(result) => return Ok(result),
            Err(e) => {
                if attempt == max_attempts {
                    eprintln!("Final attempt {attempt} failed: {e}");
                    return Err(e);
                }
                eprintln!("Attempt {attempt} failed: {e}, retrying in {delay:?}");
                sleep(delay).await;
                delay *= 2; // Exponential backoff
            }
        }
    }
    unreachable!()
}

// Usage:
// let result = retry_with_backoff(3, 100, || async {
//     reqwest::get("https://api.example.com/data").await
// }).await?;
}

生产提示 — 添加抖动（jitter）：上述函数使用纯指数退避，但生产中大量客户端同时失败会在相同间隔重试（惊群效应，thundering herd）。添加随机 抖动——例如 sleep(delay + rand_jitter)，其中 rand_jitter 为 0..delay/4——使重试在时间上分散。

异步代码中的错误处理

异步带来独特的错误传播挑战——spawn 的任务形成错误边界，超时错误包装内部错误，? 在 Future 跨越任务边界时行为不同。

thiserror 与 anyhow——选择合适工具：

#![allow(unused)]
fn main() {
// thiserror: Define typed errors for libraries and public APIs
// Every variant is explicit — callers can match on specific errors
use thiserror::Error;

#[derive(Error, Debug)]
enum DiagError {
    #[error("IPMI command failed: {0}")]
    Ipmi(#[from] IpmiError),

    #[error("Sensor {sensor} out of range: {value}°C (max {max}°C)")]
    OverTemp { sensor: String, value: f64, max: f64 },

    #[error("Operation timed out after {0:?}")]
    Timeout(std::time::Duration),

    #[error("Task panicked: {0}")]
    TaskPanic(#[from] tokio::task::JoinError),
}

// anyhow: Quick error handling for applications and prototypes
// Wraps any error — no need to define types for every case
use anyhow::{Context, Result};

async fn run_diagnostics() -> Result<()> {
    let config = load_config()
        .await
        .context("Failed to load diagnostic config")?;  // Adds context

    let result = run_gpu_test(&config)
        .await
        .context("GPU diagnostic failed")?;              // Chains context

    Ok(())
}
// anyhow prints: "GPU diagnostic failed: IPMI command failed: timeout"
}

与 tokio::spawn 的双重 ? 模式：

#![allow(unused)]
fn main() {
use thiserror::Error;
use tokio::task::JoinError;

#[derive(Error, Debug)]
enum AppError {
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),

    #[error("Task panicked: {0}")]
    TaskPanic(#[from] JoinError),
}

async fn spawn_with_errors() -> Result<String, AppError> {
    let handle = tokio::spawn(async {
        let resp = reqwest::get("https://example.com").await?;
        Ok::<_, reqwest::Error>(resp.text().await?)
    });

    // Double ?: First ? unwraps JoinError (task panic), second ? unwraps inner Result
    let result = handle.await??;
    Ok(result)
}
}

错误边界问题——tokio::spawn 擦除上下文：

#![allow(unused)]
fn main() {
// ❌ Error context is lost across spawn boundaries:
async fn bad_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work().await  // Returns Result<T, SomeError>
    });

    // handle.await returns Result<Result<T, SomeError>, JoinError>
    // The inner error has no context about what task failed
    let result = handle.await??;
    Ok(())
}

// ✅ Add context at the spawn boundary:
async fn good_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work()
            .await
            .context("worker task failed")  // Context before crossing boundary
    });

    let result = handle.await
        .context("worker task panicked")??;  // Context for JoinError too
    Ok(())
}
}

超时错误——包装 vs 替换：

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

async fn with_timeout_context() -> Result<String, DiagError> {
    let dur = Duration::from_secs(30);
    match timeout(dur, fetch_sensor_data()).await {
        Ok(Ok(data)) => Ok(data),
        Ok(Err(e)) => Err(e),                      // Inner error preserved
        Err(_) => Err(DiagError::Timeout(dur)),     // Timeout → typed error
    }
}
}

Tower：中间件模式

Tower crate 定义了可组合的 Service Trait——Rust 异步中间件的骨干（被 axum、tonic、hyper 使用）：

#![allow(unused)]
fn main() {
// Tower's core trait (simplified):
pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, req: Request) -> Self::Future;
}
}

中间件包装 Service 以添加横切行为——日志、超时、限流——而无需修改内部逻辑：

#![allow(unused)]
fn main() {
use tower::{ServiceBuilder, timeout::TimeoutLayer, limit::RateLimitLayer};
use std::time::Duration;

let service = ServiceBuilder::new()
    .layer(TimeoutLayer::new(Duration::from_secs(10)))       // Outermost: timeout
    .layer(RateLimitLayer::new(100, Duration::from_secs(1))) // Then: rate limit
    .service(my_handler);                                     // Innermost: your code
}

为何重要：若你用过 ASP.NET 中间件或 Express.js 中间件，Tower 就是 Rust 等价物。生产 Rust 服务借此添加横切关注点而无需重复代码。

练习：带 Worker 池的优雅关闭

use tokio::sync::{mpsc, watch};
use tokio::time::{sleep, Duration};

struct WorkItem { id: u64, payload: String }

#[tokio::main]
async fn main() {
    let (work_tx, work_rx) = mpsc::channel::<WorkItem>(100);
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
    let work_rx = std::sync::Arc::new(tokio::sync::Mutex::new(work_rx));

    let mut handles = Vec::new();
    for id in 0..4 {
        let rx = work_rx.clone();
        let mut shutdown = shutdown_rx.clone();
        handles.push(tokio::spawn(async move {
            loop {
                let item = {
                    let mut rx = rx.lock().await;
                    tokio::select! {
                        item = rx.recv() => item,
                        _ = shutdown.changed() => {
                            if *shutdown.borrow() { None } else { continue }
                        }
                    }
                };
                match item {
                    Some(work) => {
                        println!("Worker {id}: processing {}", work.id);
                        sleep(Duration::from_millis(200)).await;
                    }
                    None => break,
                }
            }
        }));
    }

    // Submit work
    for i in 0..20 {
        let _ = work_tx.send(WorkItem { id: i, payload: format!("task-{i}") }).await;
        sleep(Duration::from_millis(50)).await;
    }

    // On Ctrl+C: signal shutdown, wait for workers
    // NOTE: .unwrap() is used for brevity — handle errors in production.
    tokio::signal::ctrl_c().await.unwrap();
    shutdown_tx.send(true).unwrap();
    for h in handles { let _ = h.await; }
    println!("Shut down cleanly.");
}

要点回顾 — 生产模式

• 使用 watch channel + select! 实现协调的优雅关闭
• 有界 channel（mpsc::channel(N)）提供 背压——缓冲区满时发送方阻塞
• JoinSet 与 TaskTracker 提供 结构化并发：跟踪、中止并等待任务组
• 网络操作始终加超时——tokio::time::timeout(dur, fut)
• Tower 的 Service Trait 是生产 Rust 服务的标准中间件模式

另见： 第 8 章 — Tokio 深入 了解 channel 与同步原语，第 12 章 — 常见陷阱 了解关闭期间的取消风险

14. 异步是优化，而非架构 🔴

你将学到：

• 为何异步往往会污染整个代码库——以及为何这是设计缺陷而非特性
• 「同步核心、异步外壳」模式，使大部分代码可测试、可调试
• 如何处理困难情况：逻辑 同时 需要 I/O
• 何时 spawn_blocking 是修复 vs 是症状
• 何时异步确实应属于核心逻辑
• 为何同步优先的库比异步优先的库更可组合

你已花 13 章学习异步 Rust。本书尚未告诉你的最重要一点：你的大部分代码不应该是异步的。

函数着色问题（Function Coloring Problem）

Bob Nystrom 的 “What Color is Your Function?” 指出了核心问题：异步函数可以调用同步函数，但同步函数不能调用异步函数。一旦某个函数变成异步，调用链上它之上的所有函数都必须跟随。

在 Rust 中这比 C# 或 JavaScript 更严重，因为异步不仅感染函数签名——还感染类型：

每一行都是某人必须做出、做对并维护的决策——且都与业务逻辑无关。行业正在 远离 这种模式：Java 的 Project Loom（虚拟线程）和 Go 的 goroutine 都让你编写看似同步的代码，由运行时廉价地多路复用。Rust 选择显式异步以获得零成本控制，但这种控制有复杂度成本，应有意识地支付，而非默认承担。

「但线程很贵」

条件反射式的反驳：「我们需要异步因为线程很贵。」在大多数团队运作的规模上，这大体是错的。

• 栈内存： 每个 OS 线程保留 8MB 虚拟地址空间（Linux 默认），但 OS 仅在触碰时提交页——大多空闲的线程仅使用 20–80KB 物理内存。
• 上下文切换： 现代硬件上约 1–5µs。50 个并发请求时这是噪声。每秒 10 万次切换时可测量。
• 创建成本： Linux 上每线程约 10–30µs。线程池（rayon、std::thread::scope）将其摊销为零。

异步真正值得其复杂度的诚实阈值大约是 1K–10K 个大多空闲的并发连接——epoll/io_uring 的甜蜜点，每连接栈成为真实成本。低于此，线程池更简单、更易调试、且足够快。高于此，异步胜出。大多数服务低于此阈值。

困难示例：逻辑也需要 I/O

一个平凡的纯函数——fn add(a: i32, b: i32) -> i32——显然不需要异步。这不是有趣的教训。有趣的情况是业务逻辑 似乎 需要在中间做 I/O：验证库存、定价查询汇率、订单流水线查找客户。

考虑订单处理服务。处处异步的版本看起来很自然：

版本 A：核心全程异步

#![allow(unused)]
fn main() {
// orders.rs — async all the way down

pub async fn process_order(order: Order) -> Result<Receipt, OrderError> {
    // Step 1: Validate — pure business rules, no I/O
    validate_items(&order)?;
    validate_quantities(&order)?;

    // Step 2: Check inventory — needs a database call
    let stock = inventory_client.check(&order.items).await?;
    if !stock.all_available() {
        return Err(OrderError::OutOfStock(stock.missing()));
    }

    // Step 3: Calculate pricing — pure math, but async because we're already here
    let pricing = calculate_pricing(&order, &stock);

    // Step 4: Apply discount — needs an external service call
    let discount = discount_service.lookup(order.customer_id).await?;
    let final_price = pricing.apply_discount(discount);

    // Step 5: Format receipt — pure
    Ok(Receipt::new(order, final_price))
}
}

这是 合理 的异步代码。没有滥用 Arc<Mutex>——只是顺序 await。大多数开发者会这样写然后继续。但看看发生了什么：validate_items、validate_quantities、calculate_pricing 和 Receipt::new 都是纯函数，却因第 2、4 步需要 I/O 被拖入异步上下文。整个函数必须是异步的，测试需要运行时，调用链上每个调用者都被着色。

版本 B：同步核心、异步外壳

替代方案：分离 要决定什么 与 如何获取：

#![allow(unused)]
fn main() {
// core.rs — pure business logic, zero async, zero tokio dependency

pub fn validate_order(order: &Order) -> Result<ValidatedOrder, OrderError> {
    validate_items(order)?;
    validate_quantities(order)?;
    Ok(ValidatedOrder::from(order))
}

pub fn check_stock(
    order: &ValidatedOrder,
    stock: &StockResult,
) -> Result<StockedOrder, OrderError> {
    if !stock.all_available() {
        return Err(OrderError::OutOfStock(stock.missing()));
    }
    Ok(StockedOrder::from(order, stock))
}

pub fn finalize(
    order: &StockedOrder,
    discount: Discount,
) -> Receipt {
    let pricing = calculate_pricing(order);
    let final_price = pricing.apply_discount(discount);
    Receipt::new(order, final_price)
}
}

#![allow(unused)]
fn main() {
// shell.rs — thin async orchestrator
//
// Note: the `?` on network calls requires `impl From<reqwest::Error> for OrderError`
// (or a unified error enum). See ch12 for async error handling patterns.

use crate::core;

pub async fn process_order(order: Order) -> Result<Receipt, OrderError> {
    // Sync: validate
    let validated = core::validate_order(&order)?;

    // Async: fetch inventory (this is the shell's job)
    let stock = inventory_client.check(&validated.items).await?;

    // Sync: apply business rule to fetched data
    let stocked = core::check_stock(&validated, &stock)?;

    // Async: fetch discount
    let discount = discount_service.lookup(order.customer_id).await?;

    // Sync: finalize
    Ok(core::finalize(&stocked, discount))
}
}

异步外壳是 获取 → 决策 → 获取 → 决策 的流水线。每个「决策」步骤是同步函数，将 I/O 结果作为输入，而非主动向外请求。

测试差异

同步核心无需运行时或 mock 即可测试每条业务规则：

#![allow(unused)]
fn main() {
#[test]
fn out_of_stock_rejects_order() {
    let order = validated_order(vec![item("widget", 10)]);
    let stock = stock_result(vec![("widget", 3)]); // only 3 available

    let result = core::check_stock(&order, &stock);
    assert_eq!(result.unwrap_err(), OrderError::OutOfStock(vec!["widget"]));
}

#[test]
fn discount_applied_correctly() {
    let order = stocked_order(100_00); // price in cents
    let receipt = core::finalize(&order, Discount::Percent(15));
    assert_eq!(receipt.final_price, 85_00);
}
}

异步外壳得到更薄的 集成 测试，验证接线而非逻辑：

#![allow(unused)]
fn main() {
#[tokio::test]
async fn process_order_integration() {
    let mock_inventory = mock_service(/* returns stock */);
    let mock_discounts = mock_service(/* returns 10% */);
    let receipt = process_order(sample_order()).await.unwrap();
    assert!(receipt.final_price > 0);
    // Logic correctness is already proven by core tests above
}
}

为何重要

关键洞见：第 2、4 步的 I/O 调用不必 位于 业务逻辑内部。它们是逻辑的输入。 同步核心将 StockResult 和 Discount 作为参数。这些值来自 HTTP、gRPC、测试 fixture 还是缓存——是外壳的事。

spawn_blocking 的异味

第 12 章将 spawn_blocking 作为意外阻塞执行器的修复。对于一次性阻塞调用——std::fs::read、压缩库、遗留 FFI 函数——这是正确修复。

但若你发现自己在用 spawn_blocking 包装大段代码：

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // If this is your codebase, the boundary is in the wrong place
    tokio::task::spawn_blocking(move || {
        let validated = validate(&req);       // sync
        let enriched = enrich(validated);      // sync
        let result = process(enriched);        // sync
        let output = format_response(result);  // sync
        output
    }).await.unwrap()
}
}

……代码库在告诉你：这些逻辑本就不是异步的。 你不需要 spawn_blocking——你需要一个同步模块，由异步 handler 直接调用：

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // validate → enrich → process → format are all sync.
    // No spawn_blocking needed — they're fast and CPU-light.
    let response = my_core::handle(req);
    response
}
}

将 spawn_blocking 保留给真正重的 CPU 工作（解析大负载、图像处理、压缩），时间成本会实际饿死执行器。对于微秒级运行的普通业务逻辑，直接同步调用更简单且正确。

库：同步优先，异步包装可选

边界问题对库作者更为关键。同步库可被同步和异步调用方使用：

// A sync library — usable everywhere
let report = my_lib::analyze(&data);

// Caller A: sync CLI
fn main() {
    let report = my_lib::analyze(&data);
    println!("{report}");
}

// Caller B: async handler, works fine
async fn handler() -> Json<Report> {
    let report = my_lib::analyze(&data); // sync call in async context — fine
    Json(report)
}

// Caller C: heavy analysis — caller decides to offload
async fn handler_heavy() -> Json<Report> {
    let data = data.clone();
    let report = tokio::task::spawn_blocking(move || {
        my_lib::analyze(&data) // caller controls the async boundary
    }).await.unwrap();
    Json(report)
}

异步库迫使 所有 调用方进入运行时：

#![allow(unused)]
fn main() {
// An async library — only usable from async contexts
let report = my_lib::analyze(&data).await; // caller MUST be async

// Sync caller? Now you need block_on — and hope there's no nested runtime
let report = tokio::runtime::Runtime::new().unwrap().block_on(
    my_lib::analyze(&data)
); // fragile, panic-prone if already inside a runtime
}

默认使用同步 API。 若库做纯计算、数据转换或解析，没有理由异步。若做 I/O，考虑提供同步核心，并在 feature flag 后提供可选异步便利层——让调用方拥有边界决策。

何时异步属于核心

并非一切都能干净分离。以下情况异步应属于核心逻辑：

• 扇出/扇入就是逻辑。 若业务规则是「并发查询 5 个定价服务并返回最便宜」，并发 就是 逻辑，而非管道。强行用同步 + 线程是在发明更差的异步。

• 流式处理就是逻辑。 处理带背压的连续事件流——流管理是非平凡的业务逻辑，不只是 I/O 包装。

• 长生命周期有状态连接。 WebSocket handler、gRPC 双向流和协议状态机，状态转换与 I/O 事件固有绑定。第 17 章 的毕业项目——异步聊天服务器——正是此例：并发连接、基于房间扇出和优雅关闭本质上是异步工作。

检验： 若从函数中移除 async 需要用线程、channel 或手动 poll 替代，则异步物有所值。若移除 async 只需删掉关键字而无其他变化，则它本不需要异步。

决策规则

graph TD
    START["此函数是否应为异步？"] --> IO{"是否做 I/O？"}
    IO -->|否| SYNC["sync fn — 始终"]
    IO -->|是| BOUNDARY{"是否在边界？<br/>handler、主循环、accept()"}
    BOUNDARY -->|是| ASYNC_SHELL["async fn — 这是外壳"]
    BOUNDARY -->|否| CORE_IO{"I/O 是否为核心逻辑？<br/>扇出、流式、有状态连接"}
    CORE_IO -->|是| ASYNC_CORE["async fn — 有理由"]
    CORE_IO -->|否| EXTRACT["将逻辑提取为 sync fn。<br/>将 I/O 结果作为参数传入。"]

    style SYNC fill:#d4efdf,stroke:#27ae60,color:#000
    style ASYNC_SHELL fill:#e8f4f8,stroke:#2980b9,color:#000
    style ASYNC_CORE fill:#e8f4f8,stroke:#2980b9,color:#000
    style EXTRACT fill:#d4efdf,stroke:#27ae60,color:#000

经验法则： 从同步开始。仅在最外层 I/O 边界添加异步。仅当你能阐明 哪些并发 I/O 操作 值得复杂度代价时，才向内推进。

#![allow(unused)]
fn main() {
use axum::{Json, extract::Path};

async fn get_device_report(Path(device_id): Path<String>) -> Result<Json<Report>, AppError> {
    // Fetch raw telemetry from the device over HTTP
    let raw = reqwest::get(format!("http://bmc-{device_id}/telemetry"))
        .await?
        .json::<RawTelemetry>()
        .await?;

    // Business logic: convert raw sensor readings to calibrated values
    let mut readings = Vec::new();
    for sensor in &raw.sensors {
        let calibrated = (sensor.raw_value as f64) * sensor.scale + sensor.offset;
        if calibrated < sensor.min_valid || calibrated > sensor.max_valid {
            return Err(AppError::SensorOutOfRange {
                name: sensor.name.clone(),
                value: calibrated,
            });
        }
        readings.push(CalibratedReading {
            name: sensor.name.clone(),
            value: calibrated,
            unit: sensor.unit.clone(),
        });
    }

    // Business logic: classify device health
    let critical_count = readings.iter()
        .filter(|r| r.value > 90.0)
        .count();
    let health = if critical_count > 2 { Health::Critical }
                 else if critical_count > 0 { Health::Warning }
                 else { Health::Ok };

    // Fetch device metadata from inventory service
    let meta = reqwest::get(format!("http://inventory/devices/{device_id}"))
        .await?
        .json::<DeviceMetadata>()
        .await?;

    Ok(Json(Report {
        device_id,
        device_name: meta.name,
        health,
        readings,
        timestamp: chrono::Utc::now(),
    }))
}
}

#![allow(unused)]
fn main() {
// core.rs — zero async dependency

pub fn calibrate_sensors(raw: &RawTelemetry) -> Result<Vec<CalibratedReading>, AppError> {
    raw.sensors.iter().map(|sensor| {
        let calibrated = (sensor.raw_value as f64) * sensor.scale + sensor.offset;
        if calibrated < sensor.min_valid || calibrated > sensor.max_valid {
            return Err(AppError::SensorOutOfRange {
                name: sensor.name.clone(),
                value: calibrated,
            });
        }
        Ok(CalibratedReading {
            name: sensor.name.clone(),
            value: calibrated,
            unit: sensor.unit.clone(),
        })
    }).collect()
}

pub fn classify_health(readings: &[CalibratedReading]) -> Health {
    let critical_count = readings.iter()
        .filter(|r| r.value > 90.0)
        .count();
    if critical_count > 2 { Health::Critical }
    else if critical_count > 0 { Health::Warning }
    else { Health::Ok }
}

pub fn build_report(
    device_id: String,
    readings: Vec<CalibratedReading>,
    meta: &DeviceMetadata,
) -> Report {
    Report {
        device_id,
        device_name: meta.name.clone(),
        health: classify_health(&readings),
        readings,
        timestamp: chrono::Utc::now(),
    }
}
}

#![allow(unused)]
fn main() {
// shell.rs — async boundary only

pub async fn get_device_report(
    Path(device_id): Path<String>,
) -> Result<Json<Report>, AppError> {
    let raw = reqwest::get(format!("http://bmc-{device_id}/telemetry"))
        .await?
        .json::<RawTelemetry>()
        .await?;

    let readings = core::calibrate_sensors(&raw)?;

    let meta = reqwest::get(format!("http://inventory/devices/{device_id}"))
        .await?
        .json::<DeviceMetadata>()
        .await?;

    Ok(Json(core::build_report(device_id, readings, &meta)))
}
}

#![allow(unused)]
fn main() {
// core_tests.rs — no runtime needed

// Test fixture helpers — construct data without any I/O
fn sensor(name: &str, raw_value: f64, valid_range: std::ops::Range<f64>) -> RawSensor {
    RawSensor {
        name: name.into(),
        raw_value,
        scale: 1.0,
        offset: 0.0,
        min_valid: valid_range.start,
        max_valid: valid_range.end,
        unit: "unit".into(),
    }
}

fn raw_telemetry(sensors: Vec<RawSensor>) -> RawTelemetry {
    RawTelemetry { sensors }
}

fn reading(name: &str, value: f64) -> CalibratedReading {
    CalibratedReading { name: name.into(), value, unit: "unit".into() }
}

fn device_meta(name: &str) -> DeviceMetadata {
    DeviceMetadata { name: name.into() }
}

#[test]
fn sensor_out_of_range_rejected() {
    let raw = raw_telemetry(vec![sensor("gpu_temp", 105.0, 0.0..100.0)]);
    let result = core::calibrate_sensors(&raw);
    assert!(matches!(result, Err(AppError::SensorOutOfRange { .. })));
}

#[test]
fn health_classification() {
    let readings = vec![
        reading("a", 50.0),  // ok
        reading("b", 95.0),  // critical
        reading("c", 91.0),  // critical
        reading("d", 92.0),  // critical
    ];
    assert_eq!(core::classify_health(&readings), Health::Critical);
}

#[test]
fn normal_report() {
    let raw = raw_telemetry(vec![sensor("fan_rpm", 3000.0, 0.0..10000.0)]);
    let readings = core::calibrate_sensors(&raw).unwrap();
    let meta = device_meta("gpu-node-42");
    let report = core::build_report("dev-1".into(), readings, &meta);
    assert_eq!(report.health, Health::Ok);
    assert_eq!(report.readings.len(), 1);
}
}

要点回顾：

1. 异步是 I/O 多路复用优化，而非应用架构。大多数业务逻辑是同步的。
2. 同步核心、异步外壳： 将业务规则放在纯同步函数中，以 I/O 结果作为参数。异步外壳编排抓取并调用核心。
3. 若用大段 spawn_blocking 包装，边界放错了——将逻辑重构为同步模块。
4. 库应默认同步 API。 异步库迫使所有调用方进入运行时；同步库让调用方拥有异步边界。
5. 异步在 扇出/扇入、流式处理和有状态连接 上物有所值——并发 就是 业务逻辑的情况。

另见： 第 12 章 — 常见陷阱（spawn_blocking 作为战术修复）· 第 13 章 — 生产模式（背压、结构化并发）· 第 17 章 — 毕业项目：异步聊天服务器（异步是正确架构的案例）

15. 练习

练习

练习 1：异步 Echo 服务器

构建一个并发处理多个客户端的 TCP echo 服务器。

要求：

• 监听 127.0.0.1:8080
• 接受连接并回显每一行
• 优雅处理客户端断开
• 客户端连接/断开时打印日志

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Echo server listening on :8080");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("[{addr}] Connected");

        tokio::spawn(async move {
            let (reader, mut writer) = socket.into_split();
            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                line.clear();
                match reader.read_line(&mut line).await {
                    Ok(0) => {
                        println!("[{addr}] Disconnected");
                        break;
                    }
                    Ok(_) => {
                        print!("[{addr}] Echo: {line}");
                        if writer.write_all(line.as_bytes()).await.is_err() {
                            println!("[{addr}] Write error, disconnecting");
                            break;
                        }
                    }
                    Err(e) => {
                        eprintln!("[{addr}] Read error: {e}");
                        break;
                    }
                }
            }
        });
    }
}

练习 2：带限流的并发 URL 抓取器

并发抓取 URL 列表，最多 5 个并发请求。

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

async fn fetch_urls(urls: Vec<String>) -> Vec<Result<String, String>> {
    // buffer_unordered(5) ensures at most 5 futures are polled
    // concurrently — no separate Semaphore needed here.
    let results: Vec<_> = stream::iter(urls)
        .map(|url| {
            async move {
                println!("Fetching: {url}");

                match reqwest::get(&url).await {
                    Ok(resp) => match resp.text().await {
                        Ok(body) => Ok(body),
                        Err(e) => Err(format!("{url}: {e}")),
                    },
                    Err(e) => Err(format!("{url}: {e}")),
                }
            }
        })
        .buffer_unordered(5) // ← This alone limits concurrency to 5
        .collect()
        .await;

    results
}

// NOTE: Use Semaphore when you need to limit concurrency across
// independently spawned tasks (tokio::spawn). Use buffer_unordered
// when processing a stream. Don't combine both for the same limit.
}

练习 3：带 Worker 池的优雅关闭

构建任务处理器，包含：

• 基于 channel 的工作队列
• N 个 worker 任务从队列消费
• Ctrl+C 优雅关闭：停止接受新任务，完成进行中的工作

use tokio::sync::{mpsc, watch};
use tokio::time::{sleep, Duration};

struct WorkItem {
    id: u64,
    payload: String,
}

#[tokio::main]
async fn main() {
    let (work_tx, work_rx) = mpsc::channel::<WorkItem>(100);
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Spawn 4 workers
    let mut worker_handles = Vec::new();
    let work_rx = std::sync::Arc::new(tokio::sync::Mutex::new(work_rx));

    for id in 0..4 {
        let rx = work_rx.clone();
        let mut shutdown = shutdown_rx.clone();
        let handle = tokio::spawn(async move {
            loop {
                let item = {
                    let mut rx = rx.lock().await;
                    tokio::select! {
                        item = rx.recv() => item,
                        _ = shutdown.changed() => {
                            if *shutdown.borrow() { None } else { continue }
                        }
                    }
                };

                match item {
                    Some(work) => {
                        println!("Worker {id}: processing item {}", work.id);
                        sleep(Duration::from_millis(200)).await; // Simulate work
                        println!("Worker {id}: done with item {}", work.id);
                    }
                    None => {
                        println!("Worker {id}: channel closed, exiting");
                        break;
                    }
                }
            }
        });
        worker_handles.push(handle);
    }

    // Producer: submit some work
    let producer = tokio::spawn(async move {
        for i in 0..20 {
            let _ = work_tx.send(WorkItem {
                id: i,
                payload: format!("task-{i}"),
            }).await;
            sleep(Duration::from_millis(50)).await;
        }
    });

    // Wait for Ctrl+C
    tokio::signal::ctrl_c().await.unwrap();
    println!("\nShutdown signal received!");
    shutdown_tx.send(true).unwrap();
    producer.abort(); // Cancel the producer task

    // Wait for workers to finish
    for handle in worker_handles {
        let _ = handle.await;
    }
    println!("All workers shut down. Goodbye!");
}

练习 4：从零实现简单异步 Mutex

使用 channel 实现异步感知的 mutex（不使用 tokio::sync::Mutex）。

提示：使用许可数为 1 的 tokio::sync::Semaphore 来串行化访问。

#![allow(unused)]
fn main() {
use std::cell::UnsafeCell;
use std::sync::Arc;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

pub struct SimpleAsyncMutex<T> {
    data: Arc<UnsafeCell<T>>,
    semaphore: Arc<Semaphore>,
}

// SAFETY: Access to T is serialized by the semaphore (max 1 permit).
unsafe impl<T: Send> Send for SimpleAsyncMutex<T> {}
unsafe impl<T: Send> Sync for SimpleAsyncMutex<T> {}

pub struct SimpleGuard<T> {
    data: Arc<UnsafeCell<T>>,
    _permit: OwnedSemaphorePermit, // Dropped on guard drop → releases lock
}

impl<T> SimpleAsyncMutex<T> {
    pub fn new(value: T) -> Self {
        SimpleAsyncMutex {
            data: Arc::new(UnsafeCell::new(value)),
            semaphore: Arc::new(Semaphore::new(1)),
        }
    }

    pub async fn lock(&self) -> SimpleGuard<T> {
        let permit = self.semaphore.clone().acquire_owned().await.unwrap();
        SimpleGuard {
            data: self.data.clone(),
            _permit: permit,
        }
    }
}

impl<T> std::ops::Deref for SimpleGuard<T> {
    type Target = T;
    fn deref(&self) -> &T {
        // SAFETY: We hold the only semaphore permit, so no other
        // SimpleGuard exists → exclusive access is guaranteed.
        unsafe { &*self.data.get() }
    }
}

impl<T> std::ops::DerefMut for SimpleGuard<T> {
    fn deref_mut(&mut self) -> &mut T {
        // SAFETY: Same reasoning — single permit guarantees exclusivity.
        unsafe { &mut *self.data.get() }
    }
}

// When SimpleGuard is dropped, _permit is dropped,
// which releases the semaphore permit — another lock() can proceed.

// Usage:
// let mutex = SimpleAsyncMutex::new(vec![1, 2, 3]);
// {
//     let mut guard = mutex.lock().await;
//     guard.push(4);
// } // permit released here
}

练习 5：Stream 流水线

使用 stream 构建数据处理流水线：

1. 生成数字 1..=100
2. 过滤为偶数
3. 映射为平方
4. 每次并发处理 10 个（用 sleep 模拟）
5. 收集结果

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let results: Vec<u64> = stream::iter(1u64..=100)
        // Step 2: Filter evens
        .filter(|x| futures::future::ready(x % 2 == 0))
        // Step 3: Square each
        .map(|x| x * x)
        // Step 4: Process concurrently (simulate async work)
        .map(|x| async move {
            sleep(Duration::from_millis(50)).await;
            println!("Processed: {x}");
            x
        })
        .buffer_unordered(10) // 10 concurrent
        // Step 5: Collect
        .collect()
        .await;

    println!("Got {} results", results.len());
    println!("Sum: {}", results.iter().sum::<u64>());
}

练习 6：实现带超时的 Select

不使用 tokio::select! 或 tokio::time::timeout，实现一个函数，让 Future 与截止时间竞速，超时时返回 Either::Left(result) 或 Either::Right(())。

提示：基于第 6 章的 Select 组合子与同期的 TimerFuture。

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

pub enum Either<A, B> {
    Left(A),
    Right(B),
}

pub struct Timeout<F> {
    future: F,
    timer: TimerFuture, // From Chapter 6
}

impl<F: Future + Unpin> Timeout<F> {
    pub fn new(future: F, duration: Duration) -> Self {
        Timeout {
            future,
            timer: TimerFuture::new(duration),
        }
    }
}

impl<F: Future + Unpin> Future for Timeout<F> {
    type Output = Either<F::Output, ()>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Check if the main future is done
        if let Poll::Ready(val) = Pin::new(&mut self.future).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

        // Check if the timer expired
        if let Poll::Ready(()) = Pin::new(&mut self.timer).poll(cx) {
            return Poll::Ready(Either::Right(()));
        }

        Poll::Pending
    }
}

// Usage:
// match Timeout::new(fetch_data(), Duration::from_secs(5)).await {
//     Either::Left(data) => println!("Got data: {data}"),
//     Either::Right(()) => println!("Timed out!"),
// }

总结与参考卡

快速参考卡

异步心智模型

┌─────────────────────────────────────────────────────┐
│  async fn → State Machine (enum) → impl Future     │
│  .await   → poll() the inner future                 │
│  executor → loop { poll(); sleep_until_woken(); }   │
│  waker    → "hey executor, poll me again"           │
│  Pin      → "promise I won't move in memory"        │
└─────────────────────────────────────────────────────┘

常见模式速查表

Pin 快速参考

Channel 选择指南

Mutex 选择指南

决策快速参考

需要并发？
├── I/O 密集型 → async/await
├── CPU 密集型 → rayon / std::thread
└── 混合型 → CPU 部分用 spawn_blocking

选择运行时？
├── 服务器应用 → tokio
├── 库 → 与运行时无关（futures crate）
├── 嵌入式 → embassy
└── 极简 → smol

需要并发 Future？
├── 可以是 'static + Send → tokio::spawn
├── 可以是 'static + !Send → LocalSet
├── 不能是 'static → FuturesUnordered
└── 需要跟踪/中止 → JoinSet

常见错误信息与修复

延伸阅读

异步 Rust 培训指南 完

毕业项目：异步聊天服务器

本项目将全书模式整合为单一的生产风格应用。你将使用 tokio、channel、stream、优雅关闭和正确的错误处理，构建一个 多房间异步聊天服务器。

预计时间：4–6 小时 | 难度：★★★

你将练习：

• tokio::spawn 与 'static 要求（第 8 章）
• Channel：消息用 mpsc、房间用 broadcast、关闭用 watch（第 8 章）
• Stream：从 TCP 连接读取行（第 11 章）
• 常见陷阱：取消安全、在 .await 期间持有 MutexGuard（第 12 章）
• 生产模式：优雅关闭、背压（第 13 章）
• 可插拔后端的异步 Trait（第 10 章）

问题描述

构建 TCP 聊天服务器，要求：

1. 客户端 通过 TCP 连接并加入命名房间
2. 消息 广播给同一房间的所有客户端
3. 命令：/join <room>、/nick <name>、/rooms、/quit
4. 服务器在 Ctrl+C 时优雅关闭——完成进行中的消息

graph LR
    C1["客户端 1<br/>(Alice)"] -->|TCP| SERVER["聊天服务器"]
    C2["客户端 2<br/>(Bob)"] -->|TCP| SERVER
    C3["客户端 3<br/>(Carol)"] -->|TCP| SERVER

    SERVER --> R1["#general<br/>broadcast channel"]
    SERVER --> R2["#rust<br/>broadcast channel"]

    R1 -->|msg| C1
    R1 -->|msg| C2
    R2 -->|msg| C3

    CTRL["Ctrl+C"] -->|watch| SERVER

    style SERVER fill:#e8f4f8,stroke:#2980b9,color:#000
    style R1 fill:#d4efdf,stroke:#27ae60,color:#000
    style R2 fill:#d4efdf,stroke:#27ae60,color:#000
    style CTRL fill:#fadbd8,stroke:#e74c3c,color:#000

步骤 1：基础 TCP Accept 循环

从接受连接并回显行的服务器开始：

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Chat server listening on :8080");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("[{addr}] Connected");

        tokio::spawn(async move {
            let (reader, mut writer) = socket.into_split();
            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                line.clear();
                match reader.read_line(&mut line).await {
                    Ok(0) | Err(_) => break,
                    Ok(_) => {
                        let _ = writer.write_all(line.as_bytes()).await;
                    }
                }
            }
            println!("[{addr}] Disconnected");
        });
    }
}

你的任务：验证可编译，并用 telnet localhost 8080 测试。

步骤 2：用 Broadcast Channel 管理房间状态

每个房间是一个 broadcast::Sender。房间内所有客户端订阅以接收消息。

#![allow(unused)]
fn main() {
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};

type RoomMap = Arc<RwLock<HashMap<String, broadcast::Sender<String>>>>;

fn get_or_create_room(rooms: &mut HashMap<String, broadcast::Sender<String>>, name: &str) -> broadcast::Sender<String> {
    rooms.entry(name.to_string())
        .or_insert_with(|| {
            let (tx, _) = broadcast::channel(100); // 100-message buffer
            tx
        })
        .clone()
}
}

你的任务：实现房间状态，使得：

• 客户端默认在 #general
• /join <room> 切换房间（取消订阅旧房间，订阅新房间）
• 消息广播给发送者当前房间的所有客户端

#![allow(unused)]
fn main() {
loop {
    tokio::select! {
        // Client sent us a line
        result = reader.read_line(&mut line) => {
            match result {
                Ok(0) | Err(_) => break,
                Ok(_) => {
                    // Parse command or broadcast message
                }
            }
        }
        // Room broadcast received
        result = room_rx.recv() => {
            match result {
                Ok(msg) => {
                    let _ = writer.write_all(msg.as_bytes()).await;
                }
                Err(_) => break,
            }
        }
    }
}
}

步骤 3：命令

实现命令协议：

你的任务：从输入行解析命令。对于 /rooms，需从 RoomMap 读取——使用 RwLock::read() 避免阻塞其他客户端。

步骤 4：优雅关闭

添加 Ctrl+C 处理，使服务器：

1. 停止接受新连接
2. 向所有房间发送「Server shutting down…」
3. 等待进行中的消息排空
4. 干净退出

#![allow(unused)]
fn main() {
use tokio::sync::watch;

let (shutdown_tx, shutdown_rx) = watch::channel(false);

// In the accept loop:
loop {
    tokio::select! {
        result = listener.accept() => {
            let (socket, addr) = result?;
            // spawn client task with shutdown_rx.clone()
        }
        _ = tokio::signal::ctrl_c() => {
            println!("Shutdown signal received");
            shutdown_tx.send(true)?;
            break;
        }
    }
}
}

你的任务：在每个客户端的 select! 循环中添加 shutdown_rx.changed()，以便收到关闭信号时客户端退出。

步骤 5：错误处理与边界情况

生产级加固服务器：

1. 滞后接收者（lagging receivers）：若慢客户端错过消息，broadcast::recv() 返回 RecvError::Lagged(n)。优雅处理（记录日志并继续，不要崩溃）。
2. 昵称验证：拒绝空或过长昵称。
3. 背压：broadcast channel 缓冲区有界（100）。若客户端跟不上，会收到 Lagged 错误。
4. 超时：空闲超过 5 分钟的客户端断开连接。

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

// Wrap the read in a timeout:
match timeout(Duration::from_secs(300), reader.read_line(&mut line)).await {
    Ok(Ok(0)) | Ok(Err(_)) | Err(_) => break, // EOF, error, or timeout
    Ok(Ok(_)) => { /* process line */ }
}
}

步骤 6：集成测试

编写测试：启动服务器、连接两个客户端、验证消息投递：

#![allow(unused)]
fn main() {
#[tokio::test]
async fn two_clients_can_chat() {
    // Start server in background
    let server = tokio::spawn(run_server("127.0.0.1:0")); // Port 0 = OS picks

    // Connect two clients
    let mut client1 = TcpStream::connect(addr).await.unwrap();
    let mut client2 = TcpStream::connect(addr).await.unwrap();

    // Client 1 sends a message
    client1.write_all(b"Hello from client 1\n").await.unwrap();

    // Client 2 should receive it
    let mut buf = vec![0u8; 1024];
    let n = client2.read(&mut buf).await.unwrap();
    let msg = String::from_utf8_lossy(&buf[..n]);
    assert!(msg.contains("Hello from client 1"));
}
}

评估标准

扩展想法

基础聊天服务器完成后，可尝试：

1. 持久历史：每房间存储最近 N 条消息；新加入者回放
2. WebSocket 支持：使用 tokio-tungstenite 同时接受 TCP 和 WebSocket 客户端
3. 限流：使用 tokio::time::Interval 限制每客户端每秒消息数
4. 指标：通过 prometheus crate 跟踪连接客户端数、消息/秒、房间数
5. TLS：添加 tokio-rustls 实现加密连接