使用select宏从channel读取时调用next()函数出现错误的问题

报错原因 #

1. 缺少 trait 导入：next() 方法是由 StreamExt trait 为所有 Stream 类型提供的扩展方法，未导入该 trait 时编译器无法识别。
2. 同步通道不兼容异步流：std::sync::mpsc 是同步阻塞通道，其 iter() 方法会阻塞线程等待消息，直接转为 futures::stream::Iter 后在异步执行器中使用会阻塞整个 runtime，违反异步非阻塞的设计原则。

修正方案（不使用 tokio） #

改用 futures 库原生的异步通道 futures::channel::mpsc，并导入 StreamExt trait 启用流操作方法：

use futures::{select, stream::StreamExt, channel::mpsc};
use std::thread;
use std::time::Duration;

fn main() {
    // 创建带缓冲区的异步通道
    let (tx1, mut rx1) = mpsc::channel(1);
    let (tx2, mut rx2) = mpsc::channel(1);
    
    thread::spawn(move || {
        thread::sleep(Duration::from_secs(2));
        // 异步通道的send是异步方法，用block_on阻塞完成发送
        futures::executor::block_on(tx1.send("Message from thread 1")).unwrap();
    });
    
    thread::spawn(move || {
        thread::sleep(Duration::from_secs(1));
        futures::executor::block_on(tx2.send("Message from thread 2")).unwrap();
    });
    
    futures::executor::block_on(async {
        loop {
            select! {
                msg = rx1.next() => {
                    if let Some(msg) = msg {
                        println!("Received on rx1: {:?}", msg);
                    } else {
                        rx1.close();
                    }
                },
                msg = rx2.next() => {
                    if let Some(msg) = msg {
                        println!("Received on rx2: {:?}", msg);
                    } else {
                        rx2.close();
                    }
                },
                complete => break,
            }
        }
    });
}

修改说明 #

• 替换同步通道为 futures::channel::mpsc 异步通道，原生支持异步流操作，无需手动转换同步迭代器。
• 导入 StreamExt trait，解锁 Stream 类型的 next() 方法。
• 同步线程中使用 futures::executor::block_on 完成异步通道的发送操作。
• 处理通道关闭的 None 消息，关闭对应流以确保最终触发 complete 分支退出循环。

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