异步Rust中的Sink：从trait协议到反压与实现详解

写异步 Rust 写久了你会发现，真正折磨人的往往不是 borrow checker，而是那些一眼看到底、用起来却到处是暗坑的 stream 概念。Stream 是读，Sink 是写，二者一进一出，组成了异步 IO 世界里最基础的骨架。futures-rs 里的 futures-sink 包就只做一件事：定义那个常年不在标准库里的 Sink trait。我最早踩坑是在某个消息转发服务里，通道发送端用得好好的，换成自研协议时才发现，原来 channel sender 只是 Sink 的众多实现之一。今天把 Sink 怎么用、怎么实现、哪些地方容易翻车，一次性讲透。

1. Sink 是什么：异步世界的“传送带”

1.1 与 Stream 的对称关系

很多初学者会把 Sink 直接等同于“异步写入”，然后拿它和 AsyncWrite 比。这么理解方向没错，但颗粒度不够。在 futures-rs 的设计里，Stream 和 Sink 是一对极其工整的镜像抽象：Stream 是“拉”，Sink 是“推”。Stream 不断产生 Item，执行者.next().await把 Item 取出来；Sink 则接收 Item，执行者把 Item 交给它，让它“处理掉”。至于处理方式是写文件、发网络、进队列还是打印出来，Sink 内部完全自治，对外只暴露一套固定协议。

想象一个传送带。你把一个包裹放在传送带上，传送带负责把它运走。传送带不会当场告诉你包裹已经进了仓库，它只会告诉你“我准备好了，可以再接一个”。这就是 Sink 的工作方式：一次收一件货，收完告诉你结果，然后你继续放下一件。如果传送带满了，它会让你等，这就是反压。这个类比能解释 Sink 几乎所有方法的行为。

1.2 为什么是 poll 状态机而不是 async fn

有一个问题每个接触 Sink 的人都会问：为什么不直接设计成async fn write(&mut self, item: Item)？异步 IO 已经有了 async fn，写起来还简单。要回答这个，得先理解 futures-rs 的组合子生态。像forward、send_all这样的组合子，会在同一个 future 内部反复轮询多个组件，把 Stream 的 Item 一个个塞给 Sink。如果用async fn，每发一个 Item 都要创建并轮询一个子 future；更麻烦的是，取消和掉队很难控制。

Sink 采用的方法级 poll 协议，本质上是一个状态机接口。调用者拿到Pin<&mut Self>和Context，自己控制推进节奏。该 Ready 就 Ready，该 Pending 就 Pending。这样一来，组合子可以在一个 tight loop 里完成“从流里取一个 Item，再塞给 Sink”的操作，不需要中间结果落 heap，性能可控。标准库的 AsyncWrite 也沿用了类似设计，但 Sink 比 AsyncWrite 更抽象：AsyncWrite 只处理字节块[u8]，Sink 的 Item 可以是任何类型。

1.3 futures-sink 在 futures-rs 中的位置

futures-sink 本身是一个微小的 crate，只包含 Sink trait 以及相关辅助 trait 的定义。真正的实现和扩展方法分散在 futures-util 里。正常使用futures = "0.3"时，你不需要显式引入 futures-sink，所有东西都会通过futures::Sink、futures::SinkExt自动暴露出来。

值得注意，直到今天标准库都没有把 Sink 收编进去，这和Stream的命运有点像。很多库选择直接实现 futures-rs 的 Sink，比如 futures-channel 的 mpsc Sender、以及不少第三方协议库的发送半对象。也就是说，Sink 这套抽象已经是事实上的标准，理解了它，你在好几个生态之间跳转都不会觉得突兀。

2. 拆解 Sink trait：四个方法之间的契约

2.1 方法签名逐个看

Sink trait 的定义非常精简，核心就是四个方法：

pub trait Sink<Item> { type Error; fn poll_ready( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>; fn start_send( self: Pin<&mut Self>, item: Item, ) -> Result<(), Self::Error>; fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>; fn poll_close( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>; }

逐个解释。

poll_ready是反压闸门。调用它，就是问 Sink：“你准备好接收下一个 Item 了吗？”返回Poll::Ready(Ok(()))意味着可以安全地调用一次start_send。返回Poll::Pending意味着内部已经满了，调用者应该等待 waker 唤醒。这一点非常像StreamExt::next前的反压检查。

start_send是真正收 Item 的入口。它接收一个 Item，尝试把它放入 Sink 的内部缓冲或者立即处理。请注意它返回的不是Poll，而是直接的Result。因为从 Sink 的角度看，“接收 Item”本身是同步操作：要么收下，要么出错。真正的耗时动作——比如把缓冲数据写盘或者发网络——应该在poll_flush里完成。

poll_flush负责把内部所有缓冲的数据推向底层目标，并且确保底层目标完成刷新。它返回Poll，因为刷新可能是异步的。如果 Sink 内部没有缓冲，这里通常直接返回Ready(Ok(()))，但为了语义正确，最好还是转发给底层资源 flush 一下。

poll_close是收尾操作。语义上等于flush + 关闭底层资源。调用之后，Sink 不应该再接受任何 Item。某些 Sink 的poll_close会返回几次Pending，比如 TcpStream 关闭握手需要等待对端确认，这都符合预期。

2.2 调用顺序与不变量

第一次接触 Sink 的人，十有八九会踩同一个坑：直接调start_send，然后数据丢了或者 panic。Sink 的协议约束非常明确：每次调用start_send之前，必须调用一次poll_ready，并且等它返回Ready(Ok(()))。这不是建议，是硬性前置条件。实现方可以假设调用者遵守了这一点，在start_send里做侵入式检查时，经常是debug_assert。

完整发送一个 Item 的流程是：

1. 调poll_ready(cx)，直到返回Ready(Ok(()))。
2. 调start_send(item)。
3. 如果还要发下一个 Item，重复第 1 步。
4. 等全部 Item 都进入 Sink 后，调poll_flush(cx)，直到Ready。
5. 将来确认不再发送任何数据时，调poll_close(cx)。

这个协议的意义在于给实现方足够的自由度。Sink 可以在poll_ready里只检查“我内部队列有没有空位”，在start_send里把 Item 推进队列，在poll_flush里真正处理队列中的所有数据。这样组合子可以按自己的节奏决定何时等待、何时继续，不会因为一次异步写卡死整个发送循环。

还有一个容易忽略的点：如果一个 Sink 的poll_ready已经返回过Ready，但调用者连续调用了两次start_send（中间没有再次poll_ready），那么第二个start_send是不合法的。很多严谨的 Sink 实现会在这里检查内部状态，发现已经有未确认的 Item 时，直接返回错误或者断言失败。

2.3 Error 后的状态机变化

Sink 出错之后的状态处理，是网上文档很少讲明白的部分。当start_send或poll_*返回Err时，Sink 内部可能已经处于不一致状态。比如底层 socket 已经关闭，但内部缓冲还没有清理；或者某个异步任务已经取消，导致队列状态损坏。

我个人的经验是：收到Err之后，不要寄希望于“再发一次就能恢复”。正确做法是用poll_close尝试优雅清理，如果poll_close也失败，就干脆drop掉整个 Sink，让析构清理残留状态。在实现 Sink 时，同样要遵守这个约定：任何方法返回Err之后，后续所有方法都应该继续返回Err，而不是原地复活。否则调用方的错误处理逻辑会被彻底打乱。

3. 用起来：SinkExt 扩展方法与组合子

3.1 send / flush / close：三个常用方法

直接手写上面那套协议很啰嗦，所以 futures-util 给所有实现Sink的类型做了扩展，其中三个方法使用频率最高：

use futures::{Sink, SinkExt}; let mut sink = make_sink(); sink.send("hello".to_string()).await?; sink.flush().await?; sink.close().await?;

send内部帮我完成了poll_ready + start_send两步。要注意的是，send返回 Ok 只代表 Sink 成功“接收”了这个 Item，不代表它已经写到了底层目标。如果你的业务要求在返回Ok时数据一定落盘或者已经发出，紧接着调一次flush().await。这一点和AsyncWriteExt::write_all的语义很不一样，很多人第一次用 Sink 发数据后，后台迟迟看不到内容，就是因为漏了 flush。

flush对应poll_flush，close对应poll_close，行为都很直接。

3.2 forward 与 send_all：把 Stream 灌进 Sink

如果 Sink 的一端是数据源，另一端是数据汇，最优雅的做法是直接用forward。看这个例子：

use futures::{SinkExt, StreamExt}; async fn copy_stream_to_sink<St, S>(stream: &mut St, sink: &mut S) where St: StreamExt + Unpin, S: SinkExt + Unpin, { stream.forward(sink).await?; }

forward会不断从 Stream 取 Item，然后按照协议塞给 Sink，并在最后自动poll_flush一次。对于“把收到的消息转发到下游”这种典型代理场景，forward几乎不要钱。SinkExt::send_all类似，但重心更偏向“把一个已存在的批数据流灌入 Sink”。

实际项目里，我经常把forward用在连接转发的中间层。上游 TcpStream 的读取半对象是一个 Stream，下游连接的发送半对象是一个 Sink，一条forward就把双工转发写完一半。这比手写循环清晰得多，也不容易漏掉 flush。

3.3 with / with_flat_map / fanout：转换与分流

with是 Sink 最实用的组合子之一。它允许你对 Item 做一层映射，然后把结果塞给原 Sink。举个例子，原始 Sink 接收String，但你的上游产生u32：

let sink = tx.with(|id: u32| async move { Ok::<_, SendError>(format!("user-{}", id)) }); sink.send(42).await?;

注意with的闭包返回的是一个 future，所以里面可以执行异步操作。适合做编码、格式化、加前缀这类工作。with_flat_map则更进阶，它允许一个 Item 展开成多个 Item 传给下游。fanout能把一个 Item 同时送入两个 Sink，适合做日志双写、监控旁路这种场景。

这些组合子证明了一个核心观点：Sink 的价值不在于它本身能做什么，而在于它可以被无缝组合。你把一个 mpsc Sender 看成 Sink，再包一层with做格式转换，再交给forward，一条完整的数据管道就搭起来了，中间不需要任何手写循环。

3.4 futures_channel 的 mpsc：最常见的 Sink 实现

futures-channel 里的 mpsc Sender 是绝大多数人接触的第一个 Sink。它的协议实现相当标准：内部有一个有界队列，poll_ready检查队列是否还有空位，start_send把 Item 放入队列，poll_flush在这里几乎等于无操作，因为数据已经进入了接收端可见的队列。需要注意的是，mpsc Sender 在接收端被 drop 后，所有发送操作都会返回SendError，这时候错误处理要跟上。

初学时我建议直接用 mpsc Sink 做练手。它会让你直观理解poll_ready的反压作用：消息队列满时，发送方自然暂停，不会无界增长内存。理解了 mpsc 的 Sink 实现，再去看那些协议层、文件层、连接层的自定义 Sink，思路就会顺很多。

4. 从零实现一个可落地的 Sink

4.1 场景设计

为了把上面的协议落到实处，我写一个完整的LineSink：接收String类型的 Item，把字符串按行写入一个AsyncWrite类型的目标。这个场景很常用，比如异步日志收集、逐行写文件、往 TcpStream 发送文本协议。为了让示例不依赖运行时，我使用 futures-rs 自带的AsyncWritetrait，Tokio、Async-std甚至自定义 IO 对象只要实现这个 trait 就能用。

设计要点有三个：内部用一个Vec<u8>作为缓冲；start_send只做同步入缓冲，不直接写 IO；poll_flush负责把缓冲数据真正确认写入底层，并且处理写零的错误。这是 Sink 实现的典型结构，理解了它，后面所有复杂 Sink 都是在这个骨架上加状态。

4.2 完整代码

use std::io; use std::pin::Pin; use std::task::{Context, Poll}; use futures::io::AsyncWrite; use futures::Sink; pub struct LineSink<W> { writer: W, buffer: Vec<u8>, closed: bool, } impl<W> LineSink<W> { pub fn new(writer: W) -> Self { Self { writer, buffer: Vec::new(), closed: false, } } } impl<W: AsyncWrite + Unpin> Sink<String> for LineSink<W> { type Error = io::Error; fn poll_ready( self: Pin<&mut Self>, _cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { let this = self.get_mut(); if this.closed { Poll::Ready(Err(io::Error::new( io::ErrorKind::BrokenPipe, "sink closed", ))) } else { Poll::Ready(Ok(())) } } fn start_send( self: Pin<&mut Self>, item: String, ) -> Result<(), Self::Error> { let this = self.get_mut(); if this.closed { return Err(io::Error::new( io::ErrorKind::BrokenPipe, "sink closed", )); } this.buffer.extend_from_slice(item.as_bytes()); this.buffer.push(b'\n'); Ok(()) } fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { let this = self.get_mut(); while !this.buffer.is_empty() { let n = ready!(Pin::new(&mut this.writer).poll_write( cx, &this.buffer ))?; if n == 0 { return Poll::Ready(Err(io::Error::new( io::ErrorKind::WriteZero, "write zero bytes", ))); } this.buffer.drain(..n); } Pin::new(&mut this.writer).poll_flush(cx) } fn poll_close( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { if self.closed { return Poll::Ready(Ok(())); } ready!(self.as_mut().poll_flush(cx))?; let this = self.get_mut(); ready!(Pin::new(&mut this.writer).poll_close(cx))?; this.closed = true; Poll::Ready(Ok(())) } }

这段代码可以直接放进项目测试。运行时无所谓，最简单的验证办法是用Vec<u8>模拟底层 writer：

use futures::SinkExt; let mut line_sink = LineSink::new(Vec::<u8>::new()); line_sink.send("hello".to_string()).await.unwrap(); line_sink.send("world".to_string()).await.unwrap(); line_sink.flush().await.unwrap(); let bytes = line_sink.into_inner(); // 需要一个获取writer的方法 assert_eq!(bytes.as_slice(), b"hello\nworld\n");

Vec<u8>本身实现了AsyncWrite，所以测试起来非常方便。如果代码里没有into_inner，加一个方法或者直接把字段改成pub即可。

4.3 实现过程中的两个关键决策

第一个决策：start_send里只塞缓冲，不做 IO。这是 Sink 和普通write方法的关键差异。因为start_send是同步的、不能返回 Pending，如果在里面直接调用poll_write，遇到写缓冲满就会违背协议。把所有异步处理都推到poll_flush，才能保证协议一致性。

第二个决策：poll_flush必须循环写，直到buffer为空。这在真实环境中特别重要，因为单次poll_write返回的n往往小于buffer.len()，尤其是在 TcpStream 上。不要假设底层一次能吞掉所有数据。WriteZero的错误处理也很关键，写返回 0 意味着底层可能永久无进展，如果不主动暴露，调用方会卡在死循环里。

4.4 注意：Unpin 与 Pin 的边界

上面代码里W: AsyncWrite + Unpin是一个简化条件。写真实项目时，底层 writer 不一定 Unpin，这时你需要内部使用Pin<Box<W>>或者借助pin-project-lite做强制固定。Sink trait 的方法收的都是Pin<&mut Self>，这意味着实现者需要慎重思考内部状态是否自引用。只要没有自引用，结构体通常自动 Unpin，使用体验最舒服；一旦有自引用，就要转入 pin-project 那一套，复杂度会直线上升。

我个人的建议是：除非确实需要，否则不要在自定义 Sink 里引入自引用状态。把底层 IO 对象用Box固定，把内部缓冲留在同一结构体里，90% 的场景都不需要动 pin 的高级玩法。

5. 进阶细节与性能考量

5.1 为什么 Sink 一律接收 Pin 而不是 &mut self

这个问题值得展开。标准库很多 IO trait 也用Pin，但 Sink 的使用者往往感受不到原因。原因出在异步落盘的一个经典场景：你启动一次poll_write，底层驱动把buffer的地址交给内核或者网卡，在 Pending 返回期间，这个地址被外部“借用”了。如果这时候 future 被移动，原来那个 buffer 的地址可能失效，内核继续写就写到了脏内存。Pin 保证的是：只要 Sink 没被 drop，它内部的一切物理地址都保持不变。

所以对 Sink 实现者来说，只要不自己搞自引用，日子很轻松；但如果你想优化——比如让buffer和writer之间存在指针引用——就必须确保这个结构体是固定的。futures-rs 给这个场景留了口子，这也是它当年设计 Pin 的核心驱动力之一。

5.2 poll_finish 与异步 drop 的前景

在新版 futures 的 Sink trait 里，出现了一个带默认实现的方法poll_finish。它的设计目标是异步析构。Rust 的 Drop 目前是同步的，无法在析构函数里执行 await。当 Sink 持有需要优雅关闭的资源时，比如未 flush 的文件句柄、需要发送关闭帧的网络连接，直接 drop 会导致数据丢失或协议不完整。

poll_finish默认实现等价于poll_close，但语义上和poll_close有所区别：它允许在 future 被取消后再次调用，为异步 drop 做准备。虽然目前标准库还没有正式的 async drop，但这类先导性 API 正在慢慢铺路。我在实盘项目里暂时没有依赖它，只是了解它的存在，避免未来生态变了措手不及。

5.3 性能优化的两个方向

一个方向是减少系统调用次数。像LineSink这种实现，每poll_flush就把全部缓冲写出去，已经比每发一条消息就写一次 IO 高效得多。更进阶的做法是设置一个阈值：buffer 没到阈值前，poll_flush在系统层面其实不着急落盘，只有超过阈值或者调用flush时才真正写。这种“攒批”思路，配合 Sink 的 poll 协议，可以让吞吐量大幅提升。

另一个方向是减少内存拷贝。start_send时字节从String复制进Vec<u8>，poll_flush时又从Vec<u8>复制给内核。看似无法避免，但如果把 Sink 的 Item 类型设计成Bytes，就可以用引用计数替代第一层拷贝；如果底层支持write_vectored，甚至可以把多个 buffered chunk 一次性提交。性能优化永远跟业务绑定，我不建议一开始就追求极致，先写清楚、测明白，再慢慢榨。

6. 常见问题与排查实录

6.1 数据已经 send 成功，但底层就是看不到

这是 Sink 新手第一坑。send返回 Ok 只代表 Item 进入了 Sink 的内部缓冲，不代表完成了写 IO。特别是自定义 Sink 和带缓冲的 channel 实现，都会把真正的 IO 动作留到 flush。排查时先看代码里有没有flush().await，没有就补上。

如果是用send_all或forward，它们通常会在结尾自动 flush，但有些实现细节也未必保证 flush 到最底层资源。最简单的排查教训：发送完，再调一次flush，哪怕底层 flush 是无操作，也不会有什么副作用。

6.2 连续 start_send 后 panic 或者数据错乱

有些 Sink 实现会在start_send里检查内部“是否已经透支了 poll_ready”。如果你在一个 task 里手动循环，先poll_ready一次，然后连续start_send多条，就有触发条件。正确做法是每发一条之前都重新poll_ready。不要觉得多余，这正是反压机制的组成部分。

也有一种间接触发方式：在with转换闭包里做异步操作，闭包执行期间另一个并发任务也在向同一个 Sink 发送。Sink 默认并不保证多任务互斥，你需要在外层加锁或使用专门的并发 Sink 包装器。排查时看一眼是否多个 future 共享了同一个&mut Sink，如果是，改成在共享锁内调用，或者给每个任务一个独立 Sink。

6.3 错误之后再继续用 Sink，行为越来越奇怪

前面说过，Error 后的 Sink 可能已经脏了。最常见的是 mpsc channel 的接收端被 drop，之后send会立刻返回SendError。如果你顺势把错误吞掉，继续下一条send，只会反复收到同样的错误，而且可能丢失本可以恢复的 Item。

站在实现者角度，Sink 内部一旦进入错误态，后续所有方法统一返回同一个错误即可。站在使用者角度，Error 后尽快close或drop，不要试图按摩。如果这个 Sink 是长生命周期组件的一部分，通常应该重建整个组件，而不是修复这一个 Sink。

6.4 生命周期问题：想发送 &str，结果借不到地方放

Sink 的 Item 类型在 trait 上是泛型，但如果你的 Item 是&'a str，Sink 的方法只能接收这一个生命周期内的引用。很多时候数据源是异步产生的，临时 String 一 drop，借用就失效。这时建议把 Item 换成String，或者用with在进入 Sink 前完成转换。想通过借用零拷贝发送，等 Sink 生态支持 GAT 一段时间再说。

6.5 排查速查表

最后分享一点个人心得

我在实际项目里用 Sink 的最大体会是：它逼着你把“接收数据”和“真正发出去”分开想。很多人写异步发送代码失败，本质上是把这两个阶段混在一起了。Sink 的协议虽然多两个方法，但分清楚之后，反压、缓冲、批量刷盘这些复杂问题反而都有了现成的结构。

如果你一开始对 poll_ready 和 flush 的理解很模糊，建议先拿 mpsc Sender 写一个 demo，用 channel 满和空两种状态观察发送端的暂停和恢复。再把 LineSink 跑起来，接一个内存 Vec，打印出 poll_flush 被调用的时机。把这两个实验做完，Sink 这套状态机基本就长在你脑子里了。以后见到任何带 Sink 的接口，你都能在几秒内判断出它的缓冲行为、错误策略和反压特性。