bigywater.tech
Rust · 102

异步 Channel

tokio 有 mpsc、oneshot、broadcast、watch 四种常用通道,各自语义很不一样:满了会怎样、慢接收者会怎样、能不能多发。这篇每一种都用一个小程序跑出真实输出。

背压
有界 mpsc 满了之后 send().await 挂起,生产者被消费者拖住
100000 条
无界 mpsc 里积压的消息数
Lagged(4)
broadcast 慢订阅者错过 4 条的返回

有界 mpsc 与背压

实测 · 本机执行输出
    [  0 ms] 生产者 send(1) 返回
    [  0 ms] 生产者 send(2) 返回
    [  0 ms] 消费者 收到 1(处理 50 ms)
    [  0 ms] 生产者 send(3) 返回
    [ 60 ms] 消费者 收到 2(处理 50 ms)
    [ 60 ms] 生产者 send(4) 返回
    [110 ms] 消费者 收到 3(处理 50 ms)
    [110 ms] 生产者 send(5) 返回
    [170 ms] 消费者 收到 4(处理 50 ms)
    [170 ms] 生产者 send(6) 返回
    [220 ms] 消费者 收到 5(处理 50 ms)
    [280 ms] 消费者 收到 6(处理 50 ms)

标准库同步通道见Channel 消息传递与mpsc。

无界 mpsc

实测 · 本机执行输出
    生产者发完 10 万条用时 15 ms(同步 send,不 await);此时队列里积压了 100000 条
    消费者取走 100000 条

oneshot

实测 · 本机执行输出
    收到 Ok("结果")
    发送端 drop 后接收: Err(RecvError(()))
    接收端 drop 后发送: Err(5)(值被退回)

broadcast

实测 · 本机执行输出
    两个接收者都收到 1..=3: r1 = Ok(1) Ok(2) Ok(3)  r2 = Ok(1) Ok(2) Ok(3)
    慢接收者 recv: Err(Lagged(4))
    之后继续 recv: Ok(14) Ok(15) Ok(16) Ok(17)

取消安全相关的 recv 见select! 取消竞争。

watch

实测 · 本机执行输出
    连发 v1 v2 v3 之后,接收者看到的当前值: v3(中间的 v1 v2 被跳过)
    changed() 返回: Ok("v3");再次 changed 之前没有新值
    没有新值时 changed() 等待 30 ms: 超时(一直在等新值)

分步演示

异步通道
本步的对象

    本文用到的工具与边界

    • 本机 macOS 27 arm64 / rustc 1.98.1(brew)/ cargo 1.98.1;单个文件用 rustc 直接编译,需要依赖时用 cargo(经代理从 crates.io 下载);计时用 std::time::Instant,数字每次运行会有小幅抖动。tokio 1.x(rt、time、macros、sync)
    • 没有验证:多生产者多消费者的 async-channel / flume、有界 mpsc 的 permit(reserve)用法、broadcast 在大量订阅者下的性能、watch 的 borrow 持锁时间、无界通道积压对内存的实际占用
    京ICP备2026067502号-1京公网安备11010502063487号