把 Stream 理解成「异步版的 for 循环」会在生产管道里立刻失效。迭代器是拉取的：下游不 next，上游不生产。Stream 同样是拉取的：下游不 poll_next，上游不该把数据推进内存里堆着。很多包装看起来像在「处理流」，实际上在 buffer、collect、无界 channel 里把拉取变成了无限推送。CPU 还没忙过来，RSS 先到顶。
这篇文章只讲异步管道的节奏：poll_next 如何把控制权交给下游、有界队列和并发上限如何把内存钉死、慢消费者时该阻塞、丢弃还是断开。读完你应该能独立回答三个问题：为什么 collect::<Vec<_>>().await 等于取消背压、buffer_unordered(n) 的 n 限制的是什么、通道满了却还在 accept 的服务会先死在哪。
1. 背压是「不 pull 就不生产」，不是再开一个缓冲区
TCP 自己有窗口：接收方窗口缩小，发送方就慢下来。应用层管道常常把这个性质弄丢——中间用无界队列「解耦」，等于告诉上游：你能造多快造多快，我用内存当窗口。窗口一旦变成堆，背压就从「让生产者等待」变成「让 OOM killer 等待」。
有界才是默认。界可以是：channel 容量、ready_chunks 的块大小、信号量许可数、buffer_unordered 的 in-flight 数、连接 accept 的 backlog。界的单位要选对：按条数在大消息下会炸内存，按字节数在小消息下会让并发过低。生产里常常两个都要：最多 N 条 且 最多 M 字节。
缓冲的正当用途是平滑抖动：下游偶发 2ms 的停顿，上游不必每次都卡住 syscall。缓冲深度按「抖动 × 速率」估，再加很小的余量。按「尽量不让上游等」来设，你会得到一个事实上无界的队列，只是有一个永远达不到的数字写在配置里。
丢弃也是一种背压应答，但必须显式：度量丢弃率、决定丢最旧还是最新、让调用方能感知（计数器、标志位、关闭）。静默 try_send 失败后 continue，等于把丢失写进产品行为却不告诉任何人。
丢最旧还是最新取决于项的时效。行情、指标、位置：最新更有价值，队列满时丢掉队头。审计、支付、复制日志：每一项都有价值，不能丢，只能等或断。用错策略比没有背压更危险——你以为系统在保护自己，其实在丢不可丢的东西。策略写进类型或配置枚举，不要写进某次代码评审的口头同意。 有界队列的「等」要有超时。永远等会在下游死掉时把上游全部挂起，最后从入口开始连不上。超时后走丢或断，并打点。超时时间小于客户端超时，才能让调用方收到明确错误而不是自己先放弃。这三层超时（应用等队列、客户端等你、编排等进程）必须排好序，乱序会出现重试风暴。
2. Stream 的 poll 节奏由下游决定
Stream 的核心是 poll_next(self: Pin<&mut Self>, cx) -> Poll<Option<Item>>。返回 Pending 时必须登记 Waker；返回 Ready(Some) 是一项；Ready(None) 是结束。下游的 while let Some(x) = s.next().await 就是在循环 poll。下游去干别的、不 poll 你，你的 poll_next 根本不会被调用——这就是背压的机械来源。
对比 Sink：poll_ready 问下游能不能再收，start_send 把项交给它，poll_flush 刷出。正确的管道是：Sink 未 ready 时，上游 Stream 不被 poll（或 poll 了也要把项留在手里）。forward 一类组合子就是在做这件事。自己手写 loop { let x = s.next().await; sink.send(x).await; } 时，send 的 await 本身就是背压；若 send 前面先 s.next() 吃了一项，这一项已经离开上游。有界的话，一项的在途通常可接受；若你先 ready_chunks(10_000) 再 send，背压的粒度就变成一万项。
use futures::{StreamExt, SinkExt};
async fn pump<S, K>(mut src: S, mut sink: K) -> Result<(), K::Error>
where
S: StreamExt<Item = bytes::Bytes> + Unpin,
K: SinkExt<bytes::Bytes> + Unpin,
{
while let Some(item) = src.next().await {
sink.send(item).await?;
}
sink.close().await?;
Ok(())
}
这个循环一次只在途一项。要提高吞吐，用有界的并发，而不是先 collect。
FusedStream 防止结束后还 poll。组合 select! 时，结束的那一侧若不 fuse，会反复 Ready(None) 打转。这不是背压，是忙等。生产管道里 fuse() 是默认，不是优化。
3. 组合子里哪些在囤积，哪些在限制
futures / tokio_stream / StreamExt 的名字经常比行为温和：
读组合子源码里「内部缓冲」那几行，比读文档标题重要。buffer(n) 会在内部堆到 n 个已完成项；chunks 会堆到 n 项或超时；fold 会把整个流收成一个值。名字里没有 buffer 的，也可能在闭包的 Vec 里囤积——那是你自己写的。评审组合子链时，沿途把每一段的最大在途项数写在旁边，乘以每项字节，得到这条链的内存上界。写不出来的链，就是没有背压的链。
n 从配置来，不要从代码常量来，并且要有上限钳制。配置被改成 100 万时，程序应当拒绝启动或把 n 钳到预算内，而不是默默变成无界。启动时用「n × 平均项大小 × 管道条数」估 RSS，超过容器 limit 的某个比例就失败。这是把背压预算前移到部署时刻。
组合子链过长时，中间某段的错误类型被 map 丢掉，失败会从链的另一头冒出来，很难定位。宁可拆成命名的异步函数，每段有自己的 Result 和 span。调试一条匿名链的成本，通常高于多写两个函数。背压和错误处理在这里是同一条可读性要求。
map/filter：同步、一项进一项出，不囤积。闭包里不要做重 CPU，否则堵在下游的 poll 里，等于堵 worker。then/and_then：对每项起一个 future，默认串行。背压仍在，吞吐受每项延迟限制。buffer_unordered(n)/for_each_concurrent(n)：最多 n 个 in-flight future。n是内存和下游负载的旋钮。n = usize::MAX或「CPU 核数乘一个拍脑袋的 32」会在下游变慢时把飞行中的请求堆成山。buffered(n)：有序，完成了也要等前面的，额外的完成项会停在组合子内部。有序的代价是队头阻塞。ready_chunks(n)：下游一次拿一批。批大小是延迟和吞吐的交换：太大，一项很慢也能拖着一大批内存。collect::<Vec<_>>：把流读完。无限流或长度等于流量的流上，这就是取消背压的标准写法。unfold/ 自己poll_fn：你自己就是上游。没有人强迫你尊重 Pending。可以在 poll 里狂读 socket 填 Vec——编译器不管。纪律在人。
use futures::StreamExt;
async fn fanout<S>(src: S) -> anyhow::Result<()>
where
S: futures::Stream<Item = u64> + Unpin,
{
src.for_each_concurrent(32, |id| async move {
let _ = fetch(id).await;
})
.await;
Ok(())
}
32 必须来自下游真实容量：数据库连接池大小、对端 QPS 上限、本进程内存预算除以每项大小。写死 32 和写死 32_000 的差别只在事故来得早晚。配置化，并打 in-flight 的当前值。
4. Channel 是最常见的 Stream 边界
tokio::sync::mpsc 有界通道：send.await 在满时挂起发送方——这就是背压传到上游任务。try_send 满了立即失败，把决策交给调用方。unbounded 从不挂起发送方，只在内存上记账。
选择顺序：
- 有界
send.await。上游是你控制的任务，可以慢下来。 - 有界
try_send+ 明确策略（丢、断开、降级）。上游是 accept 循环或不可阻塞的回调。 - 无界。仅当生产者速率有硬上限（例如「最多 N 个 in-flight 请求，每个一条消息」）并且你能用类型或信号量证明这个上限。
broadcast 给多个消费者同一份数据。慢消费者会丢（lagged），这是显式丢弃。不要用 broadcast 当工作队列——工作队列每条消息只该被一个工人拿走，那是 mpsc。watch 只保留最新值，适合配置。oneshot 一次结果。四种结构四种背压：mpsc 阻塞或有界拒绝，broadcast 丢旧，watch 丢所有中间值，oneshot 没有流。
通道里的物件形状决定内存。塞 Vec<u8> 大包，有界 1024 也能是数 GB。塞 Bytes 仍可能钉住整块读缓冲（零拷贝那篇第 10 节）。更好的形状是 id、小结构、或已经切成需要大小的 Bytes。大包走单独的路径和更小的界。
5. 慢消费者：三种合法反应
下游处理变慢时，上游必须感知。三种合法反应：
阻塞（反压）。 发送方 await，生产减速。TCP 窗口、有界 mpsc、Sink 的 poll_ready 都是这个。适合「最终要处理完、延迟可以涨」。链路要避免环：A 等 B 的槽，B 等 A 的处理，没有超时就会死锁。超时和断开是环的出口。
丢弃。 采样、最新值覆盖、随机丢。适合指标、日志、视频帧。必须有计数。丢弃策略写进 SLA：指标允许 1% 丢失，支付流水不允许。
断开。 拒绝新连接、关掉最老的连接、对调用方返回 503。适合已经没有意义再收的输入。比无界排队更尊重进程。负载均衡器看到 503 或连接失败，才能把流量切走；看到你还在 accept 但内部排队三分钟，它会以为你很健康。
非法反应：继续 accept、继续 spawn、继续往无界队列塞，同时用日志说「下游有点慢」。这是在用内存把慢翻译成更慢，直到崩。
6. 取消如何沿着流向下传
流的上游常常是 socket、定时器、子任务。下游 drop 这个 Stream 或 select! 切走，当前 poll_next 内部的 future 会被 drop。若 poll_next 里只是在等有界 recv，drop 会解除一次发送方的等待——发送方得到接收端关闭，这是干净的。若 poll_next 内部 spawn 了孩子去拉数据，孩子不会自动死。和调度那篇同一条：取消要传到叶子。
Stream 上的 take_until(token.cancelled())、循环里 select! { cancelled, next } 是显式做法。HTTP 请求被客户端断开，handler drop，内部的下游 RPC Stream 应该停。做不到就会出现「客户端都走了，我们还在打下游」的放大。
取消安全：next().await 拿到的项已经离开上游。若在处理这项时被取消，项可能丢。需要至少一次交付，就不要在「已经 recv、尚未 ack」之间放无保护的 await；或把项先放到本地，取消时再塞回（注意顺序和死锁）。消息队列语义（at-most-once / at-least-once）在这里落地，不是在注释里落地。
7. 批处理、时间与内存的三角
chunks_timeout(n, dur) 一类：满 n 条或等到超时就吐一批。这是用延迟换吞吐（更少 syscall、更好的向量化）。三角是：
- n 太大：一项卡住整批内存，尾延迟变差。
- dur 太大：空闲时最后几条要等很久才刷出。
- 两者都小：退化成逐条，批的意义没了。
按下游的最优批次和 SLA 的延迟预算来设，并在超载时降 n——超载还坚持大批次，是在用更大的内存换已经不存在的吞吐。刷出失败要有策略：重试会让批在内存里更久；丢掉要计数；断开要让上游停。
时间源用 Tokio 的计时器，不要在 Stream 的 poll 里 std::thread::sleep。poll 里睡眠是堵 worker，背压会变成全局卡顿。
8. 和零拷贝、调度叠在一起
流里流动的如果是大 Bytes，背压的界要按字节。流里流动的如果是 id，工人再去共享存储取载荷，存储那一侧也要有界——否则 id 队列看起来很瘦，工人手里的载荷把内存打爆。
spawn 每个项一个任务，等于把 Stream 的拉取变成无界任务队列。for_each_concurrent(n) 已经给了 n 个槽，再在闭包里 spawn 就把 n 变成了无限。闭包里应当直接 await 工作，而不是再扔出去。
worker 上做重 CPU 的 map，背压表现为整颗核停在这一条流上，其它任务饿。CPU 重的变换 spawn_blocking 或专门的线程池，并且那个池的队列也要有界——否则背压只是从 Tokio 挪到池的无界队列。
9. 取舍：拉取、推送、混合
纯拉取（Stream）适合下游速率决定系统速率：消费数据库行、读文件、读 socket。纯推送（回调、无界）适合不能阻塞的生产者：中断、某些 FFI 回调。混合是生产现实：回调里 try_send 到有界队列，一个拉取任务在另一侧 recv。回调侧满了只能丢或在文档允许时阻塞极短时间。不要在中断里 send.await——那里没有 runtime。
响应式流的「request(n)」学分是明确的信用。Rust 生态较少把信用做成类型，常用有界队列近似。需要精确信用（一次请求 128 项）时，自己在协议里做 window，不要指望 mpsc(128) 完全等于窗口——隐藏的缓冲（Nagle、codec、ready_chunks）会额外囤积。
10. 排障：OOM、延迟单调涨、丢了却没人知道
- RSS 随 QPS 线性涨。 找
unbounded、collect、buffer_unordered过大、accept 和 worker 之间没有界。打每个队列的 length 直方图。 - 延迟单调涨，CPU 不高。 队列在堆里，工作没有被处理。慢消费者 + 缓冲。缩小界，让延迟在界满时变成拒绝，而不是无限变慢。
- 偶发丢数据。
try_send/ broadcast lagged / watch 覆盖。补计数和告警。确认产品允许丢。 - 死锁。 两个有界队列互相等待。加超时，或合并成单队列，或规定锁/队列顺序。
- 取消后下游仍在跑。 子任务没接到 token。用
JoinSet或显式 abort。 - 吞吐上不去。 界太小，或串行
then本该并发。看 in-flight 是否经常掉到 0（上游慢）还是经常顶满（下游慢或界太小）。顶满且下游 CPU 闲，是界以外的地方串行了（锁、单连接）。
压测要带慢下游：故意把处理 sleep 10ms，看内存是否封顶、是否开始拒绝。只压快下游，永远测不出背压缺陷。
11. 设计清单：每条管道三个数
对每一条生产管道写下三个数：最大 in-flight 项数、最大 in-flight 字节、满了时的行为（等 / 丢 / 断）。三个数进配置和度量。缺任何一个，这条管道在评审里就不该过。
数字来自预算：内存上限除以每项大小、下游池大小、延迟 SLA 反推的深度。拍脑袋的 1024 比无界好，因为它给了一面能被顶到的墙；墙顶到时必须有告警和行为（等、丢、断）。没有告警的墙等于没有墙——你只会在 RSS 图上事后看见它曾经存在。把三个数和告警规则一起提交，管道才算设计完。 数字来自预算：内存上限除以每项大小、下游池大小、SLA。拍脑袋的 1024 比无界好，但仍然是拍脑袋——至少它给了你一个能被顶到的墙，墙顶到时要有告警。
把这个清单从入口（accept）写到出口（下游 HTTP/磁盘），中间每一跳都有界，才叫端到端背压。中间任何一跳无界，墙就建在那一跳的堆上。
流量整形和控制面要分开。数据面的 Stream 处理字节和消息；控制面的「暂停消费」「调整 n」「切断某条上游」走 watch 或单独的命令通道。不要在热路径上每次 poll 都读一遍全局配置锁。watch::Receiver 在配置变了才唤醒，平时 load 一个 Arc<Config>。把背压旋钮做成可运行时改的：下游开始 503 时把 for_each_concurrent 的 n 降下来，比崩了再扩容更像服务。改 n 的实现可以是外层循环重建流，或自己写一个带 AtomicUsize 上限的并发槽。重建流要注意取消和 inflight 排空，不要两个世代的工人同时打下游。
公平性：一条热流把 worker 打满，其它流的 poll_next 变稀。合作预算是安全带，不是公平调度器。多租户系统要在应用层做每租户的并发槽和每租户队列，否则背压会变成「谁流量大谁占满，小客户被饿死」。这和锁那篇的不公平饥饿是同一现象，只是队列换成了任务。每租户一个有界 channel + 全局 worker 数上限，是最朴素的隔离。
测试背压不要只测功能正确。写一个生产者比消费者快十倍的测试，断言内存峰值低于某条线、断言开始出现拒绝或等待。再写一个消费者挂掉的测试，断言生产者在有界 send 上及时收到关闭而不是永远卡住。这两条比「能把 100 项 collect 出来」更能保住生产。
HTTP/2 和 gRPC 自带流控窗口，这是协议层的背压。你在应用层再无界缓冲，等于把窗口的信号吞掉：对端以为你还能收，其实堆在你的 mpsc 里。正确的衔接是：应用层队列深度逼近上限时，停止 poll HTTP/2 的 DATA，让窗口缩小。hyper / h2 在你不读 body Stream 时就会这样做——所以「先把 body to_bytes 再处理」不仅拷贝，还把协议窗口拉满。流式处理 body 是在把 TCP 窗口、HTTP/2 窗口、应用队列连成一条链。任何一节变成无限桶，链就断。
消息系统（Kafka、NATS、自己写的 replica log）里，offset 提交是背压和至少一次交付的交界。先处理再提交：崩溃会重复，消费者必须幂等。先提交再处理：崩溃会丢。缓冲一批再提交：延迟和重复窗口都变大。无论哪种，处理侧的有界并发决定了「未提交的 in-flight」上限，这个上限必须能用内存预算算出来。把 poll 到的消息丢进无界任务池再异步提交，offset 和实际处理就会乱序，重复和丢失同时发生。JoinSet 或信号量把 in-flight 钉死，完成后再推进 offset，顺序才说得清。
「推送」接口（webhook、回调、消息监听器）常常不能阻塞。这时背压的第一跳必须是 try_send 到有界队列，满了返回 429 或对监听器返回错误。把 200 先回给对端再慢慢处理，是在对对端撒谎：它会按你还能吃的假设继续推。撒谎的窗口等于你的无界队列。能诚实地说「现在满了」，对端或中间的重试/窗口才能起作用。产品上怕 429 不如怕 OOM，OOM 之后你什么状态码都回不了。
度量除了队列长度，还要有「满了多久」和「因满拒绝的次数」。长度长期顶满说明界太小或下游太慢；拒绝次数上升说明背压在工作——这是健康信号，不要一看到拒绝就盲目加容量。加容量之前看下游：下游 CPU 闲着，是锁或串行化；下游 CPU 满着，加你的队列只是加延迟。把拒绝做成对入口可见（503/429），负载均衡才能把流量切走。内部消化到内存里，负载均衡仍觉得你活着，会继续喂。
Stream 的生命周期经常和连接绑在一起。返回 impl Stream<Item = T> + '_ 的函数，流活着时借用着连接或缓冲。调用方 collect().await 还好；调用方把流 send 到另一个任务，借用过不了 'static。要跨任务，流必须拥有它的源（TcpStream move 进去）或源是 Arc。这是所有权篇在管道上的投影。async_stream::stream! 宏写出的生成器同样捕获环境，spawn 时一样要 Send + 'static。看起来像「只是 yield 几项」，类型上是一个藏着捕获的状态机。
背压测试要注入时间。用 tokio::time::pause 让 chunks_timeout 的超时在逻辑时间里触发，断言批次边界。用人工的慢 Sink（poll_ready 隔若干毫秒才 Ready）断言上游 poll_next 的次数不会在 Ready 之前狂飙——如果次数狂飙，上游在没有下游信用时仍在生产，背压断了。这项测试比压测更早抓住组合子用错（例如先 collect 再发）。
多路复用时，一条慢流不应堵住同连接上的其它流。HTTP/2 多流、gRPC 多 RPC 共享窗口又各有自己的。应用层若用一把全局有界队列收所有流的消息，慢消费者会让快流的项排在后面——队头阻塞从 buffered(n) 挪到了你的全局队列。按流分队列、全局再加总 in-flight 上限，两级界。实现成本更高，但是多租户和多路复用的默认。单队列只适合单消费者、同质项。
「有界」的界要能在运行时被看见。tx.max_capacity()、当前 tx.capacity()、信号量 available_permits() 导出成 metric。没有这些数，你无法区分「界太小」和「下游死了」。告警打在 capacity == 0 的持续时间上，而不是打在 QPS 上。QPS 下降可能是上游没流量，也可能是你自己把入口堵住了——后者必须用队列满的信号来说话。
Sink 只 start_send 不 flush，数据可能还在编码器里，上游却以为交出去了。关连接前 close().await，否则半包留在本地。flush 失败可能已经送出部分字节，重试语义和「从未发送」不同。测试覆盖正常 close、flush 失败、close 时仍有 in-flight。
复杂协议用阶段枚举的状态机，比三层 select! 夹组合子更容易把取消和半包落在明确阶段上。无限流没有 Ready(None)，和有限流混用时，有限流结束后要关掉 tick 分支，否则空转会把下游打出 4xx 风暴。
有界队列的深度不只是内存：最坏额外延迟约等于深度除以处理速率。按延迟预算反推深度，再和内存预算取小。只按内存把 1024 改成 102400，会在 SLA 上先爆。突增的记忆效应是：堆满之后即使输入恢复，消化堆积仍要一段时间，界越大记忆越长。
多路复用不要用一把全局队列收所有流，慢流会堵住快流。按流分队列，全局再加总 in-flight。推送回调不能 send.await，第一跳 try_send，满了诚实返回 429，让对端的窗口起作用。先 200 再慢慢处理是在撒谎，撒谎的窗口等于无界队列。
HTTP/2 窗口和应用队列必须连成一条链。to_bytes() 聚合 body 会把窗口拉满。度量导出 capacity 和「满了多久」，告警打在满的持续时间上。拒绝次数上升说明背压在工作，不是一看到拒绝就加容量。
12. 小结
Stream 的原理是拉取。背压就是下游不 poll 就不生产，用有界的队列、并发和信用把这个性质保住。缓冲只能平滑抖动，不能当无限窗口。满了必须等、丢或断，并且被度量。下一篇处理管道中间那些不得不共享的可变状态：Mutex、RwLock、以及只有争用证据确凿才该碰的 lock-free。
把三条管道（入口 accept、内部 worker、出口下游）的三个数贴到仪表盘上：in-flight 项、in-flight 字节、满了的行为计数。一周里如果「满了」从未发生，界可能过大，延迟记忆会在真正的尖峰时才被你看见。从未满过的有界队列和无限队列，在你见过的流量下无法区分。故意压一次慢下游，确认墙会顶上，确认拒绝或等待会发生。这才叫验证过背压。