Tokio广播通道在WebSocket流中频繁出现RecvErr::Lagged问题排查与优化

我开发了一个监听WebSocket流的程序，使用缓冲区大小为1的tokio::sync::broadcast::channel广播消息。WebSocket消息到达间隔的百分位数分布如下：

[5,25,50,75,95,99] // 百分位数
[7.42400000e+03, 2.48320000e+04, 1.03270400e+06, 8.12652800e+06, 1.00000000e+07, 8.18310554e+07] // 纳秒

此时广播接收器频繁出现RecvErr::Lagged错误，占比近20%。

我编写了如下测试代码：

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_tokio_broadcast() {
    use spin_sleep;
    env_logger::builder().format_timestamp_millis().init();
    let args: Vec<String> = env::args().collect();
    log::info!("{:?}", args);
    let duration_us: u64 = args[2].parse().unwrap();

    let (tx, mut rx) = tokio::sync::broadcast::channel(1);
    std::thread::spawn(move || {
        loop {
            spin_sleep::sleep(Duration::from_micros(duration_us));
            log::info!("send at {}", now_ns());
            tx.send(1).unwrap();
        }
    });
    let mut c: u64 = 0;
    loop {
        let v = rx.recv().await;
        c += 1;
        match v {
            Ok(v) => {
                log::info!("{} normal {}", now_ns(), c);
            }
            Err(err) => {
                match err {
                    RecvError::Closed => {
                        panic!();
                    }
                    RecvError::Lagged(count) => {
                        log::info!("{} lag {} count {}", now_ns(),c, count);
                    }
                }
            }
        }
    }
}

执行命令：

RUST_LOG=info cargo test --release test_tokio_broadcast -- 20 --nocapture 2>&1 | tee log

即使无额外负载，仍出现1%的Lagged错误率。

疑问 #

1. 为何WebSocket场景下错误率更高（仅替换生产者为WebSocket流）？
2. 如何优化？
3. 这是Tokio广播通道的极限吗？
4. 有无替代方案？

一、WebSocket场景错误率更高的原因 #

1. 调度延迟差异：测试代码用std::thread+spin_sleep做生产者，是主动休眠后唤醒，调度延迟极低；而WebSocket消息接收是异步任务，依赖Tokio事件循环，当事件循环有其他任务或IO操作时，消息处理的调度延迟会增加，导致接收器还未处理完上一条消息，新消息已到来触发Lagged。
2. 消息到达突发性：从百分位数数据看，消息间隔波动极大（50分位1ms，99分位高达81ms），但存在大量短间隔（如5分位7.4μs），这种突发密集消息更容易填满仅为1的缓冲区；测试代码是固定20μs间隔，规律性强，接收器更容易跟上节奏。
3. 实际处理开销：WebSocket消息的解析、预处理本身有额外开销，相比测试代码直接发送简单的1，实际场景中生产者的处理耗时更长，可能导致消息发送时机更集中，加重缓冲区溢出。

二、优化方案 #

1. 增大缓冲区大小 #

tokio::sync::broadcast::channel的缓冲区用于暂存未被所有接收器处理的最新消息，增大缓冲区可容忍短时间消息突发。根据你的数据，可将缓冲区设为8或16，覆盖大部分短间隔的消息密集场景：

// 调整缓冲区大小
let (tx, mut rx) = tokio::sync::broadcast::channel(8);

2. 轻量化接收器处理逻辑 #

确保接收器recv()后的处理逻辑尽可能简洁，若处理耗时较长，可将其放到单独的Tokio任务中，让接收器尽快回到接收状态：

loop {
    match rx.recv().await {
        Ok(msg) => {
            // 将处理逻辑丢到后台任务执行
            tokio::spawn(async move {
                // 处理msg的业务逻辑
            });
        }
        Err(RecvError::Lagged(count)) => {
            log::warn!("Lagged {} messages", count);
            // 业务允许的话可直接跳过，需补全则结合业务逻辑处理
        }
        _ => panic!("Channel closed"),
    }
}

3. 调整Tokio线程池配置 #

测试代码指定了worker_threads = 2，生产环境可根据CPU核心数调整线程池大小，让事件循环有足够线程处理WebSocket消息和广播接收任务，减少调度延迟：

use tokio::runtime::Builder;
use num_cpus;

let runtime = Builder::new_multi_thread()
    .worker_threads(num_cpus::get()) // 使用全部CPU核心
    .enable_all()
    .build()
    .unwrap();
runtime.block_on(your_main_task());

4. 业务层面过滤消息 #

若业务允许，对WebSocket高频消息做合并或过滤，比如将短时间内的重复消息合并为一条，从源头降低广播通道的消息量，减轻缓冲区压力。

三、Tokio广播通道的极限？ #

这不是Tokio广播通道的极限，而是当前配置（缓冲区1）与场景不匹配。Tokio的broadcast通道设计用于一对多消息广播、允许丢消息的场景，核心逻辑是“保留最新的N条消息，旧消息被新消息覆盖”，当接收器处理速度跟不上生产者时，必然触发Lagged。如果你的场景需要严格的消息投递，broadcast的设计本身就不适合。

四、替代方案 #

1. 使用tokio::sync::mpsc多消费者模式 #

若需要确保消息不丢失，可为每个接收器创建单独的mpsc通道，但缺点是消费者数量较多时，内存占用会增加，因为每个通道都要缓存消息。

2. 自定义环形缓冲区+通知机制 #

基于tokio::sync::Notify和环形缓冲区实现自定义广播逻辑，可灵活控制消息保留策略，比如保留固定数量的历史消息或按时间窗口保留消息，适合需要一定消息回溯能力的场景。

3. 第三方库替代 #

比如async_broadcast，它提供了更灵活的配置，支持可定制的溢出策略（阻塞生产者或丢弃消息），相比Tokio原生broadcast有更多定制选项。

内容的提问来源于stack exchange，提问作者YNX