如何正确退出 mpsc::Receiver 上的线程阻塞

Question

impl A {
    fn new() -> (A, std::sync::mpsc::Receiver<Data>) {
        let (sender, receiver) = std::sync::mpsc::channel();

        let objA = A { sender: sender, }; // A spawns threads, clones and uses sender etc
        (objA, receiver)
    }
}

impl B {
    fn new() -> B {
        let (objA, receiver) = A::new();

        B {
            a: objA,
            join_handle: Some(std::thread::spwan(move || {
                loop {
                    match receiver.recv() {
                        Ok(data) => /* Do Something, inform main thread etc */,
                        Err(_) => break,
                    }
                }
            })),
        }
    }
}

impl Drop for B {
    fn drop(&mut self) {
        // Want to do something like "sender.close()/receiver.close()" etc so that the following
        // thread joins. But there is no such function. How do i break the following thread ?
        self.join_handle().take().unwrap().join().unwrap();
    }
}

在这种情况下有没有办法干净地退出？问题是，当接收者或发送者中的任何一个被丢弃时，另一个会嗅到这个并给出错误。在receiver 的情况下，它将被唤醒并产生错误，在这种情况下，我将跳出上面的无限和阻塞循环。但是，我如何明确地使用通道的这个属性来做到这一点，而不使用与try_recv()etc 一起使用的其他标志，并确定性地干净地退出我的线程？

Answer 1

您可以创建一个新频道并将您的实际发件人与虚拟发件人交换。然后您可以删除您的发件人并因此加入该线程：

impl Drop for B {
    fn drop(&mut self) {
        let (s, _) = channel();
        drop(replace(&mut self.a.sender, s));
        self.join_handle.take().unwrap().join().unwrap();
    }
}

在游戏围栏中尝试一下：http://is.gd/y7A9L0

我不知道创建和立即删除频道的开销是多少，但它不是免费的，也不太可能被优化（那里有一个Arc）。

---

附带说明，您在receiver.recv() 上匹配的无限循环可以使用Receiver::iter 方法替换为for 循环：

for _ in receiver.iter() {
    // do something with the value
}

Answer 2

为什么不发送特定消息来关闭此线程？我不知道你的数据是什么，但大多数时候它可能是一个枚举，并在你的接收中添加一个像“MyData::Shutdown”这样的枚举变体，你可以简单地跳出循环。

Answer 3

您可以将B 类型的a 字段包装在Option 中。这样，在Drop::drop 方法中，您可以执行drop(self.a.take())，它将用None 替换字段并删除发件人。这将关闭频道，您的线程现在可以正确加入。