企业官网建设流程全解析

我去年维护某个实时消息服务时，发现一个非常有意思的错位：消息从TCP连接读上来，用Stream梳理得特别顺手，一个map接一个filter就能把逻辑串得很干净；但到了发送方向，代码就退化成一堆循环套await，每个下行连接一套手工排队、重试、关闭的逻辑，写起来又碎又容易漏。那段时间我专门去翻了futures-rs的源码，结论是：发送端本来就缺一个像Stream一样的抽象，而它叫Sink。

这篇文章围绕futures-rs生态里的futures-sink模块展开。我会先把Sinktrait的四个核心方法掰开揉碎，讲清楚它的调用协议和设计意图，然后手写一个能落地的批量提交Sink，再结合组合器和几个真实项目里的坑，帮你在自己的异步服务里敢用、会用Sink。

1. Sink存在的理由：从两个不对称说起

1.1 收数据顺手，发数据难受

Stream在Rust异步生态里已经深入人心。它本质上是一个“异步迭代器”，next().await一下，数据就出来了，配合map、filter、fold这些组合器，处理读取方向的数据流非常自然。

但发送方向长期处于半原始状态。最常见的写法是拿一个tokio::sync::mpsc::Sender到处clone，然后在这边send(msg).await，在那边while let Some(msg) = rx.recv().await。问题在于：每个send只能发一条，如果我想“把这批解析出来的消息统一转成下行协议帧再发给所有订阅者”，就得自己管理队列、批量、背压、重试，这些东西本质上和Stream那边的map/collect是同一类问题，却没有任何公共接口来抽象它们。

Sink<Item>就是给这个场景补的抽象：它描述“一个可以异步接收Item的地方”。有了它，发送方向也能拥有一套组合器，send、send_all、fanout、buffer，像流一样自由组装。

1.2 Sink trait 的核心语法与调用协议

futures0.3版本中，Sinktrait定义在futures-core的sink模块里，futures-sink这个crate主要是把它和服务端集成环境拆开，方便只依赖futures-core的最小场景。核心接口长这样：

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

这看起来比Future复杂，但其实它描述的是一个非常常见的资源使用流程，你完全可以把它类比成给打印机送纸：

• poll_ready：问打印机“现在能接收一张纸吗？”如果设备在忙，就返回Pending，等它空闲了再通过waker唤醒。
• start_send：既然你说能接收，我就把这张纸（item）真正塞进去。这个调用不会阻塞，只负责“投递”，如果投递失败就返回错误。
• poll_flush：把纸槽里所有残留的纸全部吐完，确保数据真正到达目的地。
• poll_close：关闭打印机，不再接收新纸，同时完成最后的清理。

协议要求很严格：如果你想发送一个元素，必须先poll_ready拿到Ready(Ok(()))，再调用start_send，之后可以回到poll_ready继续下一轮，也可以转入poll_flush。绝对禁止在poll_ready返回Pending或者还没有Ready的时候直接start_send，这和Future的“不poll就没进度”是同一套逻辑。

提示：start_send本身不会等待。它只负责把数据交到Sink内部，真正的“发送完成”由poll_flush保证。很多新手把start_send当成阻塞发送来用，后续调用链很容易出问题。

2. futures-sink 里到底有什么

2.1 模块分布与依赖方式

futures-sink这个crate本身内容不多，但它在futures-rs里负责一个关键的拆分：让Sinktrait可以被futures-core之外的生态直接复用。实际写业务代码时，绝大多数人依赖的是futures这个门面crate，它会把Sink、SinkExt以及各种组合器一起导出来。

在Cargo.toml里最简单的做法：

[dependencies] futures = "0.3"

如果你特别在意编译体积，也可以只引入：

[dependencies] futures-core = "0.3" futures-sink = "0.3" futures-util = { version = "0.3", features = ["sink"] }

这样只拿Sinktrait和组合器，不引入全套executor等组件。实际项目里我建议直接用futures，组件之间的版本匹配关系少，省心。

2.2 SinkExt 组合器总览

SinkExt是futures-util里最常用的发送端扩展trait，它给所有Sink实现了大量组合方法。我实际用得最多的是这些：

光看名字可能觉得抽象，我挑两个最常用的具体说说。

send_all解决的是“从Stream到Sink”的搬运问题。传统写法是一个while let Some(item) = stream.next().await { sink.send(item).await; }，而send_all把这段逻辑封装进了一个Future，它内部会连续调用poll_ready和start_send，直到Stream结束，最后自动触发一次flush。这个组合器的意义不只是少写几行代码，而是它把“消费一个异步流”和“生产到一个异步目的地”之间的背压传递做对了：每次只从Stream取当前Sink能接收的数量，不会提前把整个Stream都拉进内存。

with则更适合做协议封装层。我在某个项目里需要把内部的结构体转成下行帧再发送，结构大概是这样的：

use futures::sink::{Sink, SinkExt}; use futures::future::ready; #[derive(Debug)] struct InnerMsg { id: u32, body: Vec<u8>, } #[derive(Debug)] struct Frame { bytes: Vec<u8>, } async fn encode(msg: InnerMsg) -> Result<Frame, MyError> { // 实际是封装成带长度前缀的帧 Ok(Frame { bytes: format!("{}:{}", msg.id, String::from_utf8_lossy(&msg.body)).into_bytes() }) } let inner_sink: MySink<Frame> = MySink::new(); let mut framed_sink = inner_sink.with(encode); framed_sink.send(InnerMsg { id: 1, body: b"hello".to_vec() }).await?;

这里with接收的闭包返回一个Future，它可以做耗时操作，比如查缓存、算校验和，都不会阻塞Sink本身。

3. 手写一个 Sink：从零到能跑

3.1 目标：一个自动批量提交的 Sink

看文档是一回事，自己动手实现一遍才能理解Sink的状态流转。这里我写一个BatchSink：它接收字符串消息，攒满一批后调用你提供的异步提交函数，将整批数据一次性写入外部存储。

这种Sink在真实项目里非常有价值，比如日志收集、埋点上报、批量入库。用Sink来表达后，调用方不需要自己管理“攒批”逻辑。

3.2 状态机设计

它需要维护几个状态：

• 当前批次的缓冲：Vec<String>
• 正在执行的提交Future：Option<BoxFuture<...>>
• 批次上限
• 是否已关闭

协议映射如下：

• poll_ready：如果正在提交，先驱动提交Future；如果提交失败，返回错误；如果当前缓冲已经达到上限，说明Sink已经没有更多容量，返回Pending；否则返回Ready(Ok(()))。
• start_send：把元素推入缓冲；如果缓冲满了，立刻创建一个提交Future。
• poll_flush：如果缓冲里还有数据，先取出创建提交Future；然后驱动提交Future直到完成。
• poll_close：标记关闭，然后执行一次完整的poll_flush，确保残留数据全部提交。

3.3 完整代码与逐步讲解

use futures::future::BoxFuture; use futures::ready; use futures::sink::Sink; use std::pin::Pin; use std::task::{Context, Poll}; #[derive(Debug, Clone, PartialEq, Eq)] pub enum BatchError { AlreadyClosed, SubmitFailed(String), } pub struct BatchSink<F> where F: FnMut(Vec<String>) -> BoxFuture<'static, Result<(), BatchError>>, { submit: F, batch: Vec<String>, flushing: Option<BoxFuture<'static, Result<(), BatchError>>>, max: usize, closed: bool, } impl<F> BatchSink<F> where F: FnMut(Vec<String>) -> BoxFuture<'static, Result<(), BatchError>>, { pub fn new(submit: F, max: usize) -> Self { Self { submit, batch: Vec::new(), flushing: None, max, closed: false, } } } impl<F> Sink<String> for BatchSink<F> where F: FnMut(Vec<String>) -> BoxFuture<'static, Result<(), BatchError>>, { type Error = BatchError; fn poll_ready( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { if self.closed { return Poll::Ready(Err(BatchError::AlreadyClosed)); } let this = &mut *self; // 如果有一批正在提交，先尝试推进它 if let Some(fut) = this.flushing.as_mut() { match ready!(Pin::new(fut).poll(cx)) { Ok(()) => { this.flushing = None; } Err(e) => { this.flushing = None; return Poll::Ready(Err(e)); } } } if this.batch.len() >= this.max { return Poll::Pending; } Poll::Ready(Ok(())) } fn start_send( mut self: Pin<&mut Self>, item: String, ) -> Result<(), Self::Error> { if self.closed { return Err(BatchError::AlreadyClosed); } let this = &mut *self; this.batch.push(item); // 攒满一批，自动发起提交 if this.batch.len() >= this.max && this.flushing.is_none() { let batch = std::mem::take(&mut this.batch); this.flushing = Some((this.submit)(batch)); } Ok(()) } fn poll_flush( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { let this = &mut *self; // 如果缓冲里还有残留数据，先打包成提交Future if this.flushing.is_none() && !this.batch.is_empty() { let batch = std::mem::take(&mut this.batch); this.flushing = Some((this.submit)(batch)); } if let Some(fut) = this.flushing.as_mut() { let result = ready!(Pin::new(fut).poll(cx)); this.flushing = None; return Poll::Ready(result); } Poll::Ready(Ok(())) } fn poll_close( mut self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>> { if !self.closed { self.closed = true; } self.poll_flush(cx) } }

这段代码里有一个比较微妙的地方：poll_ready里我驱动了正在进行的提交Future。这意味着即使调用者一直没有调用poll_flush，只靠不断poll_ready+start_send，提交也会持续推进。很多Sink实现不会这么做，它们只在poll_flush里推进提交，但那样会有个隐患：如果调用者从不调用flush，批量的数据就会一直积压。让poll_ready顺手驱动提交，能降低这种风险。

还需要注意Pin::new(fut).poll(cx)为什么安全。BoxFuture本质上是Pin<Box<dyn Future + Send>>，Box里的dyn Future不需要移动，所以BoxFuture自身实现了Unpin，这里用Pin::new包一下就能直接poll，不会违反Pin的约束。

测试这个Sink也很简单：

use futures::sink::SinkExt; fn main() { let submitted = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let sink = BatchSink::new( { let submitted = std::sync::Arc::clone(&submitted); move |batch: Vec<String>| { let submitted = std::sync::Arc::clone(&submitted); Box::pin(async move { submitted.lock().unwrap().extend(batch); Ok(()) }) } }, 2, ); futures::executor::block_on(async move { let mut sink = sink; sink.send("a".to_string()).await.unwrap(); sink.send("b".to_string()).await.unwrap(); sink.close().await.unwrap(); }); let guard = submitted.lock().unwrap(); assert_eq!(&*guard, &["a".to_string(), "b".to_string()]); }

这个测试里我故意把max设为2，发送两条消息后close，两条消息应该被作为一个批次提交。

4. 实战：把 Sink 用进一个真实网络服务

4.1 场景设定：广播服务

假设一个聊天服务，每个TCP连接上来之后会被分配一个下行通道。服务端需要把收到的聊天消息广播给所有在线订阅者。这个下行通道用一个futures::channel::mpsc::Sender<String>来表示，它天然实现了Sink<String>，所以我们可以直接把它当成Sink来操作。

use futures::channel::mpsc; use futures::sink::SinkExt; use futures::stream::StreamExt; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; type SubscriberId = u64; type SharedSubscribers = Arc<RwLock<HashMap<SubscriberId, mpsc::Sender<String>>>>; async fn broadcast( subscribers: &SharedSubscribers, message: String, ) { let mut map = subscribers.write().await; let mut dead = Vec::new(); for (id, sender) in map.iter_mut() { // 关键：这里用的就是 SinkExt::send // 如果对方连接已经关闭，sender.send会返回错误，顺手收集起来 if let Err(e) = sender.send(message.clone()).await { log::warn!("subscriber {} send failed: {}", id, e); dead.push(*id); } } for id in dead { map.remove(&id); } }

4.2 用 send_all 把 Stream 直接灌进 Sink

如果数据源本身是一个Stream，最省事的做法是send_all。比如我从某个上游数据源拉取一批事件，解析后直接写入Sink：

use futures::stream::{self, StreamExt}; use futures::sink::SinkExt; async fn load_events_into_sink( mut sink: impl futures::sink::Sink<String, Error = MyError> + Unpin, ) -> Result<(), MyError> { // 模拟从数据库或消息队列拉取的事件流 let event_stream = stream::iter(vec![ "event-1".to_string(), "event-2".to_string(), "event-3".to_string(), ]) .map(Ok::<_, MyError>); let mut event_stream = Box::pin(event_stream); sink.send_all(&mut event_stream).await?; Ok(()) }

send_all会自动完成“从Stream取一个元素，Sink拿一个元素”的背压匹配，不会像collect那样先把整个流都收进内存再统一发送。

4.3 实战中的观察

我在那个广播服务里一开始用的是遍历sender.send()，后来压测时发现一个问题：每一个send内部都包含一次poll_flush，也就是说每条消息都会尝试把数据刷到TCP层。如果消息量很大，这种频繁flush会让系统调用数量暴涨。

改进办法是引入SinkExt::buffer，给每个下行通道加一个小容量缓冲：

let sender = sender.buffer(16);

buffer(16)的意思是Sink内部最多缓存16条还来不及发给底层channel的消息，poll_ready会等到缓冲有空位才返回Ready。这样send不再每次都触发flush，而是攒在缓冲里，由底层channel自然调度。代价是背压被削弱了，如果消息持续大量涌入，缓冲里的消息可能积压到16条，内存占用会上升一点，但在可控范围。

改完之后效果很明显，同样压测条件下，火焰图里的flush调用占比从30%降到了5%左右。

5. 踩坑记录：Sink 实际项目中那些不好查的 bug

5.1 poll_ready 返回 Pending 后接着 start_send，直接炸穿

这是一个协议级错误。某些Sink实现内部有严格的状态检查，如果你在poll_ready还没返回Ready时就调用start_send，轻则数据发送顺序错乱，重则panic。

排查思路：不要在业务代码里手动拆开poll_ready和start_send调用。直接用SinkExt::send或send_all，它们内部会正确维护协议顺序。如果你确实需要自己控制发送节奏，也要严格按照“Ready后发送，然后再次Ready”的循环来。

5.2 flush 与 close 的先后

close语义上等于“flushed之后再也不接收数据”。很多人在关闭一个Sink时只调用close，以为它会自动flush，但其实close是“先flush再关闭”的组合操作，Sinktrait本身就要求实现者保证这一点。

如果你手工先掉flush再掉close，注意flush之后Sink还处于可用状态，你依然可以继续send，这不一定是bug，但可能造成“明明close了，数据却还在路上”的错觉。对于我自己写的BatchSink，poll_close会置closed = true再做flush，再调用send就会返回AlreadyClosed，这是符合直觉的安全行为。

5.3 错误类型混乱是最隐性的坑

Sink::Error和Stream::Error经常是两个完全不同的类型。使用send_all时，send_all要求Stream的Item是Result<Item, Sink::Error>，也就是Stream的每个元素要能直接匹配Sink的错误类型。这很容易在类型推导上报出一大堆看不懂的错。

一个实用的做法是用fn sink_map_err把Sink的错误转换成统一业务错误类型，然后再接入send_all。如果你不需要精细的错误处理，直接把Stream映射成Ok::<_, MyError>，让MyError和Sink::Error保持一致，是最省心的方案。

5.4 Pin 与发送 Future 的纠缠

手写Sink时，如果字段需要放一个Future，最好直接存BoxFuture或者Pin<Box<dyn Future>>，而不是存一个泛型Fut然后自己管理Pin。BoxFuture内部已经做了Pin装箱，你只需要在处理时注意把&mut BoxFuture转成Pin<&mut BoxFuture>再poll。这样一来，你的Sink几乎总是Unpin的，外部使用也更方便。

5.5 问题速查表

6. 现实场景选型：什么时候该用Sink，什么时候别用

6.1 通道类型对比

实际开发中，Sink的底层多数是一个channel或者一个TCP连接。不同通信原生的Sink支持程度差别很大：

6.2 什么时候不要用Sink

Sink不是万能抽象。如果你只需要一次性把数据塞给某个后台任务，而发送方完全不关心何时被消费，用tokio::spawn+UnboundedSender就够了，引入Sink反而增加心智负担。

反过来，只要你的发送方向需要“多条数据按照背压节奏逐个送达”，或者你需要在发送前反复做转换/过滤/扇出，那Sink就是比裸channel更高一层的工具箱。它最大的价值不是性能，而是让发送方向的代码拥有了和接收方向同等级的组合能力。

我个人在实际项目里体会最深的一点是：一旦你用SinkExt把发送方向也组合化，代码里那种“一边接收流、一边手工往多处发送”的割裂感会明显消失。以后遇到要同时给多个下游发同一份数据的场景，我会先想到fanout，而不是再搓一层循环加队列。你也可以从一个小项目开始试着把某条发送链路替换成Sink实现，感受一下这种抽象给你带来的约束，原本是坏事，但在这里恰好是防止你写出混乱状态机的最大帮手。