2.2 多进程汇聚：mpsc 有界通道

2.2 多进程汇聚：mpsc 有界通道

上一节解决了「单个文件怎么读」，但真实的日志管道从来不止一个源头：一台机器上好几个服务在写日志，每台机器上又有一个采集器，采集完要往中心汇聚。汇聚这件事听起来就是「多发一收」，用 Rust 写起来第一反应都是 mpsc 通道。本篇把多源汇聚的拓扑、有界通道的Capacity选择、以及 sender 多了之后的各种坑讲清楚。

为什么是 mpsc 而不是 crossbeam

Rust 标准库的 std::sync::mpsc 与 tokio::sync::mpsc 都是多生产者单消费者通道，语义上正合适：N 个采集任务各自拿到一个 sender，消费端只有一个聚合任务。两者的区别在于消费端是否异步：tokio 的通道 recv() 是 .await 的，不阻塞工作线程；标准库的通道配线程池也行，但与 tokio 运行时混用要额外搭桥。日志管道整体跑在 tokio 上，选 tokio::sync::mpsc 最顺。

对比项	std::sync::mpsc	tokio::sync::mpsc	crossbeam-channel
与异步运行时契合	需搭桥	原生 await	需搭桥
有界/无界	都支持	都支持	都支持
依赖	标准库	tokio	第三方 crate
适用	纯线程模型	异步管道	极致性能场景

第三方通道在极限吞吐上确实有优势，但日志管道的瓶颈几乎从不在通道本身，而在下游的批处理与写入。引入额外依赖换取百分之几的通道性能，不如把精力花在攒批策略上，这个取舍在 2.4 会展开。

汇聚拓扑：先扇入再串行

多个采集源扇入（fan-in）到一个聚合点，是日志管道的标准形状。拓扑上有两种常见做法：

拓扑	形状	优点	坑
直连聚合	N sender → 1 receiver	结构简单	消费端压力大
分级汇聚	每机本地聚合 → 中心聚合	层次清晰、可限速	多一跳延迟

当天量日志全部直连一个中心 receiver 时，那个聚合任务的调度延迟会成为全链路尾巴。分级汇聚给每台机器一个本地聚合点，先做一次攒批与压缩再上行，骨干流量能降一个数量级。对于「每机一个采集器」的部署形态，分级汇聚几乎总是对的。

骨架代码大致如下：

// 多源扇入聚合
let (tx, mut rx) = tokio::sync::mpsc::channel::<LogLine>(capacity);

for path in sources {
    let tx = tx.clone();
    tokio::spawn(async move {
        let mut tail = Tail::open(path).await;
        while let Some(line) = tail.next_line().await {
            if tx.send(line).await.is_err() {
                break; // 聚合端已关闭
            }
        }
    });
}

// 唯一的聚合消费端
while let Some(line) = rx.recv().await {
    batch.push(line);
}

注意 tx.clone() 的含义：每个采集任务持有独立的 sender 句柄，通道内部用引用计数共享队列。最后一个 sender 被 drop 时，recv() 才返回 None，这是判断「所有源头都结束」的信号。

容量怎么选

有界通道的 capacity 是整个管道第一个需要拍板的数字。它不是越大越安全：容量大了，内存占用随着积压线性增长；容量小了，高峰期频繁进入背压。经验起点是「下游一个攒批周期的量」——如果下游每 100 毫秒刷一批、每批最多 5000 条，那么 capacity 设到 5000~10000 就够吸收波动。

容量	内存（每条 1KB 估）	背压频率	适用
1,000	~1 MB	高	内存极度受限
10,000	~10 MB	中	多数在线服务
100,000	~100 MB	低	尖峰流量明显

容量与背压的配合是通道设计的核心矛盾，背压怎么具体生效见下一节 2.3 与 Node.js 云开发实战里对队列容量的同类讨论。

选定容量之后还有一个工程动作：把队列深度暴露成指标。通道对象本身不提供 len() 之类的查询（tokio 的 mpsc 早期版本有，后被移除），常见替代法是消费端每秒统计一次 recv 计数、与采集端 send 计数相减，差值就是近似积压。这个数字比通道容量本身更有告警价值：容量是静止的配置，积压是活的压力，两者差三个数量级的运维价值。

send 的三种结局

刚写 mpsc 的人常以为 send 只有成功一种结局，实际上有三种，每一种都对应一种线上故障：

send 结果	含义	采集端应对
Ok(())	入队成功	继续
Err(SendError)	聚合端已关闭	停止采集并告警
阻塞（有界满）	等待空位	背压生效

第二种最危险：聚合任务崩了，所有采集端傻傻阻塞在 send 上，外面看采集器还活着，日志其实一条没收。必须给 send 加超时或让采集端感知通道关闭，否则故障会被「进程都在线」的假象掩盖。监控章节会专门讲怎么暴露这类「隐性停摆」，监控指标设计可参考 Go 即时通讯实战中连接级心跳的做法。

多源有序性：别指望先到先得

多个 sender 往同一队列塞数据，接收端拿到的顺序完全由到达先后决定，与日志原始时间无关。跨源的「全局有序」是做不到的，也没必要追求：日志检索本来就按各自时间戳排序。真正要防的是源内乱序——同一个文件的两个 chunk 被并发读出又乱序到达。解法是源内保持单任务串行读取，不做源内并行。

排序需求	可实现性	手段
全局有序	不可行	放弃
源内有序	可行	单源单任务
近似全局	可行	按时间戳再排序

如果下游必须按时间合并多源日志，正确做法是在聚合后做一次小窗口的按时间戳重排（窗口 1~2 秒），而不是在通道层面追求有序。窗口重排的代价是多等一个窗口的延迟，换来的是合并结果可解释。

优雅关闭的顺序

关闭一条多源管道，顺序反了会丢数据。正确顺序是「先停采集、再等队列排空、最后关聚合」：每个采集任务 drop 自己的 sender，等 receiver 收到 None 后聚合任务把最后一批刷下去再退出。反过来先关聚合端，所有在途消息直接判负。

阶段	动作	标志
停源	各采集任务退出	sender 全 drop
排空	聚合继续 recv	recv 返回 None
刷尾	最后一批写入	聚合任务退出

这个顺序在下一篇的攒批逻辑里还会用到，崩溃恢复相关的内容见 4.4 的 WAL 重放。

踩坑提示

最常见的坑是忘记 clone sender 给最后一个任务：主任务持有原始 sender 不 drop，receiver 永远等不到 None，进程挂在 recv 上不退出。其次是在 async 任务里用阻塞式 send：标准库 mpsc 的 send 阻塞当前线程，在 tokio 任务里会卡住整个工作线程，必须换成 blocking_send 或直接用 tokio 通道。第三是通道容量与下游批大小不匹配：批大小比容量还大，攒批永远等不满，定时刷又把延迟拉长，两者要一起调。

小结

多源汇聚的标准形状是 fan-in 加分级聚合，通道选 tokio mpsc，容量按「一个攒批周期的量」起步。send 的三种结局都要处理，全局有序不可求，源内有序靠单任务串行，关闭顺序是「停源、排空、刷尾」。汇聚层是管道的咽喉，它畅不顺，决定后面持久化层看到的是平稳流还是脉冲流。

2.2 多进程汇聚：mpsc 有界通道

2.2 多进程汇聚：mpsc 有界通道

为什么是 mpsc 而不是 crossbeam

汇聚拓扑：先扇入再串行

容量怎么选

send 的三种结局

多源有序性：别指望先到先得

优雅关闭的顺序

踩坑提示

小结

延伸阅读