使用 Rust 和 Tokio 枭构高并发日志采集服务的架构设计与性能调优记录

起因

事情是这样的，我们线上大概有二三十台机器，日志统一用 Filebeat 采集丢到 Kafka，再由下游消费。之前一直相安无事，直到有一天运营搞活动，流量翻了几倍，几台采集机的 Filebeat 直接把内存吃满触发 OOM 被 kill，重启过后又得从 registry 重新对位，高峰期就这么丢了十几分钟的日志。

当时的想法是加机器，但 leader 说先看看能不能省点钱... 行吧。

本来想直接换 Vector 的，试了一下确实不错，但我们有个需求是要在采集端对日志里的手机号做一层脱敏，试着用 VRL 写了写正则替换，这玩意儿的语法是真的别扭，最后索性决定自己用 Rust 写一个，正好也练练手。前后断断续续折腾了三周左右，这里把架构和踩的坑都记下来，免得以后自己都忘了。

整体架构：一条流水线串起来

整体思路不复杂，把采集过程拆成四个阶段，中间用有界 channel 连起来：

接收(receive) --> 解析(parse) --> 聚合(batch) --> 发送(ship)

接收端同时支持 TCP 和 UDP，TCP 走 tokio::net::TcpListener，每来一个连接就 spawn 一个读任务；解析这边开了 CPU 核数个任务，和聚合互不拖累。关键在于中间这两条通道：

let (tx_raw, rx_raw) = flume::bounded::<RawLine>(16384);
let (tx_batch, rx_batch) = mpsc::channel::<Vec<LogEvent>>(64);

这里有个小细节：tokio 自带的 mpsc::Receiver 是不能直接 clone 的，想搞多个消费者就得 Arc<Mutex<Receiver>> 包一层，锁又容易成为新瓶颈，后来干脆换了 flume，天生支持多消费者，省心。

背压就靠有界 channel 天然实现：下游慢了，通道满了，上游自然发不进去。TCP 场景下这没问题，客户端写缓冲满了它自己会慢下来；UDP 就比较麻烦，没有流控，通道满了只能丢弃，丢之前打个计数器，方便后面观察。

第一版踩过的坑

第一个坑：偷懒用了 unbounded_channel。一开始想当然觉得“日志嘛，先进内存队列再说”，压测一打，内存曲线直接起飞，5 分钟从 50MB 涨到 2 个多 G。改成 bounded 之后整个世界都清净了。教训就是：在不确定下游消费速度的情况下，永远别用无界队列。

第二个坑比较隐蔽：我在解析阶段用 std::fs::read_to_string 去读一个热更新的规则文件（脱敏规则, 支持运行时改）。现象是 QPS 一高，服务就像假死一样，TCP 连接还在就是没响应，ping 也没问题。折腾了半天才发现，tokio 文档里白纸黑字写了：不要在 async 上下文里做同步阻塞操作，尤其是文件 IO。读文件那一瞬间会把整个 worker 线程卡住，同一个线程上跑的其他任务全得陪着等。

解决办法是换 tokio::fs，或者用 spawn_blocking 包起来：

let content = tokio::fs::read_to_string(&rule_path).await?;

还有一个不算坑但值得说的：热路径上少打日志。之前每解析 1000 条打一行 debug，看着没啥，压测一对比能差出 3% 的 CPU，全花在格式化字符串上了。

批量发送这块的调优

发送端是性能大头。一开始我是一条一条往 Kafka 发，单核跑满也就 3 万条/秒，改成攒批之后直接翻了五倍。核心逻辑就是用 tokio::select! 同时等新数据和定时器：

const MAX_BATCH: usize = 1024;
const FLUSH_INTERVAL: Duration = Duration::from_millis(200);

let mut batch: Vec<LogEvent> = Vec::with_capacity(MAX_BATCH);
let mut tick = tokio::time::sleep(FLUSH_INTERVAL);
tokio::pin!(tick);

loop {
    tokio::select! {
        item = rx_batch.recv() => {
            match item {
                Ok(event) => {
                    batch.push(event);
                    if batch.len() >= MAX_BATCH {
                        shipper.send(&batch).await;
                        batch.clear();
                        tick.as_mut().reset(tokio::time::sleep(FLUSH_INTERVAL));
                    }
                }
                Err(_) => break, // channel 关闭，收尾退出
            }
        }
        _ = &mut tick => {
            if !batch.is_empty() {
                shipper.send(&batch).await;
                batch.clear();
            }
            tick.as_mut().reset(tokio::time::sleep(FLUSH_INTERVAL));
        }
    }
}

几个参数是反复试出来的：

MAX_BATCH 从 128 一路加到 1024，吞吐一路涨，但加到 4096 就几乎没变化了，还平白增加延迟，所以停在 1024。
FLUSH_INTERVAL 200ms 是个比较平衡的值，再小批量攒不满，再大端上延迟就明显了，下游告警都开始骂人了。
Kafka producer 侧记得开 linger.ms = 5 和 compression = zstd，日志这种重复度高的文本压完能到原来的 1/8 左右。

另外发送端如果走 HTTP，记得开 TCP_NODELAY，不然 Nagle 算法跟攒批逻辑撞在一起，延迟会莫名其妙地抖。

压测数据和对比

压测环境是一台 4C8G 的云主机，写了个简单的 generator 通过 TCP 灌日志，每条 200 字节左右，格式和线上差不多。最终数据（单机）：

稳定吞吐：15 万条/秒（约 30MB/s），CPU 60% 左右
内存：稳定在 60MB 上下，跑了一晚上没有增长
同样的量灌给 Filebeat：吞吐差不多，但内存要吃到 300MB 往上

用 flamegraph 看了一下热点，居然不在 Tokio 调度上，而在 serde_json 的反序列化——我们的日志是固定分隔符格式，根本不需要 JSON 解析，纯粹是图省事先转成结构体再取字段。后来改成手写 split 按位置取字段，这一项就把吞吐从 11 万提到了 15 万。所以说性能调优这事，先看火焰图再动手，比瞎猜靠谱多了。

目前的配置和还没做完的事

没做完的事也不少：本地磁盘缓冲（WAL）还没做，机器重启会丢最多一个 batch 窗口的数据，大概 200ms 的量，目前能接受但不算完美；停机处理做得也比较糙，就是简单等 channel 排空。服务在两台机器上灰度跑了两个多星期，除了那次我自己手滑改错配置重启了一下，没出过什么幺蛾子。目前配置长这样：

[server]
tcp_bind = "0.0.0.0:5140"
udp_bind = "0.0.0.0:5141"

[batch]
max_size = 1024 # <--- 调过好几轮，1024 比较甜
flush_interval_ms = 200

[rule]
mask_path = "/etc/logshipper/mask.toml"

[sink]
type = "kafka"
brokers = ["10.0.8.1:9092", "10.0.8.2:9092"]
topic = "app-logs"
compression = "zstd"

等这套东西在线上再稳一段时间，我把 WAL 和指标补齐，整理整理开源出来，到时候再写一篇续集。

使用 Rust 和 Tokio 构建高并发日志采集服务的架构设计与性能调优记录