练习与自测

本章练习共 11 题，答案折叠在每题下方。建议先自己写、编译通过后再展开答案；难度标记：★☆☆ 基础 / ★★☆ 综合 / ★★★ 挑战。

凡涉及 Tokio 的题目，答案里都标了依赖与 #[tokio::main]；答案代码都可直接放进 src/main.rs 运行（多文件题给了目录结构）。

练习 1：不用运行时手写最小 Future

难度：★☆☆

要求：写一个 async fn 返回 u32，在 main 里不用任何运行时（只用 std） 拿到值。要求说明：为什么不能直接调用 .await，以及「不 await 就不执行」怎么被验证。 提示：future 需要一个 executor；先想想 futures::executor::block_on 是什么角色。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：无（纯 std）。目标：看清「async fn 返回值是 future」这件事
fn main() {
    fn assert_future<F: std::future::Future<Output = u32>>(_f: F) {}
    assert_future(answer());
    println!("ok");
}

async fn answer() -> u32 {
    println!("这一行只有被 poll 时才会执行");
    42
}

输出：只有 ok——println! 那行没有打印，因为 future 从未被 poll。

要点解析：

• main 不是 async fn，所以里面不能写 .await（error[E0728]: awaitis only allowed inside ofasync blocks and functions）。要拿到值必须有一个 executor 去 poll。
• 三种合法做法：#[tokio::main]、futures::executor::block_on(answer())、 自己写「最小 block_on：自己写一个 executor」里那个。
• 「不 await 就不执行」的验证方法：在 async fn 体里放 println!，只构造不 await， 观察输出缺失；再把返回值赋给变量并 drop，观察连警告都没有（见练习 10）。

练习 2：判断 async 代码能否编译

难度：★☆☆

要求：下面的代码能编译吗？如果编译失败，第一个错误是什么？请只靠阅读回答，再实际编译验证。

// 依赖：无
use std::future::Future;

async fn get() -> String {
    String::from("hi")
}

fn main() {
    let fut: impl Future<Output = String> = get();
    println!("{fut:?}");
}

提示：impl Trait 能放在 let 的位置吗？Future 有 Debug 吗？

参考答案（先自己写再看）

参考答案（先自己写再看）

不能编译，有两个错误：

error[E0562]: `impl Trait` is not allowed in the type of variable bindings
 --> src/main.rs:9:14
  |
9 |     let fut: impl Future<Output = String> = get();
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |
  = note: `impl Trait` is only allowed in arguments and return types of
    functions and methods

error[E0277]: `impl Future<Output = String>` doesn't implement `Debug`
 --> src/main.rs:10:20
  |
10 |     println!("{fut:?}");
   |                    ^^^ `impl Future<Output = String>` cannot be formatted
     using `{:?}` because it doesn't implement `Debug`

修法：把 future 交给 executor，而不是试图打印它。

// 依赖：tokio = { version = "1", features = ["full"] }
#[tokio::main]
async fn main() {
    let fut = get();   // 类型由编译器推断，不写 impl Trait
    println!("{}", fut.await);
}

async fn get() -> String {
    String::from("hi")
}

输出：hi。

要点解析：

• impl Trait 在返回值位置是「存在类型」（opaque type），在 let 里没有意义， 编译器直接拒绝。想写具体类型就用 Pin<Box<dyn Future<Output = String>>>。
• 匿名 future 不实现 Debug（async 生成的类型不派生任何 trait）， 所以不能 {:?} 打印。要调试优先打印 future 的输出，而不是 future 本身。
• 这也解释了为什么「在结构体里存 future」必须指定 Pin<Box<dyn Future<...>>>： 匿名类型无法出现在字段声明里。

练习 3：用 sleep 对比串行与并发请求

难度：★★☆

要求：用 tokio::time::sleep 模拟三次「网络请求」（各 100ms），分别实现串行与并发版本， 打印总耗时，并解释差距来自哪里。 提示：并发版本用 tokio::join!，注意三次 sleep 要同时被轮询。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：tokio = { version = "1", features = ["full"] }
use std::time::{Duration, Instant};

async fn request(name: &str) -> String {
    // 100ms 表示「等网络」，期间任务挂起，不占 CPU
    tokio::time::sleep(Duration::from_millis(100)).await;
    format!("{name}: 200 OK")
}

async fn serial() -> u64 {
    let start = Instant::now();
    let a = request("A").await;
    let b = request("B").await;
    let c = request("C").await;
    println!("{a} / {b} / {c}");
    start.elapsed().as_millis() as u64
}

async fn concurrent() -> u64 {
    let start = Instant::now();
    // join! 让三个 future 在同一个任务里被并发轮询：
    // 三个 sleep 同时计时，所以总耗时 ≈ 单个耗时
    let (a, b, c) = tokio::join!(request("A"), request("B"), request("C"));
    println!("{a} / {b} / {c}");
    start.elapsed().as_millis() as u64
}

#[tokio::main]
async fn main() {
    println!("串行耗时: {}ms", serial().await);
    println!("并发耗时: {}ms", concurrent().await);
}

输出：A: 200 OK / B: 200 OK / C: 200 OK 两遍， 串行耗时: 300ms（±10ms）、并发耗时: 100ms（±10ms）。

要点解析：

• .await 是等待点，不是「阻塞」：request("A").await 让出执行权后，request("B") 可以先跑到自己的 sleep，于是三个计时器同时开始。
• 串行写法里 .await 一个接一个，第二个请求的 sleep 在第一个完成前根本不会被创建。
• 这是 async 最核心的价值：用一条线程把等待时间重叠起来。
• 想强调差异可以把 100ms 改成 1000ms；耗时应该是 3000ms vs 1000ms。

练习 4：用 tokio::spawn 并发抓取

难度：★★☆

要求：用 spawn 并发「抓取」5 个「URL」（用 sleep 模拟不同耗时），要求： 每个任务返回 (String, u64)（URL 与耗时），全部完成后打印总耗时与每个结果。 提示：JoinHandle 自己也是 future；Vec<JoinHandle<_>> 可以在循环里逐个 await。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：tokio = { version = "1", features = ["full"] }
use std::time::{Duration, Instant};

async fn fetch(url: &'static str, ms: u64) -> (String, u64) {
    tokio::time::sleep(Duration::from_millis(ms)).await;
    (format!("body of {url}"), ms)
}

#[tokio::main]
async fn main() {
    let jobs = [
        ("https://a.example", 120u64),
        ("https://b.example", 40),
        ("https://c.example", 200),
        ("https://d.example", 80),
        ("https://e.example", 160),
    ];

    let start = Instant::now();
    // 每个 URL 一个任务：任务可以跨线程真并行，因此要求 move + 'static
    let handles: Vec<_> = jobs
        .iter()
        .map(|&(url, ms)| tokio::spawn(fetch(url, ms)))
        .collect();

    for handle in handles {
        let (body, ms) = handle.await.expect("task panicked");
        println!("{body}（{ms}ms）");
    }
    println!("总耗时 {}ms（最长任务 200ms）", start.elapsed().as_millis());
}

输出：5 行 body of ...（Nms），最后 总耗时 200ms（最长任务 200ms）（±10ms）。

要点解析：

• tokio::spawn(fetch(url, ms)) 直接把返回的 future 交给运行时；jobs 是 &'static str 数组，所以即使不加 move 也满足 'static。
• JoinHandle<T> 本身是 future，循环里 .await 是按顺序收集结果， 但任务早已并发跑完，收集顺序不影响总耗时。
• 用 join! 也能达到同样的总耗时，区别是 spawn 让每个请求成为独立任务 （可被单独取消、panic 被隔离到 JoinError），代价是要求 'static + Send。
• 若耗时之和明显大于 200ms，说明某个任务实际是串行的——常见原因是在循环里 spawn 之后立刻 .await，那样就退化成串行了。

练习 5：用 timeout 做超时控制

难度：★★☆

要求：给一个「可能很慢的」操作加 200ms 超时，超时打印 timeout，否则打印结果。 再补一个版本：超时后不丢弃结果，而是让后台任务继续跑完并在之后取回（提示：spawn + timeout）。 提示：tokio::time::timeout 返回 Result；超时时内层 future 被 drop。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：tokio = { version = "1", features = ["full"] }
use std::time::Duration;
use tokio::time::{timeout, Elapsed};

async fn slow(ms: u64) -> String {
    tokio::time::sleep(Duration::from_millis(ms)).await;
    format!("done in {ms}ms")
}

#[tokio::main]
async fn main() {
    // 版本 1：真超时——超时后 future 被 drop，结果永久丢失
    match timeout(Duration::from_millis(200), slow(500)).await {
        Ok(v) => println!("版本1 完成: {v}"),
        Err(Elapsed { .. }) => println!("版本1 timeout（slow 已被 drop）"),
    }

    // 版本 2：不丢结果——先把任务 spawn 出去，再对 JoinHandle 加超时
    let handle = tokio::spawn(slow(500));
    match timeout(Duration::from_millis(200), &mut handle).await {
        Ok(Ok(v)) => println!("版本2 及时完成: {v}"),
        Ok(Err(join_err)) => println!("版本2 任务失败: {join_err}"),
        Err(_) => {
            println!("版本2 超时，但后台任务还在跑");
            // 注意：这里 handle 是 &mut，timeout 只取消"等待"，不取消任务
            let v = handle.await.expect("task panicked");
            println!("版本2 稍后取回: {v}");
        }
    }
}

输出：版本1 timeout（slow 已被 drop），然后 版本2 超时，但后台任务还在跑、 约 300ms 后 版本2 稍后取回: done in 500ms。

要点解析：

• timeout 是「给 future 加时限」，超时后内层 future 被 drop——这就是取消。 已经从流里读走的数据、已经发出的请求都不会回退（取消安全，见「select! 的语义与取消安全」）。
• 版本 2 的关键是把「任务」和「等待」分开：spawn 出的任务的生命周期独立于 handle，所以超时只取消 handle.await 这一层等待。
• 想取消后台任务要显式 handle.abort()，且被 abort 时任务可能在任意挂起点被 drop。
• timeout(Duration::from_millis(200), &mut handle) 能这样写是因为 &mut JoinHandle 也实现了 Future（JoinHandle: Unpin），超时后还能继续用同一个 handle。

练习 6：用 Semaphore 限制并发数

难度：★★☆

要求：用 tokio::sync::Semaphore 限制并发数为 3，启动 10 个任务，每个任务打印 「开始/结束 + 自己的编号」，观察任意时刻最多 3 个任务在跑。 提示：Arc<Semaphore> + acquire_owned()，permit 要活到任务结束。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：tokio = { version = "1", features = ["full"] }
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Semaphore;
use tokio::time::sleep;

async fn job(id: u32) {
    println!("任务 {id} 开始");
    sleep(Duration::from_millis(100)).await;
    println!("任务 {id} 结束");
}

#[tokio::main]
async fn main() {
    let start = Instant::now();
    let sem = Arc::new(Semaphore::new(3));  // 最多 3 个并发

    let mut handles = Vec::new();
    for id in 0..10 {
        let sem = Arc::clone(&sem);
        handles.push(tokio::spawn(async move {
            // acquire_owned 需要 Arc<Semaphore>，permit 因此是 'static 的，
            // 可以安全地活在整个任务里
            let permit = sem.acquire_owned().await.expect("semaphore closed");
            job(id).await;
            drop(permit);   // 显式释放；其实函数结束也会 drop
        }));
    }

    for h in handles {
        h.await.expect("task panicked");
    }
    // 10 个任务、每个 100ms、每次最多 3 个 → 4 批 → 约 400ms
    println!("总耗时 {}ms", start.elapsed().as_millis());
}

输出：10 行「开始/结束」交错，任意时刻最多 3 对；总耗时 400ms（±20ms）。

要点解析：

• Semaphore::acquire_owned 返回 OwnedSemaphorePermit，它拥有一个 Arc<Semaphore> 的克隆，因此不需要生命周期参数，能直接跨线程移动进 'static 任务。 这也是为什么要把信号量放进 Arc。
• 不写 drop(permit) 也可以：permit 在任务结束时随作用域自动释放。
• 一定要在拿到 permit 之后再 spawn 之外做重活；如果先做重活再等 permit， 限流就形同虚设。
• 替代方案：futures::stream::iter(0..10).map(job).buffer_unordered(3).collect::<Vec<_>>().await， 一行搞定同样的事，见练习 11。

练习 7：实现 TCP 回声服务器

难度：★★★

要求：实现一个完整的 TCP 回声服务器（监听 127.0.0.1:8080，每个连接一个任务）， 再写一个客户端连上去发 3 行并读回 3 行。要求给出目录结构与 Cargo.toml。 提示：TcpListener::accept + tokio::spawn；按行处理用 BufReader::lines。

参考答案（先自己写再看）

参考答案（先自己写再看）

echo/
├── Cargo.toml
└── src/
    ├── server.rs
    └── client.rs

# Cargo.toml
[package]
name = "echo"
version = "0.1.0"
edition = "2024"

[[bin]]
name = "server"
path = "src/server.rs"

[[bin]]
name = "client"
path = "src/client.rs"

[dependencies]
tokio = { version = "1", features = ["full"] }

// src/server.rs：每个连接一个任务，互不阻塞
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("listening on {}", listener.local_addr()?);

    loop {
        let (stream, peer) = listener.accept().await?;
        println!("connected: {peer}");
        tokio::spawn(async move {
            if let Err(e) = echo(stream).await {
                eprintln!("{peer} error: {e}");
            }
            println!("disconnected: {peer}");
        });
    }
}

async fn echo(stream: TcpStream) -> std::io::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut lines = BufReader::new(read_half).lines();

    // next_line 返回 None 表示对端关闭（EOF）
    while let Some(line) = lines.next_line().await? {
        write_half.write_all(line.as_bytes()).await?;
        write_half.write_all(b"\n").await?;
        write_half.flush().await?;
    }
    Ok(())
}

// src/client.rs：发 3 行，读回 3 行
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let stream = TcpStream::connect("127.0.0.1:8080").await?;
    let (read_half, mut write_half) = stream.into_split();

    for msg in ["ping", "rust", "bye"] {
        write_half.write_all(msg.as_bytes()).await?;
        write_half.write_all(b"\n").await?;
    }
    write_half.flush().await?;
    // 关掉写半边，服务器才会收到 EOF 并退出循环
    drop(write_half);

    let mut lines = BufReader::new(read_half).lines();
    while let Some(line) = lines.next_line().await? {
        println!("echo: {line}");
    }
    Ok(())
}

运行：终端 1 cargo run --bin server，终端 2 cargo run --bin client。 客户端输出 echo: ping、echo: rust、echo: bye；服务端打印连接/断开日志。

要点解析：

• TcpListener::accept 是异步的：没有新连接时任务挂起，线程可以去跑别的任务。 所以一个 loop + spawn 就撑住了上万个连接。
• into_split() 把 TcpStream 拆成可独立移动的读写两半（内部是 Arc）， 这样「读任务」和「写任务」可以分开；也可以用 tokio::io::split 处理任意 AsyncRead + AsyncWrite。
• 忘记 flush 常见于「数据卡在缓冲区」的 bug；write_all 只保证写进缓冲。
• 客户端 drop(write_half) 会发送 FIN，服务器 next_line 才会返回 None。 不 drop 就会出现「双方都在等对方说话」的死锁。
• 出错时不要让整个服务器退出：把 handle 的错误打印出来即可， 这正是 tokio::spawn 与「一连接一线程 + 全局 panic」相比更健壮的地方。

练习 8：tokio::spawn 与 Send 边界

难度：★★☆

要求：下面两个函数都能通过编译，但只有一个能安全地被 tokio::spawn。指出是哪一个， 说明为什么，并给出两种修法。

// 依赖：tokio = { version = "1", features = ["full"] }
use std::sync::{Arc, Mutex};
use tokio::sync::Mutex as AsyncMutex;

async fn a(lock: &Mutex<u32>) -> u32 {
    let g = lock.lock().unwrap();
    tokio::task::yield_now().await;
    *g
}

async fn b(lock: &AsyncMutex<u32>) -> u32 {
    let g = lock.lock().await;
    tokio::task::yield_now().await;
    *g
}

#[tokio::main]
async fn main() {
    let m1 = Arc::new(Mutex::new(1));
    let m2 = Arc::new(AsyncMutex::new(2));
    // tokio::spawn(a(&m1));
    // tokio::spawn(b(&m2));
}

提示：看跨 .await 存活的类型是否 Send，以及 &Mutex<u32> 的 'static 问题。

参考答案（先自己写再看）

参考答案（先自己写再看）

两个都不能，而且失败原因不同：

error[E0277]: `std::sync::MutexGuard<'_, u32>` cannot be sent between threads safely
   |
   = help: within `impl Future<Output = u32>`, the trait `Send` is not implemented
     for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |     let g = lock.lock().unwrap();
   |         - has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
   |     tokio::task::yield_now().await;
   |                      ^^^^^ await occurs here, with `g` maybe used later

error[E0521]: borrowed data escapes outside of function
   |
   |     tokio::spawn(b(&m2));
   |                  ^^^^^^^ `m2` is borrowed here, but the function requires
   |                          `'static` data

• a：std::sync::MutexGuard 不是 Send（它绑定在特定线程上）， 跨 .await 存活就让整个 future 变成 !Send。而且 lock: &Mutex<u32> 是借用， 也不满足 'static。
• b：tokio::sync::MutexGuard 是 Send，Send 这一关能过， 但仍然因为 &AsyncMutex<u32> 是借用而不满足 'static。

两种修法：

// 依赖：tokio = { version = "1", features = ["full"] }
use std::sync::{Arc, Mutex};
use tokio::sync::Mutex as AsyncMutex;

// 修法 1：缩小临界区，让守卫在任何 await 之前 drop —— 最佳实践
async fn a_fixed(lock: Arc<Mutex<u32>>) -> u32 {
    let snapshot = {
        let mut g = lock.lock().unwrap();
        *g += 1;
        *g
    };                          // 守卫在此 drop
    tokio::task::yield_now().await;   // 现在跨 await 的只有一个 u32
    snapshot
}

// 修法 2：确实需要跨 await 持锁时，用 tokio 的锁 + Arc（满足 'static）
async fn b_fixed(lock: Arc<AsyncMutex<u32>>) -> u32 {
    let mut g = lock.lock().await;
    tokio::task::yield_now().await;
    *g += 1;
    *g
}

#[tokio::main]
async fn main() {
    let m1 = Arc::new(Mutex::new(1));
    let m2 = Arc::new(AsyncMutex::new(2));
    let h1 = tokio::spawn(a_fixed(Arc::clone(&m1)));
    let h2 = tokio::spawn(b_fixed(Arc::clone(&m2)));
    println!("{} {}", h1.await.expect("join failed"), h2.await.expect("join failed"));
}

输出：2 3。

要点解析：

• 两条独立的规则要同时满足：Send（跨线程）与 'static（生命周期）。 只修一个仍会报错，这一点最容易漏。
• 为什么 std::sync::MutexGuard 是 !Send？因为它依赖「解锁发生在同一线程」， 而 Windows 的 SRWLOCK 等实现有线程亲和性；Send 掉它可能导致未定义行为。
• tokio::sync::MutexGuard 是 Send，但持锁跨 await 有死锁风险： 临界区里 await 的东西如果又需要同一把锁（直接或间接），任务会永久挂起。 优先级：缩小临界区 > tokio::sync::Mutex。
• 若临界区不可能 await，直接继续用 std::sync::Mutex（更快），不要无脑换成 tokio 版本。

练习 9：递归 async fn 与 Box::pin

难度：★★☆

要求：写一个 async fn 递归函数 sum_to(n: u64) -> u64（返回 1+2+...+n）， 要求它能在 tokio 里被 await。先用普通递归写一次并记录编译错误，再用正确写法实现。 提示：error[E0733]；修法是引入间接（Box::pin）。

参考答案（先自己写再看）

参考答案（先自己写再看）

错误写法与真实诊断：

error[E0733]: recursion in an async fn requires boxing
 --> src/main.rs:1:1
  |
1 | async fn sum_to(n: u64) -> u64 {
  | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
...
5 |         n + sum_to(n - 1).await
  |             ------------------ recursive call here
  |
  = note: a recursive `async fn` call must introduce indirection such as
    `Box::pin` to avoid an infinitely sized future

正确写法（纯 std，不需要任何运行时就能通过类型检查）：

// 依赖：无
use std::future::Future;
use std::pin::Pin;

// 返回装箱的 future：大小固定为一个指针，打破无限递归的类型
fn sum_to(n: u64) -> Pin<Box<dyn Future<Output = u64> + Send>> {
    Box::pin(async move {
        if n == 0 {
            0
        } else {
            n + sum_to(n - 1).await
        }
    })
}

fn main() {
    // 这里只交出 future、不驱动它；真正 await 需要一个 executor（如
    // #[tokio::main] 或 futures::executor::block_on）
    fn assert_future<F: Future<Output = u64>>(_f: F) {}
    assert_future(sum_to(100));
    println!("future 构造完成，等待被 executor 驱动");
}

输出：future 构造完成，等待被 executor 驱动。放进 #[tokio::main] async fn main() { println!("{}", sum_to(100).await); } 会得到 5050。

要点解析：

• 状态机的大小必须在编译期确定，而「状态机里嵌一个同类型状态机」意味着无穷大， 编译器用 E0733 直接拦住。
• Box::pin 同时解决两件事：把值放到堆上（大小变成指针）、把地址固定（Pin， 因为 async 状态机是 !Unpin）。
• 工程上递归 async 要小心两件事：堆分配开销（每层一次）和 栈/堆深度（每层的 Poll 都会往上展开，深递归可能栈溢出）。 目录遍历这类场景更适合显式 Vec 栈或 ignore/walkdir 的 spawn_blocking 版本。
• 如果只在内部分用，用 futures::future::BoxFuture<'static, u64> 别名更易读， 并且 .boxed() 比手写 Box::pin 更简洁。

练习 10：排查 Future 不执行的原因

难度：★★☆

要求：下面代码「不会打印任何东西」，找出原因并修复：

// 依赖：tokio = { version = "1", features = ["full"] }
async fn log(msg: &str) {
    println!("{msg}");
}

#[tokio::main]
async fn main() {
    log("hello");
    let f = log("world");
    drop(f);
}

提示：Future 的 #[must_use] 只覆盖了一条路径。

参考答案（先自己写再看）

参考答案（先自己写再看）

原因：log("hello") 和 let f = log("world"); 都只构造了 future，没有 await， 所以两行 println! 都不会执行。let f 之后 drop(f) 让编译器认为变量被使用了， 因此没有警告，bug 完全静默。

修复：

// 依赖：tokio = { version = "1", features = ["full"] }
async fn log(msg: &str) {
    println!("{msg}");
}

#[tokio::main]
async fn main() {
    log("hello").await;         // 必须 await
    let f = log("world");
    f.await;                    // 或者显式 await 变量
    // 想并发就用 join!，而不是"构造了再丢掉"
    tokio::join!(log("a"), log("b"));
}

输出：hello、world、a、b（最后两行的顺序由调度决定）。

要点解析：

• #[must_use] 只作用在「表达式语句产生的值被立即丢弃」这一种情况； 赋值给变量后再 drop 就绕过了检查。
• Clippy 的 let_underscore_future（let _ = async_fn();）能抓住部分漏写， 但仍无法覆盖「赋值给变量」的写法——所以「函数名叫 xxx，返回值没被 await」 这件事只能靠 code review 与习惯。
• 记忆法：在 Rust 里，看到 async 函数调用就要想到「这里必须有一个 .await 或显式交给 executor（spawn/join!/select!）」；spawn 是例外， 它是「提交即开始」，不需要 .await 也能运行（但要保留 JoinHandle 才能拿结果）。

练习 11：并发上限与失败重试的组合

难度：★★★

要求：给定一个「一批 URL 需要抓取、并发上限 4、失败要重试 1 次」的场景， 用 futures::stream 的 buffer_unordered（或 tokio::sync::Semaphore）实现， 统计成功数与失败数。要求错误类型统一、代码可运行（用 sleep 模拟抓取）。 提示：buffer_unordered(n) 内部就是有界并发；重试用一个 for 循环包住单次尝试。

参考答案（先自己写再看）

参考答案（先自己写再看）

// 依赖：tokio = { version = "1", features = ["full"] }, futures = "0.3"
use futures::stream::{self, StreamExt};
use std::time::{Duration, Instant};

#[derive(Debug)]
struct FetchError(String);

impl std::fmt::Display for FetchError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl std::error::Error for FetchError {}

/// 单次尝试：用 sleep 模拟网络；url 里带 "bad" 就失败
async fn try_once(url: &str) -> Result<String, FetchError> {
    let ms = 30 + (url.len() as u64 % 5) * 10;
    tokio::time::sleep(Duration::from_millis(ms)).await;
    if url.contains("bad") {
        Err(FetchError(format!("{url} -> 500")))
    } else {
        Ok(format!("{url} -> 200 ({ms}ms)"))
    }
}

/// 带 1 次重试的抓取
async fn fetch_with_retry(url: &str) -> Result<String, FetchError> {
    match try_once(url).await {
        Ok(body) => Ok(body),
        Err(first) => {
            // 重试前稍微退避；真实项目用指数退避
            tokio::time::sleep(Duration::from_millis(20)).await;
            try_once(url).await.map_err(|_| first)
        }
    }
}

#[tokio::main]
async fn main() {
    let urls: Vec<String> = (1..=12)
        .map(|i| {
            if i % 5 == 0 {
                format!("https://bad{i}.example/api")
            } else {
                format!("https://host{i}.example/api")
            }
        })
        .collect();

    let start = Instant::now();
    // 用 iter() 借出 &String，再用 as_str() 得到 &str：和 fetch_with_retry 的签名一致
    let borrows = urls.iter().map(String::as_str);
    // buffer_unordered(4)：最多同时 4 个 future 在飞，谁先好谁先出
    let results: Vec<Result<String, FetchError>> = stream::iter(borrows)
        .map(fetch_with_retry)
        .buffer_unordered(4)
        .collect()
        .await;

    let mut ok = 0usize;
    let mut failed = 0usize;
    for r in results {
        match r {
            Ok(body) => {
                ok += 1;
                println!("OK   {body}");
            }
            Err(e) => {
                failed += 1;
                println!("FAIL {e}");
            }
        }
    }
    println!("成功 {ok}，失败 {failed}，总耗时 {:?}", start.elapsed());
}

输出：按完成顺序打印 OK ... / FAIL ...（bad5、bad10 会失败两次尝试后报错）， 最后是 成功 10，失败 2，总耗时 ...ms（约 200ms 上下，取决于耗时分布）。

要点解析：

• buffer_unordered(n) = 「最多 n 个并发 + 完成即产出」，等价于一个内建的 Semaphore， 但不需要手动管理 permit。需要结果保序时用 buffered(n)。
• 最常见的编译错误来源是元素类型与函数签名不匹配：fetch_with_retry 收 &str， 而 urls.iter() 产出 &String，所以要先 .map(String::as_str)（或者把参数 改成 url: &String / url: String）。这里的 borrows 借用 urls， 而 buffer_unordered(...).collect().await 在同一个 main 里就结束，借用合法。
• 错误类型统一成 FetchError，所以 buffer_unordered 的元素类型是 Result<String, FetchError>，无需 Box<dyn Error>。
• 重试要用退避，否则失败的服务会被立刻再次打满；并且要注意整体重试次数上限， 避免重试风暴。真实项目里 tower::retry + tower::limit 已经把这套逻辑做好了。
• 若任务需要独立取消或 panic 隔离，改用 JoinSet + Semaphore。

本章小结 / 自测清单

• 我能解释为什么「一连接一线程」在 C10K 场景会崩溃（栈内存、上下文切换、调度）。
• 我能说出 epoll 与 IOCP 的一句话区别（就绪通知 vs 完成通知）。
• 我知道 async fn f() -> T 的返回类型是 impl Future<Output = T>，且不 poll 就不执行。
• 我能画出 .await 如何把函数切成状态机，并说明「跨 .await 的变量会成为状态机字段」。
• 我能解释 Pin 为什么存在：防止自引用状态机被移动；!Unpin 类型拿不到 &mut。
• 我知道 std 不提供 executor，futures::executor::block_on 没有 IO 驱动。
• 我能写出 #[tokio::main] / #[tokio::test]，并说清 multi-thread 与 current-thread 的差别。
• 我能区分 join!（并发全部完成）、try_join!（失败短路）、spawn（'static + Send）、 select!（竞速，其余被 drop）。
• 我知道「取消 future 就是 drop 它」，并理解取消安全（cancel safety）的含义。
• 我能解释为什么不能跨 .await 持 std::sync::MutexGuard，以及三种修法。
• 我知道 CPU 密集和同步 IO 必须放进 spawn_blocking（或专用线程池），否则会卡住 runtime。
• 我知道递归 async 要 Box::pin，以及 1.75+ 的 async fn in trait 不能用 dyn， 要改成 impl Future + Send 或 #[async_trait]。
• 我能用「IO 密集 / CPU 密集 / 并发量」三个维度判断该不该用 async。
• 我能说出与其他语言的关键差异：Rust future 惰性、无内建调度、零成本状态机。