Tokio Broadcast仅接收单条消息而非全部的问题排查求助

你的代码存在几个关键问题，导致只能收到一条消息：

1. 阻塞睡眠破坏异步调度 #

thread::sleep是阻塞式睡眠，会占用整个Tokio工作线程，直接破坏异步任务的调度逻辑。当发送任务进入睡眠时，接收任务会被卡住，无法继续执行rx.recv().await来接收后续消息。

2. 变量名拼写错误 #

代码中tx.send(valu).unwrap();的valu是笔误，正确变量名应为value，这会直接导致编译错误，你能运行并收到一条消息应该是测试时临时修正了该问题，但这个错误必须修正才能保证代码正常工作。

3. 广播通道容量过小（潜在风险） #

你创建的广播通道容量仅为1，当发送速度超过接收速度时，旧消息会被新消息覆盖。虽然这不是当前只能收到一条的核心原因，但在高并发场景下会导致消息丢失，建议根据实际需求调整容量。

修复后的代码 #

use tokio::sync::broadcast;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = broadcast::channel(10); // 适当增大容量，避免消息覆盖
    let handle = tokio::spawn(async move {
        let mut value = 10;
        loop {
            value += 10;
            tx.send(value).unwrap(); // 修正变量名拼写
            sleep(Duration::from_secs(1)).await; // 使用Tokio异步睡眠
            if value >= 100 {
                break;
            }
        }
    });

    let handle1 = tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(value) => {
                    println!("=====> : {}", value);
                }
                Err(e) => {
                    println!("err : {}", e);
                    break;
                }
            }
        }
    });

    // 等待两个任务执行完成，避免程序提前退出
    handle.await.unwrap();
    handle1.await.unwrap();
}

关键修复点说明 #

• 替换阻塞睡眠为异步睡眠：tokio::time::sleep会主动让出工作线程，让Tokio调度器可以切换到接收任务执行，确保后续消息能被及时处理。
• 修正变量名：将valu改为value，保证代码编译通过。
• 调整通道容量：从1增大到10，降低消息被覆盖的概率。
• 添加任务等待：在主函数中等待两个子任务完成，避免程序提前终止导致后续消息无法接收。

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