工作线程通过一个通道向主线程发送许多消息，但只有第一个被传递

Question

我一直在尝试扩展本书中多线程 Web 服务器章节中的线程池示例。原始示例工作正常，并通过 spsc 通道（入口）正确地将消息分派给工作人员，但现在我想通过 mpsc 通道（出口）从工作线程返回值（字符串）。不知何故，出口通道只发送一条消息而不是 10 条。egress_tx.send() 似乎执行了 10 次，但 egress_rx.recv() 只给我一条消息，然后程序完成（即没有死锁等）。工作线程在 Drop trait 实现中正确终止（此代码未显示）。对于调试此类问题的任何建议，我将不胜感激：设置断点 ar recv() 并尝试在其内部找到有意义的东西并没有太大帮助。

type Job = Box<dyn FnOnce(usize) -> String + Send + 'static>;
enum Message {
    Run(Job),
    Halt,
}
struct Worker {
    id: usize,
    thread: Option<thread::JoinHandle<()>>,
}
pub struct ThreadPool {
    workers: Vec<Worker>,
    ingress_tx: Sender<Message>,
    pub egress_rx: Receiver<String>
}

impl Worker {
    fn new(id: usize, rx: Arc<Mutex<mpsc::Receiver<Message>>>, tx: mpsc::Sender<String>) -> Worker {
        let thread = thread::spawn(move ||
            loop {
                let msg = rx.lock().unwrap().recv().unwrap();
                match msg {
                    Message::Run(job) => {
                        let s = job(id);
                        println!("Sending \"{}\"", s);
                        tx.send(s).unwrap();
                    },
                    Message::Halt => break,
                }
            }
        );
        Worker {id, thread: Some(thread)}
    }
}
impl ThreadPool {
    pub fn new(size: usize) -> Result<ThreadPool, ThreadPoolError> {
        if size <= 0 {
            return Err(ThreadPoolError::ZeroSizedPool)
        }
        let (ingress_tx, ingress_rx) = mpsc::channel();
        let ingress_rx = Arc::new(Mutex::new(ingress_rx));
        
        let (egress_tx, egress_rx) = mpsc::channel();
    
        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            workers.push(Worker::new(id, ingress_rx.clone(), egress_tx.clone()));
        }
        
        Ok(ThreadPool {workers, ingress_tx, egress_rx})
    }
    pub fn execute<F>(&self, f: F)
        where F: FnOnce(usize) -> String + Send + 'static
    {
        let j = Box::new(f);
        self.ingress_tx.send(Message::Run(j)).unwrap();
    }
}

fn run_me(id: usize, i: usize) -> String {
    format!("Worker {} is processing tile {}...", id, i).to_string()
}

#[cfg(test)]
mod threadpool_tests {
    use super::*;
    #[test]
    fn tp_test() {
        let tpool = ThreadPool::new(4).expect("Cannot create threadpool");
        
        for i in 0..10 {
            let closure = move |worker_id| run_me(worker_id, i);
            tpool.execute(closure);
        }
        for s in tpool.egress_rx.recv() {
            println!("{}", s);
        }        
    }
}

输出是：

Sending "Worker 0 is processing tile 0..."
Sending "Worker 0 is processing tile 2..."
Sending "Worker 3 is processing tile 1..."
Sending "Worker 3 is processing tile 4..."
Sending "Worker 2 is processing tile 3..."
Sending "Worker 2 is processing tile 6..."
Sending "Worker 1 is processing tile 5..."
Sending "Worker 0 is processing tile 7..."
Sending "Worker 0 is processing tile 9..."
Sending "Worker 3 is processing tile 8..."
Receiving "Worker 0 is processing tile 0..."
Process finished with exit code 0

Answer 1

在您的代码中，您有for s in tpool.egress_rx.recv()，它并没有完全按照您的意愿行事。不是迭代通道接收到的值，而是接收一个元素（包装在 Result 中）然后对其进行迭代，因为 Result 实现 IntoIterator 以迭代成功值（或者没有，如果它包含一个错误）。

只需将其更改为 for s in tpool.egress_rx 即可解决此问题，因为频道也实现了 IntoIterator。