tokio::mpsc::Sender进入循环后为何关闭？技术排查求助

你的代码出现通道提前关闭的核心原因有两个：

1. Oneshot通道的错误使用导致接收函数提前退出 #

在read_stream函数中，你创建了一个oneshot通道后直接丢弃了Sender端：

let (_, mut rx) = oneshot::channel::<()>();

Tokio的oneshot通道规则是：当所有Sender被销毁时，Receiver的recv()会立即返回错误。这导致你在select!中设置的timeout分支瞬间触发（因为rx.recv()已经处于就绪状态），read_stream函数直接return，MPSC通道的Receiver实例被销毁（drop）。

2. Tokio MPSC通道的关闭逻辑 #

Tokio的mpsc通道遵循：当所有Receiver实例被销毁时，Sender端会判定通道已关闭，因为没有接收方再处理消息。这就是为什么msg_stream函数开头检查通道未关闭，但第一次循环时就显示已关闭的原因——此时read_stream已经退出，Receiver已被销毁。

接收函数（修复超时逻辑） #

移除多余的oneshot通道，直接用sleep实现超时退出：

async fn read_stream(mut receiver: mpsc::Receiver<Message>) {
    println!("Receiver started");
    loop {
        tokio::select! {
            // 10秒无消息则退出接收逻辑
            _ = tokio::time::sleep(Duration::from_secs(10)) => {
                println!("Timeout reached, exiting receiver");
                return;
            }
            message = receiver.recv() => {
                match message {
                    Some(msg) => println!("Received message = {:?}", msg),
                    None => {
                        println!("All senders closed, exiting receiver");
                        return;
                    }
                }
            }
        }
    }
}

主函数（修正函数名调用） #

确保调用的是定义好的read_stream函数（你的原代码中写的是read(rx)，属于笔误）：

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<Message>(8);
    tokio::join!(msg_stream(tx), read_stream(rx));
}

可以在read_stream的开头和结尾添加打印语句，直观确认函数的生命周期：

async fn read_stream(mut receiver: mpsc::Receiver<Message>) {
    println!("Receiver started");
    // ... 原有逻辑 ...
    println!("Receiver exiting");
}

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