Rust crossbeam多消费者并行处理通道消息异常问题

你的代码有两处核心逻辑错误，直接导致线程提前终止、并行失效：

• 线程启动逻辑错误：for _ in [0..3] 是对长度为1的数组迭代（数组内唯一元素是范围对象0..3），实际只会启动1个工作线程，根本没有创建3个并行线程。
• 消息消费逻辑错误：每个启动的线程仅调用1次recv()，拿到1条消息处理完就直接退出，不会持续消费通道内剩余消息；同时你把通道耗尽、发送端全关闭时recv()返回的正常终止信号当成错误触发panic，不符合多消费者通道的使用逻辑。

use std::thread::sleep;
use std::time::Duration;
use crossbeam_utils::thread::scope;

// 重负载处理逻辑
fn process(s: &str) {
    println!("receive: {:?}", s);
    sleep(Duration::from_secs(3));
}

fn main() {
    let files_to_process = vec!["file1.csv", "file2.csv", "file3.csv", "file4.csv", "file5.csv"];
    let (s, r) = crossbeam::channel::unbounded();
    for e in files_to_process {
        println!("sending: {:?}", e);
        s.send(e).unwrap();
    }

    // 提前drop发送端，所有消息发完后通道会触发关闭信号
    drop(s);

    scope(|scope| {
        // 直接迭代范围0..3，启动3个并行工作线程
        for worker_id in 0..3 {
            // 每个工作线程独立克隆一份通道接收器
            let worker_rx = r.clone();
            scope.spawn(move |_| {
                // 循环拉取消息，直到通道无剩余消息且所有发送端断开
                loop {
                    match worker_rx.recv() {
                        Ok(msg) => process(msg),
                        Err(_) => break, // 收到终止信号，正常退出线程
                    }
                }
            });
        }
    }).unwrap();
}

• 并行线程数修正：去掉范围外层的方括号，直接迭代0..3范围，会真正循环3次启动3个独立工作线程，实现并行处理。
• 消费逻辑修正：每个工作线程内部加循环，持续从通道拉取消息处理，只有当通道内无剩余消息、所有发送端被回收时，才正常退出线程，不会提前终止。
• 所有权处理修正：每个工作线程独立克隆一份接收器，通过move关键字把接收器所有权转移到对应线程内，避免跨线程所有权冲突。
• 错误处理修正：recv()返回错误时直接退出线程即可，这是crossbeam通道多消费者模式的标准终止写法，不需要触发panic。

运行修正后的代码，5个文件会被3个线程并行处理，总耗时约6秒（前3个文件第一批并行处理3秒，后2个文件第二批并行处理3秒），所有文件条目都会被正常打印。

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