怎么读这一卷

主线位置：主线的总览。本页把「宏展开」和「运行时」两段各画一遍，再用一段程序把整条主线走一遍。

flowchart TB
  start[从这里开始] --> q{读这一卷为了什么}
  q -->|会用| use1[00 导读]
  use1 --> use2[04 常见坑]
  q -->|读懂实现| all["00 到 06 按顺序，01 和 02 连着读"]

锁定版本

源码链接都用 tag，形如 https://github.com/tokio-rs/async-stream/blob/v0.3.6/async-stream/src/yielder.rs#L37。0.3.6 发布已经两年，仓库的 master 在这之后又合了四个提交，都还没发布：

读 master 会看到 sync_wrapper.rs 这个 0.3.6 没有的文件，行号也对不上。你 cargo add async-stream 拿到的是 0.3.6。

想在本地对照着读：

git clone https://github.com/tokio-rs/async-stream
cd async-stream
git checkout v0.3.6

两个 crate 的分工

async-stream 必须分成两个 crate：proc-macro 只能放在 proc-macro = true 的 crate 里，而这种 crate 不能导出普通的类型和函数。于是改写语法的部分在 async-stream-impl，运行时要用的类型在 async-stream，用户只依赖后者。

lib.rs 的 242 行里只有 31 行是代码，其余是 crate 文档和两个宏的示例。整个运行时是 yielder.rs 加 async_stream.rs 一百多行代码。

依赖也很少：async-stream 只依赖 futures-core（Stream trait）和 pin-project-lite（给 AsyncStream 做字段投影）；async-stream-impl 依赖 proc-macro2、quote 和打开了 full、visit-mut 两个 feature 的 syn 2。两个 crate 都没有自己的 feature，也没有 cfg 分支，所有平台上是同一份代码。

用户能看到的公开 API 只有两个宏。AsyncStream、pair、next 和两个 proc-macro 都在 __private 模块里，带 #[doc(hidden)]，只供宏展开出来的代码调用。

主线

只有一条：yield 怎么从 async 块里把值递出去。它分成两段，每一章开头都会标出本章在哪一段。

宏展开（编译期）：

flowchart TB
  A["stream! { ... }"] --> B["macro_rules!：塞进 ($crate)"]
  B --> C["proc-macro stream_inner"]
  C --> D["replace_for_await：for await 换成 #[await_] for"]
  D --> E["Scrub：改写 yield、?、#[await_] for"]
  E --> F["AsyncStream::new(rx, async move { ... })"]

运行时（每次 poll_next）：

sequenceDiagram
  participant C as 消费者
  participant S as AsyncStream::poll_next
  participant T as 线程局部 STORE
  participant B as async 块
  C->>S: poll_next(cx)
  S->>S: 栈上放 dst = None
  S->>T: 写入 &mut dst 的地址
  S->>B: poll(cx)
  B->>T: send(v)：取出地址，写入 v
  B-->>S: Pending
  S->>T: 恢复旧值
  S-->>C: Ready(Some(v))

下面这段程序在 async 块里和消费者那边都打印，看两边怎么交替执行：

async-stream = "=0.3.6"
futures = "0.3"
tokio = { version = "=1.54.1", features = ["full"] }

use async_stream::stream;
use futures::{pin_mut, StreamExt};

#[tokio::main]
async fn main() {
  let s = stream! {
    println!("  [block] start");
    for i in 0..2 {
      println!("  [block] yield {i}");
      yield i; // 宏展开：改写成 __yield_tx.send(i).await
      println!("  [block] resumed after {i}");
    }
    println!("  [block] end");
  };
  println!("[consumer] stream created, nothing ran yet");
  pin_mut!(s); // AsyncStream 里是一个 async 块，!Unpin，next() 之前要先 pin
  while let Some(v) = s.next().await {
    println!("[consumer] got {v}");
  }
  println!("[consumer] got None");
}

输出（跑 3 次相同）：

[consumer] stream created, nothing ran yet
  [block] start
  [block] yield 0
[consumer] got 0
  [block] resumed after 0
  [block] yield 1
[consumer] got 1
  [block] resumed after 1
  [block] end
[consumer] got None

逐段对应：

• 第一行：stream! 只是构造了一个 AsyncStream，里面的 async 块还没被 poll，一行都没执行。
• start、yield 0、got 0：第一次 next() 调到 AsyncStream::poll_next，它 poll async 块；块一直跑到 send(0).await，Send::poll 把 0 写进 poll_next 栈上的槽，返回 Pending；poll_next 看到槽里有值，返回 Ready(Some(0))。
• resumed after 0：第二次 next() 再 poll async 块，它从上次停下的 .await 接着跑。这次 Send::poll 发现值已经交出去了，返回 Ready(())，块继续往下。
• end、got None：块跑完，poll_next 记下 done，返回 Ready(None)。

async 块里的代码只在消费者 poll 的时候执行，执行到下一个 yield 就停。这就是「惰性」，为什么需要它会把它和通道的写法放在一起比较。

测试与 CI

async-stream/tests/ 下有四个集成测试文件（stream.rs 237 行、try_stream.rs 87 行、for_await.rs、spans_preserved.rs），外加两个小 crate tests/reexporter 和 tests/test-reexport，专门验证宏被别的 crate 重新导出之后还能用。

tests/ui/ 下有 6 个编译失败用例，用 trybuild 比对编译器的完整报错。这个测试只在 stable 上跑：报错的措辞随 rustc 版本变，nightly 上一定对不上。本卷引用这些用例时，都抄进自己的实验工程，用本机的 rustc 1.99.0 重新编译，贴的是本机的报错。

阅读路线

• 只想会用：读 00 导读和 04 常见坑。读完知道 stream! 和 try_stream! 的区别、? 在两个宏里的行为、为什么要 pin，以及几条常见报错从哪来。
• 想读懂实现：按顺序读。01 讲宏把代码改写成了什么，02 讲改写出来的 send().await 在运行时做了什么，这两部分要连着读。03 把取舍汇总，06 从零写一个最小的 stream!。

摘自 async-stream 的代码片段一律保留原来的格式（4 空格缩进），方便和原文件对照；本站自己写的示例用 2 空格缩进。宏展开的结果用 cargo expand 实际生成，只摘关键几行。

前置知识

这一卷默认你会写 async Rust，也大致知道 proc-macro 是什么。用到的概念在别的卷里讲过，这里只链过去：

目录

小结

• 全卷锁定 tag v0.3.6（b0b2f22）；master 上之后的四个提交（syn 3、MSRV 1.71、Sync 包装等）都没发布，行号以 tag 为准。
• async-stream-impl 是 proc-macro，负责改写语法；async-stream 放运行时类型和两个 macro_rules! 外壳。用户可见的只有 stream! 和 try_stream!。
• 主线分两段：宏把 yield v 改写成 __yield_tx.send(v).await；运行时 send 借线程局部指针把值写进 poll_next 栈上的 Option，返回 Pending。
• async 块只在消费者 poll 时执行，跑到下一个 yield 就停。
• 只想会用读 00、04；想读懂实现按顺序读，01 和 02 连着读。