Async Rust：从 Future 到生产级应用

作者介绍

• 微软 SCHIE（硅片与云硬件基础设施工程）团队首席固件架构师
• 行业资深专家，专长于安全、系统编程（固件、操作系统、虚拟机监控器）、CPU 和平台架构以及 C++ 系统
• 2017 年开始使用 Rust 编程（@AWS EC2），从此爱上了这门语言

这是一本深入讲解 Rust 异步编程的指南。与大多数从 tokio::main 开始、对内部原理一笔带过的异步教程不同，本指南从第一性原理构建理解——Future trait、polling、状态机——然后逐步深入到实际模式、runtime 选择和生产环境中的陷阱。

适合读者

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

前置要求

你应该熟悉以下内容：

• 所有权、借用和生命周期
• Trait 和泛型（包括 impl Trait）
• 使用 Result<T, E> 和 ? 操作符
• 基础多线程（std::thread::spawn、Arc、Mutex）

不需要有 Rust 异步编程经验。

如何使用本书

第一遍请线性阅读。 第一部分到第三部分的内容循序渐进。每章包含：

每章还包括：

• 顶部的 “你将学到” 模块
• Mermaid 图表 帮助视觉学习者理解
• 内联练习 附带隐藏解答
• 关键要点 总结核心思想
• 交叉引用 指向相关章节

学习进度指南

总预估时间：22–30 小时

完成练习

每章内容都有内联练习。综合项目（第 16 章）将所有内容整合为一个完整项目。为了最大化学习效果：

1. 在查看答案前先尝试完成练习 — 挣扎是学习发生的地方
2. 动手敲代码，不要复制粘贴 — 肌肉记忆对 Rust 语法很重要
3. 运行每个示例 — cargo new async-exercises 并随时测试

目录

第一部分：异步工作原理

• 1. 为什么 Rust 的异步不同 🟢 — 根本区别：Rust 没有内置 runtime
• 2. The Future Trait 🟡 — poll()、Waker 和让它工作的契约
• 3. Poll 工作原理 🟡 — polling 状态机和最小化 executor
• 4. Pin 和 Unpin 🔴 — 为什么自引用 struct 需要 pinning
• 5. 状态机揭秘 🟢 — 编译器从 async fn 实际生成的代码

第二部分：生态系统

• 6. 手动构建 Futures 🟡 — 从零开始实现 TimerFuture、Join、Select
• 7. Executors 和 Runtimes 🟡 — tokio、smol、async-std、embassy — 如何选择
• 8. Tokio 深入 🟡 — Runtime 类型、spawn、channels、同步原语
• 9. 何时 Tokio 不是最佳选择 🟡 — LocalSet、FuturesUnordered、runtime 无关设计
• 10. Async Traits 🟡 — RPITIT、dyn 分发、trait_variant、async 闭包

第三部分：生产级异步

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

附录

• 总结与速查表 — 快速查阅表和决策树
• 综合项目：异步聊天服务器 — 构建完整的异步应用

1. 为什么 Rust 的异步不同 🟢

你将学到：

• 为什么 Rust 没有内置的异步 runtime（这对你意味着什么）
• 三个关键特性：惰性执行、无 runtime、零成本抽象
• 何时异步是正确的工具（何时更慢）
• Rust 的模型与 C#、Go、Python 和 JavaScript 的对比

根本区别

大多数带有 async/await 的语言隐藏了内部机制。C# 有 CLR 线程池。JavaScript 有事件循环。Go 有 goroutines 和内置于 runtime 的调度器。Python 有 asyncio。

Rust 什么都没有。

没有内置 runtime，没有线程池，没有事件循环。async 关键字是一个零成本的编译策略——它将你的函数转换为一个实现 Future trait 的状态机。必须由其他人（一个 executor）来驱动这个状态机向前执行。

Rust 异步的三个关键特性

graph LR
    subgraph "C# / JS / Go"
        EAGER["Eager Execution<br/>任务立即开始"]
        BUILTIN["内置 Runtime<br/>包含线程池"]
        GC["GC 管理<br/>无生命周期问题"]
    end

    subgraph "Rust（和 Python*）"
        LAZY["Lazy Execution<br/>直到被 poll/await 才执行"]
        BYOB["自带 Runtime<br/>你选择 executor"]
        OWNED["所有权适用<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 一样是惰性的——它们不会被执行直到被 awaited 或调度。但是，Python 仍然使用 GC 且没有所有权/生命周期的顾虑。

没有内置 Runtime

// 这段代码编译通过但什么都不做：
async fn fetch_data() -> String {
    "hello".to_string()
}

fn main() {
    let future = fetch_data(); // 创建 Future，但不执行它
    // future 只是一个躺在栈上的 struct
    // 没有输出，没有副作用，什么都没发生
    drop(future); // 静默丢弃——工作从未开始
}

与 C# 对比，Task 是立即开始执行的：

// C# —— 这立即开始执行：
async Task<string> FetchData() => "hello";

var task = FetchData(); // 已经在运行！
var result = await task; // 只是等待完成

Lazy Futures vs Eager Tasks

这是最重要的思维转变：

// 要真正运行 future，需要一个 executor：
#[tokio::main]
async fn main() {
    let result = fetch_data().await; // 现在才执行
    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 个网络连接，异步表现出色。如果你在 crunch 数字，使用 rayon 或 OS 线程。

何时异步可能更慢

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

基准测试指导：如果少于约 10 个并发 I/O 操作，在承诺使用异步之前先 profile。简单的每个连接一个 std::thread::spawn 在现代 Linux 上可以很好地扩展到数百个线程。

练习：何时使用异步？

关键要点——为什么异步不同

• Rust future 是惰性的——在被 executor poll 之前什么都不做
• 没有内置 runtime——你选择（或构建）自己的 runtime
• 异步是产生状态机的零成本编译策略
• 异步在I/O 绑定并发表现出色；对于 CPU 绑定工作，使用线程或 rayon

另见： 第 2 章 — The Future Trait 了解实现这一切的 trait，第 7 章 — Executors 和 Runtimes 了解如何选择 runtime

2. The Future Trait 🟡

你将学到：

• Future trait：Output、poll()、Context、Waker
• Waker 如何告诉 executor “再次 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),   // Future 已完成，返回值 T
    Pending,    // Future 还未就绪——稍后再次调用我
}
}

就是这样。Future 是任何可以被 poll 的东西——被问“你完成了吗？“——然后回答“是的，这是结果“或“还没，我会在准备好时唤醒你。”

Output、poll()、Context、Waker

sequenceDiagram
    participant E as Executor
    participant F as Future (Task)
    participant OS as Operating System<br/>(e.g., epoll/kqueue)
    participant R as Reactor (Runtime)

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

    E->>E: (Executor 运行其他任务或睡眠)
    R->>OS: epoll_wait() / Poll 询问 OS 事件

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

    E->>F: 再次调用 poll(cx)
    Note right of F: Future 再次尝试操作
    F->>OS: 系统调用 (e.g., 读取 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};

// 一个立即返回 42 的 future
struct Ready42;

impl Future for Ready42 {
    type Output = i32; // Future 最终产生的值

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<i32> {
        Poll::Ready(42) // 总是就绪——无需等待
    }
}
}

组件：

• Output —— future 完成时产生的值的类型
• poll() —— 由 executor 调用来检查进度；返回 Ready(value) 或 Pending
• Pin<&mut Self> —— 确保 future 不会在内存中移动（我们将在第 4 章解释原因）
• Context —— 携带 Waker 以便 future 在准备好继续执行时向 executor 发信号

Waker 契约

Waker 是回调机制。当 future 返回 Pending 时，它必须安排好稍后调用 waker.wake()——否则 executor 再也不会 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;

/// 一个在延迟后完成的 future（玩具实现）
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<()> {
        // 在存储 waker 之前检查是否已经完成
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        // 存储 waker - executor 可能在每次 poll 时传递一个新的 waker
        *self.waker_stored.lock().unwrap() = Some(cx.waker().clone());

        // 在第一次 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;

                // 关键：唤醒 executor 以便它再次 poll 我们
                if let Some(w) = waker.lock().unwrap().take() {
                    w.wake(); // "嘿 executor，我准备好了——再次 poll 我！"
                }
            });
        }

        // 在存储 waker 后双重检查完成状态（处理竞态条件）
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        Poll::Pending // 还未完成
    }
}
}

核心洞察：在 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(); // 立即调度重新 poll
            Poll::Pending
        }
    }
}
}

关键要点——The Future Trait

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

另见： 第 3 章 — Poll 工作原理 了解 executor 循环，第 6 章 — 手动构建 Futures 了解更复杂的实现

3. Poll 工作原理 🟡

你将学到：

• Executor 的 poll 循环：poll → pending → wake → 再次 poll
• 如何从零构建最小化 executor
• 虚假唤醒规则及其重要性
• 工具函数：poll_fn() 和 yield_now()

Polling 状态机

Executor 运行一个循环：poll 一个 future，如果是 Pending，挂起它直到 waker 触发，然后再次 poll。这与 OS 线程有根本不同，OS 线程由内核处理调度。

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

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

最小化 Executor

为了揭开 executor 的神秘面纱，让我们构建最简单的一个：

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

/// 最简单的 executor：忙循环 poll 直到 Ready
fn block_on<F: Future>(mut future: F) -> F::Output {
    // 将 future pin 到栈上
    // SAFETY: `future` 从这个点起永远不会被移动——我们只
    // 通过 pinned 引用访问它直到完成。
    let mut future = unsafe { Pin::new_unchecked(&mut future) };

    // 创建一个无操作 waker（只是不断 poll——低效但简单）
    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() 返回一个有效的 RawWaker 和正确的 vtable。
    let waker = unsafe { Waker::from_raw(noop_raw_waker()) };
    let mut cx = Context::from_waker(&waker);

    // 忙循环直到 future 完成
    loop {
        match future.as_mut().poll(&mut cx) {
            Poll::Ready(value) => return value,
            Poll::Pending => {
                // 真实的 executor 会在这里 park 线程
                // 并等待 waker.wake() —— 我们只是自旋
                std::thread::yield_now();
            }
        }
    }
}

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

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

唤醒通知

真实的 executor 是事件驱动的。当所有 futures 都是 Pending 时，executor 睡眠。waker 是一种中断机制：

#![allow(unused)]
fn main() {
// 真实 executor 主循环的概念模型：
fn executor_loop(tasks: &mut TaskQueue) {
    loop {
        // 1. Poll 所有被唤醒的 tasks
        while let Some(task) = tasks.get_woken_task() {
            match task.poll() {
                Poll::Ready(result) => task.complete(result),
                Poll::Pending => { /* task 留在队列中，等待唤醒 */ }
            }
        }

        // 2. 睡眠直到有东西唤醒我们（epoll_wait、kevent 等）
        //    这就是 mio/polling 做繁重工作的地方
        tasks.wait_for_events(); // 阻塞直到 I/O 事件或 waker 触发
    }
}
}

虚假唤醒

即使 I/O 未就绪，future 也可能被 poll。这被称为 虚假唤醒。Futures 必须正确处理这种情况：

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

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Data> {
        // ✅ 正确：始终重新检查实际条件
        if let Some(data) = self.try_read_data() {
            Poll::Ready(data)
        } else {
            // 重新注册 waker（它可能已经改变了！）
            self.register_waker(cx.waker());
            Poll::Pending
        }

        // ❌ 错误：假设 poll 意味着数据就绪
        // let data = self.read_data(); // 可能阻塞或 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> {
        // 始终重新检查实际条件——从不完全相信唤醒
        if self.flag.load(Ordering::Acquire) {
            return Poll::Ready(());
        }

        // 存储/更新 waker 以便我们收到通知
        let mut slot = self.waker_slot.lock().unwrap();
        *slot = Some(cx.waker().clone());

        // 在存储 waker 后重新检查以避免竞态：
        // 标志可能在第一次检查和存储 waker 之间被设置
        if self.flag.load(Ordering::Acquire) {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }
}

// 设置标志的一侧（例如，另一个线程或 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();
    }
}

// 使用 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：从闭包创建一个一次性 future
let value = poll_fn(|cx| {
    // 对 cx.waker() 做些什么，返回 Ready 或 Pending
    Poll::Ready(42)
}).await;

// 现实用例：将基于回调的 API 桥接到异步
async fn read_when_ready(source: &MySource) -> Data {
    poll_fn(|cx| source.poll_read(cx)).await
}
}

#![allow(unused)]
fn main() {
// yield_now：自愿将控制权交给 executor
// 在 CPU 密集的异步循环中有用，避免饿死其他 tasks
async fn cpu_heavy_work(items: &[Item]) {
    for (i, item) in items.iter().enumerate() {
        process(item); // CPU 工作

        // 每 100 个元素，yield 让其他 tasks 运行
        if i % 100 == 0 {
            tokio::task::yield_now().await;
        }
    }
}
}

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

关键要点——Poll 工作原理

• Executor 重复调用被唤醒的 futures 的 poll()
• Futures 必须处理虚假唤醒——始终重新检查实际条件
• poll_fn() 让你从闭包创建临时 futures
• yield_now() 是 CPU 密集异步代码的协作式调度逃生舱

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

4. Pin 和 Unpin 🔴

你将学到：

• 为什么自引用 struct 在内存中移动时会破坏
• Pin<P> 保证什么以及它如何防止移动
• 三个实用的 pinning 模式：Box::pin()、tokio::pin!()、Pin::new()
• 何时 Unpin 给你逃生舱

为什么 Pin 存在

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

问题：自引用 Structs

当编译器将 async fn 转换为状态机时，该状态机可能包含对其自身字段的引用。这创建了自引用 struct——在内存中移动它会使这些内部引用失效。

#![allow(unused)]
fn main() {
// 编译器生成的内容（简化）对于：
// async fn example() {
//     let data = vec![1, 2, 3];
//     let reference = &data;       // 指向上面的 data
//     use_ref(reference).await;
// }

// 变成类似：
enum ExampleStateMachine {
    State0 {
        data: Vec<i32>,
        // reference: &Vec<i32>,  // 问题：指向 `data` 上面
        //                        // 如果这个 struct 移动，指针就悬垂了！
    },
    State1 {
        data: Vec<i32>,
        reference: *const Vec<i32>, // 指向 data 字段的内部指针
    },
    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

自引用 Structs

这不是一个学术问题。每个 async fn 如果在 .await 点持有引用，就会创建一个自引用状态机：

#![allow(unused)]
fn main() {
async fn problematic() {
    let data = String::from("hello");
    let slice = &data[..]; // slice 借用 data
    
    some_io().await; // <-- .await 点：状态机同时存储 data 和 slice
    
    println!("{slice}"); // 在 await 后使用引用
}
// 生成的状态机有 `data: String` 和 `slice: &str`
// 其中 slice 指向 data 内部。移动状态机 = 悬垂指针。
}

实践中的 Pin

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

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

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

// Pin 它——现在它不能移动了
let pinned: Pin<&mut String> = Pin::new(&mut data);

// 仍然可以使用它：
println!("{}", pinned.as_ref().get_ref()); // "hello"

// 但我们不能取回 &mut String（这将允许 mem::swap）：
// let mutable: &mut String = Pin::into_inner(pinned); // 仅当 String: Unpin
// String 是 Unpin，所以这实际上对 String 有效。
// 但对于自引用状态机（它们是 !Unpin），它被阻止。
}

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

#![allow(unused)]
fn main() {
// 1. poll() 签名——所有 futures 都通过 Pin 被 poll
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Output>;

// 2. Box::pin() —— 堆分配并 pin 一个 future
let future: Pin<Box<dyn Future<Output = i32>>> = Box::pin(async { 42 });

// 3. tokio::pin!() —— 在栈上 pin 一个 future
tokio::pin!(my_future);
// 现在 my_future: Pin<&mut impl Future>
}

Unpin 逃生舱

Rust 中的大多数类型是 Unpin——它们不包含自引用，所以 pinning 没什么特别的。只有编译器生成的状态机（来自 async fn）是 !Unpin。

#![allow(unused)]
fn main() {
// 这些都是 Unpin——pinning 它们没什么特别作用：
// i32、String、Vec<T>、HashMap<K,V>、Box<T>、&T、&mut T

// 这些是 !Unpin——它们必须被 pinned 才能 poll：
// 由 `async fn` 和 `async {}` 生成的状态机

// 实际含义：
// 如果你手动编写 Future 且没有自引用，
// 实现 Unpin 让它更容易使用：
impl Unpin for MySimpleFuture {} // "我可以安全移动，相信我"
}

快速参考

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

// 片段 B
let fut = async { 42 };
tokio::pin!(fut);
let moved = fut; // 移动 pinned future
let result = moved.await;

// 片段 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;        // 移动 Pin<&mut> 包装器——OK
// fut.await;           // ❌ 错误：fut 被移动
let result = moved.await; // ✅ 使用 moved 代替
}

#![allow(unused)]
fn main() {
let fut = async { 42 };
let pinned = Box::pin(fut); // 堆 pin——适用于 !Unpin
}

关键要点——Pin 和 Unpin

• Pin<P> 是一个包装器，防止被指向的值被移动——对自引用状态机至关重要
• Box::pin() 是安全的、简单的默认方式，用于在堆上 pin futures
• tokio::pin!() 在栈上 pin——你可以移动 Pin<&mut> 包装器，但底层的 future 保持不动
• Unpin 是一个自动 trait opt-out：实现 Unpin 的类型即使被 pinned 也可以移动（大多数类型是 Unpin；async 块不是）

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

5. 状态机揭秘 🟢

你将学到：

• 编译器如何将 async fn 转换为 enum 状态机
• 并排对比：源代码 vs 生成的状态
• 为什么 async fn 中的大栈分配会撑大 future 大小
• Drop 优化：值一旦不再需要就立即 drop

编译器实际生成的内容

当你编写 async fn 时，编译器将你看似顺序的代码转换为基于 enum 的状态机。理解这个转换是理解 Rust 异步性能特征和许多怪癖的关键。

并排对比：async fn vs 状态机

#![allow(unused)]
fn main() {
// 你写的：
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：准备调用 http_get 获取 page1
    Start,

    // State 1：等待 page1，持有 future
    WaitingPage1 {
        fut1: HttpGetFuture,
    },

    // State 2：已获取 page1，等待 page2
    WaitingPage2 {
        page1: String,
        fut2: HttpGetFuture,
    },

    // 终止状态
    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"),
            }
        }
    }
}
}

注意：这个反糖化是概念性的。真实的编译器输出使用 unsafe pin 投影——这里显示的 get_mut() 调用需要 Unpin，但 async 状态机是 !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）

为什么这对性能很重要

零成本：状态机是栈分配的 enum。每个 future 没有堆分配、没有垃圾回收器、没有 boxing——除非你显式使用 Box::pin()。

大小：enum 的大小是所有变体的最大值。每个 .await 点创建一个新的变体。这意味着：

#![allow(unused)]
fn main() {
async fn small() {
    let a: u8 = 0;
    yield_now().await;
    let b: u8 = 0;
    yield_now().await;
}
// 大小 ≈ max(size_of(u8), size_of(u8)) + 判别式 + future 大小
//      ≈ 很小！

async fn big() {
    let buf: [u8; 1_000_000] = [0; 1_000_000]; // 栈上 1MB！
    some_io().await;
    process(&buf);
}
// 大小 ≈ 1MB + 内部 future 大小
// ⚠️ 不要在 async 函数中栈分配巨大的缓冲区！
// 使用 Vec<u8> 或 Box<[u8]> 代替。
}

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

实用规则：async fn 中的大栈分配会撑大 future 的大小。如果你在异步代码中看到栈溢出，检查是否有大数组或深度嵌套的 futures。如果需要，使用 Box::pin() 在堆上分配子 futures。

练习：预测状态机

#![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 点一个变体的 enum
• Future 的大小 = 所有变体大小的最大值——大栈值会撑大它
• 编译器在状态转换时自动插入 drop
• 当 future 大小成为问题时使用 Box::pin() 或堆分配

另见： 第 4 章 — Pin 和 Unpin 了解为什么生成的 enum 需要 pinning，第 6 章 — 手动构建 Futures 亲手构建这些状态机

6. 手动构建 Futures 🟡

你将学到：

• 用基于线程的唤醒实现 TimerFuture
• 构建 Join 组合器：并发运行两个 futures
• 构建 Select 组合器：竞赛两个 futures
• 组合器如何组合——futures 一直向下

一个简单的 Timer Future

现在让我们从头构建真实的、有用的 futures。这将巩固第 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,
        }));

        // 生成一个线程，在 duration 后设置 completed=true
        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(); // 通知 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 {
            // 存储 waker 以便定时器线程可以唤醒我们
            // 重要：始终更新 waker——executor 可能
            // 在两次 poll 之间改变了它
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

// 使用：
// async fn example() {
//     println!("Starting timer...");
//     TimerFuture::new(Duration::from_secs(2)).await;
//     println!("Timer done!");
// }
//
// ⚠️ 这为每个定时器生成一个 OS 线程——学习还可以，但在
// 生产环境中使用 `tokio::time::sleep`，它由共享的
// 定时器轮支持，不需要额外的线程。
}

Join：并发运行两个 Futures

Join poll 两个 futures 并在两者都完成时完成。这是 tokio::join! 内部的工作方式：

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

/// 并发 poll 两个 futures，返回两个结果的元组
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 已被取走
}

// MaybeDone<F> 存储 F::Output，即使 F: Unpin 编译器也无法证明
// 它是 Unpin。因为我们只将 Join 与 Unpin futures 一起使用
// 且从不对字段进行 pin-project，所以手动实现 Unpin 是安全的，
// 让我们可以在 poll() 中调用 self.get_mut()。
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();

        // 如果 A 未完成则 poll
        if let MaybeDone::Pending(ref mut fut) = this.a {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.a = MaybeDone::Done(val);
            }
        }

        // 如果 B 未完成则 poll
        if let MaybeDone::Pending(ref mut fut) = this.b {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.b = MaybeDone::Done(val);
            }
        }

        // 两者都完成了？
        match (&this.a, &this.b) {
            (MaybeDone::Done(_), MaybeDone::Done(_)) => {
                // 取走两个 output
                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, // 至少还有一个仍在 pending
        }
    }
}

// 使用（async 块是 !Unpin，所以用 Box::pin 包装）：
// let (page1, page2) = Join::new(
//     Box::pin(http_get("https://example.com/a")),
//     Box::pin(http_get("https://example.com/b")),
// ).await;
// 两个请求并发运行！
}

核心洞察：这里的“并发“意味着在同一线程上交错。 Join 不生成线程——它在同一个 poll() 调用中 poll 两个 futures。 这是协作式并发，不是并行。

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：竞赛两个 Futures

Select 在任一 future 先完成时完成（另一个被 drop）：

#![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),
}

/// 返回先完成的 future；drop 另一个
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
        if let Poll::Ready(val) = Pin::new(&mut self.a).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

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

        Poll::Pending
    }
}

// 带超时的使用：
// match Select::new(http_get(url), TimerFuture::new(timeout)).await {
//     Either::Left(response) => println!("Got response: {}", response),
//     Either::Right(()) => println!("Request timed out!"),
// }
}

公平性说明：我们的 Select 总是先 poll A——如果两者都就绪，A 总是赢。Tokio 的 select! 宏为了公平性随机化 poll 顺序。

#![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>> 总是 Unpin，所以当 F 和 E 是 Unpin 时 struct 也是 Unpin。
        // 这让我们可以安全地使用 get_mut() 而无需任何 unsafe 代码。
        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)()));
                            // 循环立即 poll 新的 future
                        } 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()));
            }
        }
    }
}

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

关键要点——手动构建 Futures

• future 需要三件事：状态、poll() 实现和 waker 注册
• Join poll 两个子 futures；Select 返回先完成的那个
• 组合器本身是包装其他 futures 的 futures——无穷尽的 turtles
• 手动构建 futures 提供深刻洞察，但在生产环境中使用 tokio::join!/select!

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

7. Executors 和 Runtimes 🟡

你将学到：

• Executor 做什么：poll + 高效睡眠
• 六个主要 runtime：mio、io_uring、tokio、async-std、smol、embassy
• 选择合适 runtime 的决策树
• 为什么 runtime 无关的库设计很重要

Executor 做什么

Executor 有两项工作：

1. Poll futures 当它们准备好继续执行时
2. 高效睡眠 当没有 futures 就绪时（使用 OS I/O 通知 API）

graph TB
    subgraph Executor["Executor (e.g., tokio)"]
        QUEUE["Task Queue"]
        POLLER["I/O Poller<br/>(epoll/kqueue/io_uring)"]
        THREADS["Worker Thread Pool"]
    end

    subgraph Tasks
        T1["Task 1<br/>(HTTP request)"]
        T2["Task 2<br/>(DB query)"]
        T3["Task 3<br/>(File read)"]
    end

    subgraph OS["Operating System"]
        NET["Network Stack"]
        DISK["Disk I/O"]
    end

    T1 --> QUEUE
    T2 --> QUEUE
    T3 --> QUEUE
    QUEUE --> THREADS
    THREADS -->|"poll()"| T1
    THREADS -->|"poll()"| T2
    THREADS -->|"poll()"| T3
    POLLER <-->|"register/notify"| NET
    POLLER <-->|"register/notify"| DISK
    POLLER -->|"wake tasks"| QUEUE

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

mio：基础层

mio（Metal I/O）不是一个 executor——它是最低级的跨平台 I/O 通知库。它包装了 epoll（Linux）、kqueue（macOS/BSD）和 IOCP（Windows）。

#![allow(unused)]
fn main() {
// 概念性的 mio 使用（简化）：
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)?;

// 事件循环——阻塞直到有事发生
loop {
    poll.poll(&mut events, None)?; // 睡眠直到 I/O 事件
    for event in events.iter() {
        match event.token() {
            Token(0) => { /* 服务器有新连接 */ }
            _ => { /* 其他 I/O 就绪 */ }
        }
    }
}
}

大多数开发者从不直接使用 mio——tokio 和 smol 构建在它之上。

io_uring：基于 Completion 的 Future

Linux 的 io_uring（内核 5.1+）代表了从 mio/epoll 使用的基于 readiness 的 I/O 模型的根本转变：

基于 Readiness（epoll / mio / tokio）：
  1. 问："这个 socket 可读吗？"     → epoll_wait()
  2. 内核："是的，它就绪了"           → EPOLLIN 事件
  3. 应用：read(fd, buf)                → 可能仍然短暂阻塞！

基于 Completion（io_uring）：
  1. 提交："从这个 socket 读到这个缓冲区"  → SQE
  2. 内核：异步执行读取
  3. 应用：获取完成的结果和数据           → CQE

graph LR
    subgraph "Readiness Model (epoll)"
        A1["App：它就绪吗？"] --> K1["内核：是的"]
        K1 --> A2["App：现在 read()"]
        A2 --> K2["内核：这是数据"]
    end

    subgraph "Completion Model (io_uring)"
        B1["App：为我读这个"] --> K3["内核：处理中..."]
        K3 --> B2["App：得到结果 + 数据"]
    end

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

所有权挑战：io_uring 要求内核在操作完成前拥有缓冲区。这与 Rust 标准 AsyncRead trait 借用缓冲区的方式冲突。这就是为什么 tokio-uring 有不同的 I/O traits：

#![allow(unused)]
fn main() {
// 标准 tokio（基于 readiness）——借用缓冲区：
let n = stream.read(&mut buf).await?;  // buf 被借用

// tokio-uring（基于 completion）——获取缓冲区所有权：
let (result, buf) = stream.read(buf).await;  // buf 被移入，然后返回
let n = result?;
}

// Cargo.toml: tokio-uring = "0.5"
// 注意：仅限 Linux，需要内核 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 或网络，其中系统调用开销是瓶颈（数据库、存储引擎、代理服务于 100k+ 连接）。对于大多数应用，标准的 tokio 与 epoll 是正确的选择。

tokio：功能齐全的 Runtime

Rust 生态系统中占主导地位的 async runtime。被 Axum、Hyper、Tonic 和大多数生产级 Rust 服务器使用。

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

#[tokio::main]
async fn main() {
    // 生成一个带工作窃取调度器的多线程 runtime
    let handle = tokio::spawn(async {
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        "done"
    });

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

tokio 特性：Timer、I/O、TCP/UDP、Unix sockets、信号处理、同步原语（Mutex、RwLock、Semaphore、channels）、fs、process、tracing 集成。

async-std：标准库镜像

用 async 版本镜像 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：极简主义 Runtime

小型、零依赖的 async runtime。对想要 async 但不想引入 tokio 的库很好。

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

fn main() {
    smol::block_on(async {
        let result = smol::unblock(|| {
            // 在线程池上运行阻塞代码
            std::fs::read_to_string("hello.txt")
        }).await.unwrap();
        println!("{result}");
    });
}

embassy：嵌入式的 Async（no_std）

用于嵌入式系统的 async runtime。不需要堆分配，不需要 std。

// 运行在微控制器上（例如 STM32、nRF52、RP2040）
#[embassy_executor::main]
async fn main(spawner: embassy_executor::Spawner) {
    // 用 async/await 闪烁 LED——不需要 RTOS！
    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;
    }
}

Runtime 决策树

graph TD
    START["选择 Runtime"]

    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["🔵 runtime 无关<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

Runtime 对比表

// ----- tokio 版本 -----
// 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 版本 -----
// 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 版本 -----
// 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}");
}

关键要点——Executors 和 Runtimes

• Executor 的工作：当被唤醒时 poll futures，使用 OS I/O API 高效睡眠
• tokio 是服务器的默认选择；smol 用于最小占用；embassy 用于嵌入式
• 你的业务逻辑应该依赖 std::future::Future，而不是特定的 runtime
• io_uring（Linux 5.1+）是高性能 I/O 的未来，但生态系统仍在成熟中

另见： 第 8 章 — Tokio 深入 了解 tokio 细节，第 9 章 — 何时 Tokio 不是最佳选择 了解替代方案

8. Tokio Deep Dive 🟡

你将学到：

• Runtime 配置：多线程 vs 当前线程以及何时使用
• tokio::spawn、'static 要求和 JoinHandle
• 任务取消语义（drop 时取消）
• 同步原语：Mutex、RwLock、Semaphore 和四种 channel 类型

Runtime 配置：多线程 vs 当前线程

Tokio 提供两种 runtime 配置：

// 多线程（默认，使用 #[tokio::main]）
// 使用工作窃取线程池 —— 任务可以在线程间移动
#[tokio::main]
async fn main() {
    // N 个工作线程（默认 = CPU 核心数）
    // 任务需要 Send + 'static
}

// 当前线程 —— 所有事情在一个线程上运行
#[tokio::main(flavor = "current_thread")]
async fn main() {
    // 单线程 —— 任务不需要 Send
    // 更轻量，适合简单工具或 WASM
}

// 手动构建 runtime：
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 "Multi-Thread (default)"
        MT_Q1["Thread 1<br/>Task A, Task D"]
        MT_Q2["Thread 2<br/>Task B"]
        MT_Q3["Thread 3<br/>Task C, Task E"]
        STEAL["Work Stealing:<br/>idle threads steal from busy ones"]
        MT_Q1 <--> STEAL
        MT_Q2 <--> STEAL
        MT_Q3 <--> STEAL
    end

    subgraph "Current-Thread"
        ST_Q["Single Thread<br/>Task A → Task B → Task C → Task 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 放入 runtime 的任务队列。因为它可能在任何工作线程上、在任何时间运行，future 必须是 Send + 'static：

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

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

    // ✅ 可行：将所有权移入任务
    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");

    // ❌ 失败：data 是借用，不是 'static
    // task::spawn(async {
    //     println!("{data}"); // 借用 `data` —— 不是 'static
    // });

    // ❌ 失败：Rc 不是 Send
    // let rc = std::rc::Rc::new(42);
    // task::spawn(async move {
    //     println!("{rc}"); // Rc 是 !Send —— 不能跨线程边界
    // });
}
}

为什么需要 'static？ 派生的任务独立运行 —— 它可能比创建它的作用域存活得更久。编译器无法证明引用会保持有效，所以要求拥有数据的所有权。

为什么需要 Send？ 任务可能在不同线程上恢复执行，而不是它挂起时所在的线程。所有在 .await 点之间持有的数据必须安全地在线程间发送。

#![allow(unused)]
fn main() {
// 常见模式：克隆共享数据到任务中
let shared = Arc::new(config);

for i in 0..10 {
    let shared = Arc::clone(&shared); // 克隆 Arc，不是数据本身
    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()
    });

    // 通过 drop handle 来取消任务？不 —— 任务继续运行！
    // drop(handle); // 任务在后台继续运行

    // 要真正取消，调用 abort()：
    handle.abort();

    // 等待被取消的任务返回 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}"),
    }
}
}

重要：Drop JoinHandle 不会取消 tokio 中的任务。 任务变成分离的并继续运行。你必须显式调用 .abort() 来取消它。 这与直接 drop Future 不同 —— 直接 drop 会取消/停止底层的计算。

Tokio 同步原语

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

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

// --- Mutex ---
// 异步互斥锁：lock() 方法是异步的，不会阻塞线程
let data = Arc::new(Mutex::new(vec![1, 2, 3]));
{
    let mut guard = data.lock().await; // 非阻塞锁
    guard.push(4);
} // Guard 在这里 drop —— 锁释放

// --- Channels ---
// mpsc: 多生产者，单消费者
let (tx, mut rx) = mpsc::channel::<String>(100); // 有界缓冲区

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

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

// oneshot: 单值，单消费者
let (tx, rx) = oneshot::channel::<i32>();
tx.send(42).unwrap(); // 不需要 await —— 要么发送成功，要么失败
let val = rx.await.unwrap();

// broadcast: 多生产者，多消费者（所有接收者都收到每条消息）
let (tx, _) = broadcast::channel::<String>(100);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

// watch: 单值，多消费者（只有最新值）
let (tx, rx) = watch::channel(0u64);
tx.send(42).unwrap();
println!("Latest: {}", *rx.borrow());
}

注意：这些 channel 示例中为了简洁使用了 .unwrap()。 在生产环境中，优雅地处理发送/接收错误 —— 失败的 .send() 意味着 接收者被 drop 了，失败的 .recv() 意味着 channel 已关闭。

graph LR
    subgraph "Channel Types"
        direction TB
        MPSC["mpsc<br/>N→1<br/>Buffered queue"]
        ONESHOT["oneshot<br/>1→1<br/>Single value"]
        BROADCAST["broadcast<br/>N→N<br/>All receivers get all"]
        WATCH["watch<br/>1→N<br/>Latest value only"]
    end

    P1["Producer 1"] --> MPSC
    P2["Producer 2"] --> MPSC
    MPSC --> C1["Consumer"]

    P3["Producer"] --> ONESHOT
    ONESHOT --> C2["Consumer"]

    P4["Producer"] --> BROADCAST
    BROADCAST --> C3["Consumer 1"]
    BROADCAST --> C4["Consumer 2"]

    P5["Producer"] --> WATCH
    WATCH --> C5["Consumer 1"]
    WATCH --> C6["Consumer 2"]

案例研究：为通知服务选择合适的 Channel

你正在构建一个通知服务，其中：

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

每种情况选择哪种 channel？

graph LR
    subgraph "Notification Service"
        direction TB
        API1["API Handler 1"] -->|mpsc| BATCH["Batcher"]
        API2["API Handler 2"] -->|mpsc| BATCH
        CONFIG["Config Watcher"] -->|watch| RATE["Rate Limiter"]
        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 在任务运行时持有，然后 drop
            task().await
        });
        handles.push(handle);
    }

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

// 使用：
// let tasks: Vec<_> = urls.into_iter().map(|url| {
//     move || async move { fetch(url).await }
// }).collect();
// let results = run_with_limit(tasks, 10).await; // 最多 10 个并发
}

关键要点——Tokio Deep Dive

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

另见： 第 9 章 — 何时 Tokio 不是最佳选择 了解 spawn 的替代方案，第 12 章 — 常见陷阱 了解 MutexGuard 跨 await 的 bug

9. When Tokio Isn’t the Right Fit 🟡

你将学到：

• 'static 问题：何时 tokio::spawn 迫使你到处使用 Arc
• LocalSet 用于 !Send futures
• FuturesUnordered 用于借用友好的并发（不需要 spawn）
• JoinSet 用于管理的任务组
• 编写 runtime 无关的库

graph TD
    START["Need concurrent futures?"] --> STATIC{"Can futures be 'static?"}
    STATIC -->|Yes| SEND{"Are futures Send?"}
    STATIC -->|No| FU["FuturesUnordered<br/>Runs on current task"]
    SEND -->|Yes| SPAWN["tokio::spawn<br/>Multi-threaded"]
    SEND -->|No| LOCAL["LocalSet<br/>Single-threaded"]
    SPAWN --> MANAGE{"Need to track/abort tasks?"}
    MANAGE -->|Yes| JOINSET["JoinSet / TaskTracker"]
    MANAGE -->|No| 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 futures。这意味着你不能在派生的任务中借用本地数据：

#![allow(unused)]
fn main() {
async fn process_items(items: &[String]) {
    // ❌ 不能这样做 —— items 是借用，不是 'static
    // for item in items {
    //     tokio::spawn(async {
    //         process(item).await;
    //     });
    // }

    // 😐 变通方案 1：克隆所有数据
    for item in items {
        let item = item.clone();
        tokio::spawn(async move {
            process(&item).await;
        });
    }

    // 😐 变通方案 2：使用 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 —— 完全避免 'static（不需要 spawn！）
use futures::stream::{FuturesUnordered, StreamExt};

async fn process_items(items: &[String]) {
    let futures: FuturesUnordered<_> = items
        .iter()
        .map(|item| async move {
            // ✅ 可以借用 item —— 不需要 spawn，不需要 'static！
            process(item).await
        })
        .collect();

    // 驱动所有 futures 完成
    futures.for_each(|result| async move {
        println!("Result: {result:?}");
    }).await;
}

// 2. tokio::task::LocalSet —— 在当前线程运行 !Send futures
//    ⚠️  仍然需要 'static —— 解决 Send 问题，不是 'static
use tokio::task::LocalSet;

let local_set = LocalSet::new();
local_set.run_until(async {
    tokio::task::spawn_local(async {
        // 可以在这里使用 Rc、Cell 和其他 !Send 类型
        let rc = std::rc::Rc::new(42);
        println!("{rc}");
    }).await.unwrap();
}).await;

// 3. tokio JoinSet (tokio 1.21+) —— 管理派生任务集
//    ⚠️  仍然需要 'static + Send —— 解决任务*管理*问题，
//    不是 'static 问题。适合跟踪、中止和加入动态任务组。
use tokio::task::JoinSet;

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

    for i in 0..10 {
        // i 是 Copy 并被移入闭包 —— 已经是 'static。
        // 对于借用数据，你仍然需要 Arc 或 clone。
        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？
无法让 futures 成为 'static	FuturesUnordered	✅ 是	✅ 是
Futures 是 'static 但 !Send	LocalSet	❌ 否	✅ 是
需要跟踪/中止派生的任务	JoinSet	❌ 否	❌ 否

库的轻量级 Runtimes

如果你在编写库 —— 不要强迫用户使用 tokio：

#![allow(unused)]
fn main() {
// ❌ 坏：库强迫用户使用 tokio
pub async fn my_lib_function() {
    tokio::time::sleep(Duration::from_secs(1)).await;
    // 现在你的用户*必须*使用 tokio
}

// ✅ 好：库是 runtime 无关的
pub async fn my_lib_function() {
    // 只使用来自 std::future 和 futures crate 的类型
    do_computation().await;
}

// ✅ 好：为 I/O 操作接受泛型 future
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（或它们选择的 runtime）。 这保持了生态系统的可组合性。

#![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]);
}
}

关键要点——When Tokio Isn’t the Right Fit

• FuturesUnordered 在当前位置上并发运行 futures —— 没有 'static 要求
• LocalSet 在单线程 executor 上启用 !Send futures
• JoinSet (tokio 1.21+) 提供管理的任务组，带有自动清理
• 对于库：只依赖 std::future::Future + futures crate，不直接依赖 tokio

另见： 第 8 章 — Tokio Deep Dive 了解何时 spawn 是正确的工具，第 11 章 — Streams 了解 buffer_unordered() 作为另一种并发限制器

10. Async Traits 🟡

你将学到：

• 为什么 async trait methods 花了数年时间才稳定
• RPITIT：原生的 async trait 方法（Rust 1.75+）
• dyn 派发的挑战和 trait_variant 变通方案
• Async 闭包（Rust 1.85+）：async Fn() 和 async FnOnce()

graph TD
    subgraph "Async Trait Approaches"
        direction TB
        RPITIT["RPITIT (Rust 1.75+)<br/>async fn in trait<br/>Static dispatch only"]
        VARIANT["trait_variant<br/>Auto-generates Send variant<br/>Enables dyn dispatch"]
        BOXED["Box&lt;dyn Future&gt;<br/>Manual boxing<br/>Works everywhere"]
        CLOSURE["Async Closures (1.85+)<br/>async Fn() / async FnOnce()<br/>Callbacks & middleware"]
    end

    RPITIT -->|"Need dyn?"| VARIANT
    RPITIT -->|"Pre-1.75?"| BOXED
    CLOSURE -->|"Replaces"| 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

历史：为什么花了这么长时间

Async methods in traits 多年来一直是 Rust 最请求的功能。问题是：

#![allow(unused)]
fn main() {
// 这在 Rust 1.75 (2023 年 12 月) 之前无法编译：
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
}
// 为什么？因为 async fn 返回 `impl Future<Output = T>`，
// 而 trait 返回位置不支持 `impl Trait`。
}

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

RPITIT: Return Position Impl Trait in 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()
    }
}

// ✅ 适用于泛型（静态派发）：
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() {
// ❌ 无法工作：
// async fn lookup_dyn(store: &dyn DataStore, key: &str) { ... }
// Error: the trait `DataStore` is not dyn compatible because method `get`
//        is `async`

// ✅ 变通方案：返回 box future
trait DynDataStore {
    fn get(&self, key: &str) -> Pin<Box<dyn Future<Output = Option<String>> + Send + '_>>;
}

// 或者使用 trait_variant 宏（见下文）
}

Send 问题：在多线程 runtime 中，派生的任务必须是 Send。但 async trait 方法不会自动添加 Send 约束：

#![allow(unused)]
fn main() {
trait Worker {
    async fn run(self); // Future 可能 Send 也可能不 Send
}

struct MyWorker;

impl Worker for MyWorker {
    async fn run(self) {
        // 如果这使用 !Send 类型，future 就是 !Send
        let rc = std::rc::Rc::new(42);
        some_work().await;
        println!("{rc}");
    }
}

// ❌ 这失败是因为 future 是 !Send（Rc 是 !Send）：
// tokio::spawn(worker.run()); // Requires Send + 'static
//
// 注意：这里我们用 `self`（owned）是因为 tokio::spawn 也需要 'static —— 
// 一个借用 &self 的 future 不能是 'static。
// 即使没有 Rc，`async fn run(&self)` 也不能被 spawn。
}

trait_variant Crate

trait_variant crate（来自 Rust async 工作组）自动生成一个 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);
}

// 现在你有两个 traits：
// - DataStore: futures 上没有 Send 约束
// - SendDataStore: 所有 futures 都是 Send
// 两者有相同的方法，实现者实现 DataStore
// 如果他们的 futures 是 Send，则免费获得 SendDataStore。

// 当你需要 spawn 时使用 SendDataStore：
async fn spawn_lookup(store: Arc<dyn SendDataStore>) {
    tokio::spawn(async move {
        store.get("key").await;
    });
}
}

快速参考：Async Traits

推荐：对于新代码（Rust 1.75+），使用原生 async traits， 当你需要 dyn 派发时使用 trait_variant。async-trait crate 仍然 广泛使用，但 box 每个 future —— 原生方法对静态派发是零成本的。

Async 闭包（Rust 1.85+）

从 Rust 1.85 开始，async 闭包 稳定了 —— 捕获环境并返回 future 的闭包：

#![allow(unused)]
fn main() {
// 1.85 之前：笨拙的变通方案
let urls = vec!["https://a.com", "https://b.com"];
let fetchers: Vec<_> = urls.iter().map(|url| {
    let url = url.to_string();
    // 返回一个非 async 闭包，返回一个 async 块
    move || async move { reqwest::get(&url).await }
}).collect();

// 1.85 之后：async 闭包直接可用
let fetchers: Vec<_> = urls.iter().map(|url| {
    async move || { reqwest::get(url).await }
    // ↑ 这是 async 闭包 —— 捕获 url，返回一个 Future
}).collect();
}

Async 闭包实现了新的 AsyncFn、AsyncFnMut 和 AsyncFnOnce traits，它们镜像 Fn、FnMut、FnOnce：

#![allow(unused)]
fn main() {
// 接受 async 闭包的泛型函数
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);
}

// --- 内存中实现 ---
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);
    }
}

// --- 模拟 Redis 实现 ---
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; // 模拟网络往返
        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);
    }
}

// --- 泛型函数适用于任何 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;
}

关键要点——Async Traits

• 从 Rust 1.75 开始，你可以在 traits 中直接编写 async fn（不需要 #[async_trait] crate）
• trait_variant::make 为动态派发自动生成 Send 变体
• Async 闭包（async Fn()）在 1.85 稳定 —— 用于回调和中间件
• 对于性能关键代码，优先使用静态派发（<S: Service>）而不是 dyn

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

11. Streams and AsyncIterator 🟡

你将学到：

• Stream trait：异步迭代多个值
• 创建 streams：stream::iter、async_stream、unfold
• Stream 组合器：map、filter、buffer_unordered、fold
• Async I/O traits：AsyncRead、AsyncWrite、AsyncBufRead

Stream Trait 概述

Stream 对于 Iterator 就像 Future 对于单个值 —— 它异步地产出多个值：

#![allow(unused)]
fn main() {
// std::iter::Iterator (同步，多个值)
trait Iterator {
    type Item;
    fn next(&mut self) -> Option<Self::Item>;
}

// futures::Stream (异步，多个值)
trait Stream {
    type Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}
}

graph LR
    subgraph "Sync"
        VAL["Value<br/>(T)"]
        ITER["Iterator<br/>(multiple T)"]
    end

    subgraph "Async"
        FUT["Future<br/>(async T)"]
        STREAM["Stream<br/>(async multiple T)"]
    end

    VAL -->|"make async"| FUT
    ITER -->|"make async"| STREAM
    VAL -->|"make multiple"| ITER
    FUT -->|"make multiple"| 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

创建 Streams

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

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

// 2. 从 async generator（使用 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. 从 tokio interval
let tick_stream = IntervalStream::new(interval(Duration::from_secs(1)));

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

// 5. 从 unfold（从异步状态生成）
let s = stream::unfold(0u32, |state| async move {
    if state >= 5 {
        None // Stream 结束
    } else {
        let next = state + 1;
        Some((state, next)) // yield `state`，新状态是 `next`
    }
});
}

消费 Streams

#![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 —— 处理每个 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 —— 并发处理 N 个 items
    let results: Vec<_> = stream::iter(vec!["url1", "url2", "url3"])
        .map(|url| async move {
            // 模拟 HTTP 获取
            tokio::time::sleep(Duration::from_millis(100)).await;
            format!("response from {url}")
        })
        .buffer_unordered(10) // 最多 10 个并发获取
        .collect()
        .await;

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

与 C# IAsyncEnumerable 对比

#![allow(unused)]
fn main() {
// Rust: 数据库行的 Stream
// 注意：当在 body 内使用 ? 时需要 try_stream!（不是 stream!）。
// stream! 不传播错误 —— try_stream! 产出 Err(e) 并结束。
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?);
        }
    }
}

// 消费：
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# 等价实现：
async IAsyncEnumerable<User> GetUsers() {
    await using var reader = await db.QueryAsync("SELECT * FROM users");
    while (await reader.ReadAsync()) {
        yield return User.FromRow(reader);
    }
}

// 消费：
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> + Unpin>(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);
}
}

Async I/O Traits：AsyncRead、AsyncWrite、AsyncBufRead

就像 std::io::Read/Write 是同步 I/O 的基础一样，它们的 async 版本是 async I/O 的基础。这些 traits 由 tokio::io 提供（或 futures::io 用于 runtime 无关的代码）：

#![allow(unused)]
fn main() {
// tokio::io —— std::io traits 的 async 版本

/// 异步从源读取字节
pub trait AsyncRead {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,  // Tokio 对未初始化内存的安全包装
    ) -> Poll<io::Result<()>>;
}

/// 异步写字节到 sink
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<()>>;
}

/// 支持行的缓冲读取
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_* 方法。相反，使用扩展 traits 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 等
    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(())
}
}

实现自定义 async I/O —— 在原始 TCP 之上包装协议：

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

/// 长度前缀协议：[u32 length][payload bytes]
struct FramedStream<T> {
    inner: T,
}

impl<T: AsyncRead + AsyncReadExt + Unpin> FramedStream<T> {
    /// 读取一个完整的帧
    async fn read_frame(&mut self) -> tokio::io::Result<Vec<u8>>
    {
        // 读取 4 字节长度前缀
        let len = self.inner.read_u32().await? as usize;

        // 准确读取那么多字节
        let mut payload = vec![0u8; len];
        self.inner.read_exact(&mut payload).await?;
        Ok(payload)
    }
}

impl<T: AsyncWrite + AsyncWriteExt + Unpin> FramedStream<T> {
    /// 写入一个完整的帧
    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 vs futures I/O traits：它们相似但不完全相同 —— tokio 的 AsyncRead 使用 ReadBuf（安全处理未初始化内存），而 futures::AsyncRead 使用 &mut [u8]。使用 tokio_util::compat 在它们之间转换。

Copy 工具：tokio::io::copy(&mut reader, &mut writer) 是 std::io::copy 的 async 版本 —— 用于代理服务器或文件传输。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)
}

// 适用于任何 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?;
}

关键要点——Streams and AsyncIterator

• Stream 是 Iterator 的 async 版本 —— 产出 Poll::Ready(Some(item)) 或 Poll::Ready(None)
• .buffer_unordered(N) 并发处理 N 个 stream items —— streams 的关键并发工具
• async_stream::stream! 是创建自定义 streams 的最简单方式（使用 yield）
• AsyncRead/AsyncBufRead 支持通用的、可复用的 I/O 代码，适用于文件、socket 和管道

另见： 第 9 章 — 何时 Tokio 不是最佳选择 了解 FuturesUnordered（相关模式），第 13 章 — 生产级模式 了解有界 channel 的背压

12. Common Pitfalls 🔴

你将学到：

• 9 个常见 async Rust bug 及如何修复
• 为什么阻塞 executor 是头号错误（以及 spawn_blocking 如何修复）
• 取消风险：当 future 在 await 中途被 drop 时会发生什么
• 调试：tokio-console、tracing、#[instrument]
• 测试：#[tokio::test]、time::pause()、基于 trait 的 mock

阻塞 Executor

async Rust 中的头号错误：在 async executor 线程上运行阻塞代码。这会饿死其他任务。

#![allow(unused)]
fn main() {
// ❌ 错误：阻塞整个 executor 线程
async fn bad_handler() -> String {
    let data = std::fs::read_to_string("big_file.txt").unwrap(); // 阻塞！
    process(&data)
}

// ✅ 正确：将阻塞工作卸载到专用的线程池
async fn good_handler() -> String {
    let data = tokio::task::spawn_blocking(|| {
        std::fs::read_to_string("big_file.txt").unwrap()
    }).await.unwrap();
    process(&data)
}

// ✅ 也正确：使用 tokio 的 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 "❌ Blocking Call on Executor"
        T1_BAD["Thread 1: std::fs::read()<br/>🔴 BLOCKED for 500ms"]
        T2_BAD["Thread 2: handling requests<br/>🟢 Working alone"]
        TASKS_BAD["100 pending tasks<br/>⏳ Starved"]
        T1_BAD -->|"can't poll"| TASKS_BAD
    end

    subgraph "✅ spawn_blocking"
        T1_GOOD["Thread 1: polling futures<br/>🟢 Available"]
        T2_GOOD["Thread 2: polling futures<br/>🟢 Available"]
        BT["Blocking pool thread:<br/>std::fs::read()<br/>🔵 Separate pool"]
        TASKS_GOOD["100 tasks<br/>✅ All making progress"]
        T1_GOOD -->|"polls"| TASKS_GOOD
        T2_GOOD -->|"polls"| TASKS_GOOD
    end

std::thread::sleep vs tokio::time::sleep

#![allow(unused)]
fn main() {
// ❌ 错误：阻塞 executor 线程 5 秒
async fn bad_delay() {
    std::thread::sleep(Duration::from_secs(5)); // 线程无法 poll 其他事情！
}

// ✅ 正确：yield 给 executor，其他任务可以运行
async fn good_delay() {
    tokio::time::sleep(Duration::from_secs(5)).await; // 非阻塞！
}
}

在 .await 之间持有 MutexGuard

#![allow(unused)]
fn main() {
use std::sync::Mutex; // std Mutex —— 不是 async 感知

// ⚠️ 风险：MutexGuard 跨 .await 持有
async fn bad_mutex(data: &Mutex<Vec<String>>) {
    let mut guard = data.lock().unwrap();
    guard.push("item".into());
    some_io().await; // Guard 在这里持有 —— 阻塞其他线程锁定！
    guard.push("another".into());
}
// 注意：这能编译！std::sync::MutexGuard 是 !Send，但编译器只在
// 你将 Future 传递给需要它的东西时（例如 tokio::spawn）才强制执行 Send。
// 直接调用 bad_mutex(...).await 编译没问题。
}

为什么这通常是个问题 —— 但并不总是：

跨 .await 持有 std::sync::Mutex 会阻塞 OS 线程 整个 I/O 过程， 阻止 executor 在该线程上 poll 其他任务。对于短临界区这是浪费， 对于长 I/O 这是性能陷阱。

然而，有些合法情况你必须跨 .await 持有锁 —— 就像数据库 事务在读取和提交之间持有锁一样。drop 和重新获取锁引入了 TOCTOU (time-of-check to time-of-use 竞态)：另一个任务可以在 你的两个临界区之间修改数据。正确的修复取决于用例：

#![allow(unused)]
fn main() {
// 方案 1：限定 guard 作用域 —— 适用于操作独立时
async fn scoped_mutex(data: &Mutex<Vec<String>>) {
    {
        let mut guard = data.lock().unwrap();
        guard.push("item".into());
    } // Guard 在这里 drop
    some_io().await; // 锁释放 —— 其他任务可以进行
    {
        let mut guard = data.lock().unwrap();
        guard.push("another".into());
    }
}
// ⚠️ 注意：另一个任务可以在两次 push 之间锁定 + 修改 Vec。
//    如果两次 push 是独立的这没问题，但如果 "another" 依赖于
//    "item" 设置的状态就错了。

// 方案 2：使用 tokio::sync::Mutex —— 跨 .await 持有锁而不阻塞 OS 线程。
//           当你需要在 await 点之间进行事务性读 - 改 - 写时最好。
use tokio::sync::Mutex as AsyncMutex;

async fn async_mutex(data: &AsyncMutex<Vec<String>>) {
    let mut guard = data.lock().await; // 异步锁 —— 不阻塞线程
    guard.push("item".into());
    some_io().await; // OK —— tokio Mutex guard 是 Send
    guard.push("another".into());
    // Guard 整个过程持有 —— 没有 TOCTOU 竞态，没有线程阻塞。
}
}

何时使用哪种 Mutex：

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

经验法则：不要盲目地在 .await 周围分割临界区。 问自己两半是否真正独立。如果不是 —— 如果第二半依赖于 第一半的状态 —— 使用 tokio::sync::Mutex 或重新设计数据流。

取消风险

Drop future 会取消它 —— 但这可能让事物处于不一致状态：

#![allow(unused)]
fn main() {
// ❌ 危险：取消时资源泄漏
async fn transfer(from: &Account, to: &Account, amount: u64) {
    from.debit(amount).await;  // 如果在这里取消...
    to.credit(amount).await;   // ...钱消失了！
}

// ✅ 安全：让操作原子化或使用补偿
async fn safe_transfer(from: &Account, to: &Account, amount: u64) -> Result<(), Error> {
    // 使用数据库事务（全有或全无）
    let tx = db.begin_transaction().await?;
    tx.debit(from, amount).await?;
    tx.credit(to, amount).await?;
    tx.commit().await?; // 只有所有事情成功才提交
    Ok(())
}

// ✅ 也安全：使用 tokio::select! 与取消感知
tokio::select! {
    result = transfer(from, to, amount) => {
        // 转账完成
    }
    _ = shutdown_signal() => {
        // 不要在转账中途取消 —— 让它完成
        // 或者：显式回滚
    }
}
}

没有 Async Drop

Rust 的 Drop trait 是同步的 —— 你不能在 drop() 内部 .await。这是一个常见的混淆源：

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

impl Drop for DbConnection {
    fn drop(&mut self) {
        // ❌ 不能这样做 —— drop() 是同步的！
        // self.connection.shutdown().await;

        // ✅ 变通方案 1：生成清理任务（即发即弃）
        let conn = self.connection.take();
        tokio::spawn(async move {
            let _ = conn.shutdown().await;
        });

        // ✅ 变通方案 2：使用同步关闭
        // self.connection.blocking_close();
    }
}
}

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

select! 公平性和饥饿

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

// ❌ 不公平：busy_stream 总是赢，slow_stream 饥饿
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}"),
            // 如果两者都就绪，tokio 随机选择一个。
            // 但如果 `fast` 总是就绪，`slow` 很少被 poll。
        }
    }
}

// ✅ 公平：使用 biased select 或批量 drain
async fn fair(mut fast: mpsc::Receiver<i32>, mut slow: mpsc::Receiver<i32>) {
    loop {
        tokio::select! {
            biased; // 总是按顺序检查 —— 显式优先级

            Some(v) = slow.recv() => println!("slow: {v}"),  // 优先级！
            Some(v) = fast.recv() => println!("fast: {v}"),
        }
    }
}
}

意外的顺序执行

#![allow(unused)]
fn main() {
// ❌ 顺序执行：总共花费 2 秒
async fn slow() {
    let a = fetch("url_a").await; // 1 秒
    let b = fetch("url_b").await; // 1 秒（等 a 先完成！）
}

// ✅ 并发执行：总共花费 1 秒
async fn fast() {
    let (a, b) = tokio::join!(
        fetch("url_a"), // 两者立即启动
        fetch("url_b"),
    );
}

// ✅ 也并发：使用 let + join
async fn also_fast() {
    let fut_a = fetch("url_a"); // 创建 future（惰性 —— 还未启动）
    let fut_b = fetch("url_b"); // 创建 future
    let (a, b) = tokio::join!(fut_a, fut_b); // 现在两者并发运行
}
}

陷阱：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
• 为 async 函数添加 #[tracing::instrument] 用于 span 跟踪
• 在 staging 中运行 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)); // 限流
        let mut guard = results.lock().unwrap();
        guard.push(response);
        expensive_parse(&guard).await; // 解析目前为止的所有结果
    }
    
    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> {
    // 修复 4：使用 buffer_unordered 并发处理 URLs
    let results: Vec<String> = stream::iter(urls)
        .map(|url| async move {
            let response = reqwest::get(&url).await.unwrap().text().await.unwrap();
            // 修复 2：使用 tokio::time::sleep 而不是 std::thread::sleep
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            response
        })
        .buffer_unordered(10) // 最多 10 个并发请求
        .collect()
        .await;

    // 修复 3：收集后解析 —— 根本不需要 mutex！
    for result in &results {
        expensive_parse(result).await;
    }

    results
}
}

调试 Async 代码

Async 堆栈跟踪出了名的难懂 —— 它们显示 executor 的 poll 循环而不是你的逻辑调用链。这些是必备的调试工具。

tokio-console：实时任务检查器

tokio-console 给你一個 htop 风格的视图，显示每个派生任务：它的状态、poll 持续时间、waker 活动和资源使用。

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

#[tokio::main]
async fn main() {
    console_subscriber::init(); // 替代默认 tracing subscriber
    // ... 应用的其余部分
}

然后在另一个终端：

$ RUSTFLAGS="--cfg tokio_unstable" cargo run   # 需要的编译时标志
$ tokio-console                                # 连接到 127.0.0.1:6669

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

tracing crate 理解 Future 生命周期。Spans 跨 .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 跨 .await 保持打开
    info!(email = %user.email, "found user");
    let orders = fetch_orders(user_id).await?;     // 仍然是同一个 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 中获取任务级指标。这是编译时标志，不是运行时标志。

测试 Async 代码

Async 代码引入独特的测试挑战 —— 你需要 runtime、时间控制以及测试并发行为的策略。

使用 #[tokio::test] 的基本 async 测试：

#![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");
}

// 单线程测试（用于 !Send 类型）：
#[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);
}

// 多线程，显式指定 worker 数量：
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_behavior() {
    // 用真实并发测试竞态条件
    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() {
    // 暂停时间 —— sleep() 立即推进，没有真实的挂钟延迟
    time::pause();

    let start = Instant::now();
    time::sleep(Duration::from_secs(3600)).await; // "等待" 1 小时 —— 实际 0ms
    assert!(start.elapsed() >= Duration::from_secs(3600));
    // 测试在几毫秒内运行，而不是一小时！
}

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

    // 测试重试逻辑等待预期的持续时间
    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 退避（指数）
    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 {
            // 模拟慢操作
            time::sleep(Duration::from_secs(10)).await;
            "done"
        }
    ).await;

    assert!(result.is_err()); // 超时
}
}

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

#![allow(unused)]
fn main() {
// 为依赖定义 trait：
trait Storage {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// 生产实现：
struct RedisStorage { /* ... */ }
impl Storage for RedisStorage {
    async fn get(&self, key: &str) -> Option<String> {
        // 真实的 Redis 调用
        todo!()
    }
    async fn set(&self, key: &str, value: String) {
        todo!()
    }
}

// 测试 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);
    }
}

// 被测试函数对 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();

    // 第一次调用：miss → 计算并存储
    let val = cache_lookup(&mock, "key1").await;
    assert_eq!(val, "computed");

    // 第二次调用：hit → 返回存储的值
    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 在这里 drop —— channel 关闭
    });

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

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

关键要点——Common Pitfalls

• 永远不要阻塞 executor —— 对 CPU/同步工作使用 spawn_blocking
• 永远不要跨 .await 持有 MutexGuard —— 严格限定锁的作用域或使用 tokio::sync::Mutex
• 取消立即 drop future —— 对部分操作使用“cancel-safe“模式
• 使用 tokio-console 和 #[tracing::instrument] 调试 async 代码
• 使用 #[tokio::test] 和 time::pause() 测试 async 代码以实现确定性计时

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

13. Production Patterns 🔴

你将学到：

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

优雅关闭

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

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

async fn main_server() {
    // 创建关闭信号 channel
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // 派生服务器
    let server_handle = tokio::spawn(run_server(shutdown_rx.clone()));

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

    // 通知所有任务关闭
    // 注意：为了简洁使用 .unwrap()。生产代码应该处理
    // 所有接收者都被 drop 的情况。
    shutdown_tx.send(true).unwrap();

    // 等待服务器完成（带超时）
    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! {
            // 接受新连接
            conn = accept_connection() => {
                let shutdown = shutdown.clone();
                tokio::spawn(handle_connection(conn, shutdown));
            }
            // 关闭信号
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    println!("Stopping accepting new connections");
                    break;
                }
            }
        }
    }
    // 进行中的连接会自己完成
    // 因为它们有自己的 shutdown_rx clone
}

async fn handle_connection(conn: Connection, mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            request = conn.next_request() => {
                // 完整处理请求 —— 不要在中途放弃
                process_request(request).await;
            }
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    // 完成当前请求，然后退出
                    break;
                }
            }
        }
    }
}

sequenceDiagram
    participant OS as OS Signal
    participant Main as Main Task
    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: Finish current request
    Note over W2: Finish current request

    W1-->>Main: Task complete
    W2-->>Main: Task complete
    Main->>Main: All workers done → exit

有界 Channel 的背压

如果生产者比消费者快，无界 channel 会导致 OOM。在生产中始终使用有界 channel：

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

async fn backpressure_example() {
    // 有界 channel：最多缓冲 100 个 items
    let (tx, mut rx) = mpsc::channel::<WorkItem>(100);

    // 生产者：当缓冲区满时自然减速
    let producer = tokio::spawn(async move {
        for i in 0..1_000_000 {
            // send() 是异步的 —— 如果缓冲区满则等待
            // 这创造了自然背压！
            tx.send(WorkItem { id: i }).await.unwrap();
        }
    });

    // 消费者：按自己的节奏处理 items
    let consumer = tokio::spawn(async move {
        while let Some(item) = rx.recv().await {
            process(item).await; // 慢处理没问题 —— 生产者等待
        }
    });

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

// 与无界对比 —— 危险：
// let (tx, rx) = mpsc::unbounded_channel(); // 没有背压！
// 生产者可以无限填充内存
}

结构化并发：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();

    // 派生一批任务
    for url in get_urls() {
        set.spawn(async move {
            fetch_and_process(url).await
        });
    }

    // 收集所有结果（顺序不保证）
    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}"),
        }
    }

    // 所有任务都完成了 —— 没有悬而未决的后台工作
    println!("Processed {} items", results.len());
}

// TaskTracker (tokio-util 0.7.9+) —— 等待所有派生的任务
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(); // 不再添加任务
    tracker.wait().await; // 等待所有跟踪的任务
    println!("All tasks finished");
}
}

超时和重试

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

// 简单超时
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),
    }
}

// 指数退避重试
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; // 指数退避
            }
        }
    }
    unreachable!()
}

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

生产提示 —— 添加抖动：上面的函数使用纯指数退避，但在生产中， 许多客户端同时失败会在相同时间间隔重试（惊群效应）。 添加随机抖动 —— 例如 sleep(delay + rand_jitter)，其中 rand_jitter 是 0..delay/4 —— 这样重试分散在一段时间内。

Async 代码中的错误处理

Async 引入独特的错误传播挑战 —— 派生的任务创建错误边界、超时错误包装内部错误、? 在 futures 跨任务边界时交互不同。

thiserror vs anyhow —— 选择正确的工具：

#![allow(unused)]
fn main() {
// thiserror：为库和公共 API 定义类型化错误
// 每个变体都是显式的 —— 调用者可以匹配特定错误
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：用于应用程序和原型的快速错误处理
// 包装任何错误 —— 不需要为每种情况定义类型
use anyhow::{Context, Result};

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

    let result = run_gpu_test(&config)
        .await
        .context("GPU diagnostic failed")?;              // 链接上下文

    Ok(())
}
// anyhow 打印："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?)
    });

    // 双重 ?：第一个 ? 解包 JoinError（任务 panic），第二个 ? 解包内部 Result
    let result = handle.await??;
    Ok(result)
}
}

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

#![allow(unused)]
fn main() {
// ❌ 错误上下文跨 spawn 边界丢失：
async fn bad_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work().await  // 返回 Result<T, SomeError>
    });

    // handle.await 返回 Result<Result<T, SomeError>, JoinError>
    // 内部错误没有关于哪个任务失败的上下文
    let result = handle.await??;
    Ok(())
}

// ✅ 在 spawn 边界添加上下文：
async fn good_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work()
            .await
            .context("worker task failed")  // 上下文在跨边界之前
    });

    let result = handle.await
        .context("worker task panicked")??;  // JoinError 也有上下文
    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),                      // 保留内部错误
        Err(_) => Err(DiagError::Timeout(dur)),     // 超时 → 类型化错误
    }
}
}

Tower：中间件模式

Tower crate 定义可组合的 Service trait —— Rust 中 async 中间件的支柱（被 axum、tonic、hyper 使用）：

#![allow(unused)]
fn main() {
// Tower 的核心 trait（简化）：
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)))       // 最外层：超时
    .layer(RateLimitLayer::new(100, Duration::from_secs(1))) // 然后：限流
    .service(my_handler);                                     // 最内层：你的代码
}

为什么这很重要：如果你用过 ASP.NET 中间件或 Express.js 中间件，Tower 是 Rust 等价物。这是生产级 Rust 服务添加横切关注点而不代码重复的方式。

练习：带工作池的优雅关闭

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,
                }
            }
        }));
    }

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

    // 在 Ctrl+C 时：发送关闭信号，等待工人
    // 注意：为了简洁使用 .unwrap() —— 生产环境中处理错误。
    tokio::signal::ctrl_c().await.unwrap();
    shutdown_tx.send(true).unwrap();
    for h in handles { let _ = h.await; }
    println!("Shut down cleanly.");
}

关键要点——Production Patterns

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

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

14. Async Is an Optimization, Not an Architecture 🔴

你将学到：

• 为什么 async 往往会污染整个代码库——以及为什么这是设计缺陷，不是特性
• “Sync core, async shell“模式让大多数代码可测试、可调试
• 如何处理困难情况：也需要 I/O 的逻辑
• spawn_blocking 何时是修复方案 vs. 症状
• async 何时真正属于你的核心逻辑
• 为什么 sync-first 库比 async-first 库更具可组合性

你已经花了 13 章学习 Rust 异步。但这本书还没告诉你的最重要的事情是：你的大多数代码不应该是 async。

函数染色问题

Bob Nystrom 的 “What Color is Your Function?” 识别了核心问题：async 函数可以调用 sync 函数，但 sync 函数不能调用 async 函数。一旦一个函数变成 async，调用链中它上面的所有函数都必须跟随。

在 Rust 中这比 C# 或 JavaScript 更糟，因为 async 不仅感染函数签名——它还感染类型：

每一行都是一个必须做出、做对并维护的决定——而这些都与业务逻辑无关。业界正在远离这种模式：Java 的 Project Loom（虚拟线程）和 Go 的 goroutine 都让你编写看起来同步的代码，由 runtime 廉价地多路复用。Rust 选择显式 async 来实现零成本控制，但这种控制有复杂度成本，应该有意识地支付，而不是默认支付。

“但线程很昂贵”

反射式的反驳：“我们需要 async 因为线程很昂贵。“在大多数团队运营的规模上，这大多是错误的。

• 栈内存： 每个 OS 线程预留 8MB 虚拟地址空间（Linux 默认），但 OS 只提交触及的页面——一个基本空闲的线程使用 20-80KB 物理内存。
• 上下文切换： 现代硬件上约 1-5µs。在 50 个并发请求下，这是噪音。在 100K 次切换/秒时，这是可测量的。
• 创建成本： Linux 上每个线程约 10-30µs。线程池（rayon、std::thread::scope）将此摊销到零。

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

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

一个简单的纯函数——fn add(a: i32, b: i32) -> i32——显然不需要 async。这不是一个有趣的教训。有趣的情况是当业务逻辑似乎需要在中间进行 I/O：检查库存的验证、查询汇率的定价、查找客户的订单管道。

考虑一个订单处理服务。async-everywhere 版本看起来自然：

版本 A：Async 贯穿核心

#![allow(unused)]
fn main() {
// orders.rs — async 一路向下

pub async fn process_order(order: Order) -> Result<Receipt, OrderError> {
    // 步骤 1：验证——纯业务规则，无 I/O
    validate_items(&order)?;
    validate_quantities(&order)?;

    // 步骤 2：检查库存——需要数据库调用
    let stock = inventory_client.check(&order.items).await?;
    if !stock.all_available() {
        return Err(OrderError::OutOfStock(stock.missing()));
    }

    // 步骤 3：计算定价——纯数学，但 async 因为我们已经在这里
    let pricing = calculate_pricing(&order, &stock);

    // 步骤 4：应用折扣——需要外部服务调用
    let discount = discount_service.lookup(order.customer_id).await?;
    let final_price = pricing.apply_discount(discount);

    // 步骤 5：格式化收据——纯
    Ok(Receipt::new(order, final_price))
}
}

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

版本 B：Sync Core, Async Shell

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

#![allow(unused)]
fn main() {
// core.rs — 纯业务逻辑，零 async，零 tokio 依赖

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 — 薄 async 编排器
//
// 注意：网络调用上的 `?` 需要 `impl From<reqwest::Error> for OrderError`
//（或统一的错误 enum）。参见 ch12 了解 async 错误处理模式。

use crate::core;

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

    // Async：获取库存（这是 shell 的工作）
    let stock = inventory_client.check(&validated.items).await?;

    // Sync：将业务规则应用于获取的数据
    let stocked = core::check_stock(&validated, &stock)?;

    // Async：获取折扣
    let discount = discount_service.lookup(order.customer_id).await?;

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

async shell 是一个 fetch → decide → fetch → decide 的管道。每个 “decide” 步骤是一个 sync 函数，它将 I/O 结果作为输入而不是自己去获取。

测试差异

Sync core 测试每个业务规则而不需要 runtime 或 mocks：

#![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)]); // 只有 3 个可用

    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); // 价格（美分）
    let receipt = core::finalize(&order, Discount::Percent(15));
    assert_eq!(receipt.final_price, 85_00);
}
}

Async shell 获得一个更薄的集成测试来验证连接，而不是逻辑：

#![allow(unused)]
fn main() {
#[tokio::test]
async fn process_order_integration() {
    let mock_inventory = mock_service(/* 返回库存 */);
    let mock_discounts = mock_service(/* 返回 10% */);
    let receipt = process_order(sample_order()).await.unwrap();
    assert!(receipt.final_price > 0);
    // 逻辑正确性已经由上面的 core 测试证明
}
}

为什么这很重要

关键见解：步骤 2 和 4 中的 I/O 调用不需要在业务逻辑内部。它们是它的输入。 Sync core 将 StockResult 和 Discount 作为参数。这些值来自哪里——HTTP、gRPC、测试夹具、缓存——是 shell 的事情。

spawn_blocking 气味

第 12 章介绍了 spawn_blocking 作为意外阻塞 executor 的修复方案。当你有一个临时的阻塞调用时，这是正确的修复——std::fs::read、压缩库、传统 FFI 函数。

但如果你发现自己将大块代码包装在 spawn_blocking 中：

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // 如果这是你的代码库，边界在错误的地方
    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()
}
}

…这是代码库在告诉你：这个逻辑从来就不是 async。 你不需要 spawn_blocking——你需要一个 async handler 直接调用的 sync 模块：

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // validate → enrich → process → format 都是 sync。
    // 不需要 spawn_blocking——它们很快且 CPU 轻量。
    let response = my_core::handle(req);
    response
}
}

保留 spawn_blocking 用于真正繁重的 CPU 工作（解析大型负载、图像处理、压缩），此时时间成本实际上会使 executor 饥饿。对于在微秒内运行的普通业务逻辑，直接 sync 调用更简单且正确。

库：Sync First, Async Wrapper Optional

边界问题对库作者影响更大。一个 sync 库可以被 sync 和 async 调用者使用：

// 一个 sync 库——到处可用
let report = my_lib::analyze(&data);

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

// 调用者 B：async handler，工作正常
async fn handler() -> Json<Report> {
    let report = my_lib::analyze(&data); // async 上下文中的 sync 调用——没问题
    Json(report)
}

// 调用者 C：繁重的分析——调用者决定卸载
async fn handler_heavy() -> Json<Report> {
    let data = data.clone();
    let report = tokio::task::spawn_blocking(move || {
        my_lib::analyze(&data) // 调用者控制 async 边界
    }).await.unwrap();
    Json(report)
}

一个 async 库强制所有调用者进入 runtime：

#![allow(unused)]
fn main() {
// 一个 async 库——只能从 async 上下文使用
let report = my_lib::analyze(&data).await; // 调用者*必须*是 async

// Sync 调用者？现在你需要 block_on——并希望内部没有 nested runtime
let report = tokio::runtime::Runtime::new().unwrap().block_on(
    my_lib::analyze(&data)
); // 脆弱，如果已经在 runtime 内则容易 panic
}

默认使用 sync API。 如果你的库做纯计算、数据转换或解析，没有理由让它变成 async。如果它做 I/O，考虑提供一个 sync core 并在特性标志后提供一个可选的 async 便利层——让调用者拥有边界决定。

Async 何时属于核心

不是所有事情都能干净地分离。当以下情况时，Async 属于你的核心逻辑：

• Fan-out/fan-in 就是逻辑。 如果你的业务规则是“并发查询 5 个定价服务并返回最便宜的“，并发就是逻辑，不是管道。强制通过 sync + 线程是重新发明一个更糟的 async。

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

• 长持有状态连接。 WebSocket 处理器、gRPC 双向流和协议状态机具有与 I/O 事件固有绑定的状态转换。ch17 中的综合项目——一个异步聊天服务器——正是这种情况：并发连接、基于房间的 fan-out 和优雅关闭是根本上的 async 工作。

测试： 如果从函数中移除 async 需要用线程、channel 或手动轮询替换它，那么 async 在发挥作用。如果移除 async 只是删除关键字而没有其他更改，它从来就不需要是 async。

决策规则

graph TD
    START["这个函数应该是 async 吗？"] --> IO{"它做 I/O 吗？"}
    IO -->|否 | SYNC["sync fn —— 总是"]
    IO -->|是 | BOUNDARY{"它在边界吗？<br/>handler, main loop, accept()"}
    BOUNDARY -->|是 | ASYNC_SHELL["async fn —— 这是 shell"]
    BOUNDARY -->|否 | CORE_IO{"I/O 是核心逻辑吗？<br/>fan-out, streaming, stateful conn"}
    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

经验法则： 从 sync 开始。只在最外层的 I/O 边界添加 async。只有当你能阐明哪些并发 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> {
    // 通过 HTTP 从设备获取原始遥测
    let raw = reqwest::get(format!("http://bmc-{device_id}/telemetry"))
        .await?
        .json::<RawTelemetry>()
        .await?;

    // 业务逻辑：将原始传感器读数转换为校准值
    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(),
        });
    }

    // 业务逻辑：分类设备健康状态
    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 };

    // 从库存服务获取设备元数据
    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 — 零 async 依赖

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 边界

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 — 不需要 runtime

// 测试夹具辅助函数——构造数据而无任何 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. Async 是I/O 多路复用优化，不是应用架构。大多数业务逻辑是 sync。
2. Sync core, async shell： 将业务规则保存在纯 sync 函数中，这些函数将 I/O 结果作为参数。Async shell 编排获取并调用核心。
3. 如果你在大块代码周围包装 spawn_blocking，边界在错误的地方——改为将逻辑重构为 sync 模块。
4. 库应该默认使用 sync API。 一个 async 库强制所有调用者进入 runtime；一个 sync 库让调用者拥有 async 边界。
5. Async 在fan-out/fan-in、streaming 和状态连接中赚取其价值——这些情况下并发就是业务逻辑。

另见： Ch12 — 常见陷阱（spawn_blocking 作为战术修复）· Ch13 — 生产级模式（背压、结构化并发）· Ch17 — 综合项目：异步聊天服务器（一个 async 是正确架构的案例）

15. Exercises

练习

练习 1：Async 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) 确保最多 5 个 futures 并发 poll
    // —— 这里不需要单独的 Semaphore。
    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) // ← 这 alone 限制并发度为 5
        .collect()
        .await;

    results
}

// 注意：当你需要在独立派生的任务（tokio::spawn）之间限制并发度时使用 Semaphore。
// 当处理 stream 时使用 buffer_unordered。不要为相同的限制结合两者。
}

练习 3：带工作池的优雅关闭

构建一个任务处理器，具有：

• 基于 channel 的工作队列
• N 个工作任务从队列消费
• 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);

    // 派生 4 个工作者
    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; // 模拟工作
                        println!("Worker {id}: done with item {}", work.id);
                    }
                    None => {
                        println!("Worker {id}: channel closed, exiting");
                        break;
                    }
                }
            }
        });
        worker_handles.push(handle);
    }

    // 生产者：提交一些工作
    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;
        }
    });

    // 等待 Ctrl+C
    tokio::signal::ctrl_c().await.unwrap();
    println!("\nShutdown signal received!");
    shutdown_tx.send(true).unwrap();
    producer.abort(); // 取消生产者任务

    // 等待工作者完成
    for handle in worker_handles {
        let _ = handle.await;
    }
    println!("All workers shut down. Goodbye!");
}

练习 4：从零开始构建简单的 Async Mutex

实现一个 async 感知的 mutex，使用 channel（不使用 tokio::sync::Mutex）。

提示：使用带有 1 个 permit 的 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>,
}

// 安全性：对 T 的访问被 semaphore 序列化（最多 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, // 在 guard drop 时释放——释放锁
}

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 {
        // 安全性：我们持有唯一的 semaphore permit，所以没有其他
        // SimpleGuard 存在——保证独占访问。
        unsafe { &*self.data.get() }
    }
}

impl<T> std::ops::DerefMut for SimpleGuard<T> {
    fn deref_mut(&mut self) -> &mut T {
        // 安全性：同样的推理——单个 permit 保证独占性。
        unsafe { &mut *self.data.get() }
    }
}

// 当 SimpleGuard 被 drop 时，_permit 被 drop，
// 这释放 semaphore permit——另一个 lock() 可以进行。

// 用法：
// let mutex = SimpleAsyncMutex::new(vec![1, 2, 3]);
// {
//     let mut guard = mutex.lock().await;
//     guard.push(4);
// } // permit 在这里释放
}

练习 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)
        // 步骤 2：过滤偶数
        .filter(|x| futures::future::ready(x % 2 == 0))
        // 步骤 3：平方每个
        .map(|x| x * x)
        // 步骤 4：并发处理（模拟 async 工作）
        .map(|x| async move {
            sleep(Duration::from_millis(50)).await;
            println!("Processed: {x}");
            x
        })
        .buffer_unordered(10) // 10 个并发
        // 步骤 5：收集
        .collect()
        .await;

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

练习 6：用 Timeout 实现 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, // 来自第 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> {
        // 检查主 future 是否完成
        if let Poll::Ready(val) = Pin::new(&mut self.future).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

        // 检查定时器是否到期
        if let Poll::Ready(()) = Pin::new(&mut self.timer).poll(cx) {
            return Poll::Ready(Either::Right(()));
        }

        Poll::Pending
    }
}

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

Summary and Reference Card

快速参考卡

Async 思维模型

┌─────────────────────────────────────────────────────┐
│  async fn → 状态机（enum） → impl Future            │
│  .await   → poll() 内部 future                       │
│  executor → loop { poll(); sleep_until_woken(); }   │
│  waker    → "嘿 executor，再 poll 我一次"             │
│  Pin      → "承诺我不在内存中移动"                   │
└─────────────────────────────────────────────────────┘

常见模式速查表

Pinning 快速参考

Channel 选择指南

Mutex 选择指南

决策快速参考

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

选择 runtime？
├── 服务器应用 → tokio
├── 库 → runtime 无关（futures crate）
├── 嵌入式 → embassy
└── 最小化 → smol

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

常见错误信息和修复

进一步阅读

Async Rust 训练指南结束

Capstone Project: Async Chat Server

本项目将整本书的模式集成到一个生产级应用程序中。你将构建一个多房间异步聊天服务器，使用 tokio、channel、stream、优雅关闭和适当的错误处理。

估计时间：4-6 小时 | 难度：★★★

你将练习：

• tokio::spawn 和 'static 要求（第 8 章）
• Channel：mpsc 用于消息、broadcast 用于房间、watch 用于关闭（第 8 章）
• Stream：从 TCP 连接读取行（第 11 章）
• 常见陷阱：取消安全、跨 .await 的 MutexGuard（第 12 章）
• 生产级模式：优雅关闭、背压（第 13 章）
• Async 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/>广播 channel"]
    SERVER --> R2["#rust<br/>广播 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 接受循环

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

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：带广播 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 消息缓冲
            tx
        })
        .clone()
}
}

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

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

#![allow(unused)]
fn main() {
loop {
    tokio::select! {
        // 客户端发送给我们一行
        result = reader.read_line(&mut line) => {
            match result {
                Ok(0) | Err(_) => break,
                Ok(_) => {
                    // 解析命令或广播消息
                }
            }
        }
        // 房间广播接收
        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);

// 在接受循环中：
loop {
    tokio::select! {
        result = listener.accept() => {
            let (socket, addr) = result?;
            // 派生客户端任务，带 shutdown_rx.clone()
        }
        _ = tokio::signal::ctrl_c() => {
            println!("Shutdown signal received");
            shutdown_tx.send(true)?;
            break;
        }
    }
}
}

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

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

生产级强化服务器：

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

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

// 用超时包装读取：
match timeout(Duration::from_secs(300), reader.read_line(&mut line)).await {
    Ok(Ok(0)) | Ok(Err(_)) | Err(_) => break, // EOF、错误或超时
    Ok(Ok(_)) => { /* 处理行 */ }
}
}

步骤 6：集成测试

编写一个测试，启动服务器，连接两个客户端，并验证消息传递：

#![allow(unused)]
fn main() {
#[tokio::test]
async fn two_clients_can_chat() {
    // 在后台启动服务器
    let server = tokio::spawn(run_server("127.0.0.1:0")); // 端口 0 = OS 选择

    // 连接两个客户端
    let mut client1 = TcpStream::connect(addr).await.unwrap();
    let mut client2 = TcpStream::connect(addr).await.unwrap();

    // 客户端 1 发送消息
    client1.write_all(b"Hello from client 1\n").await.unwrap();

    // 客户端 2 应该接收到它
    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 用于加密连接