使用橫梁通道和作用域執行緒的多執行緒代碼無休止地卡住

我有以下代碼使用容量（20）小于我想通過橫梁通道發送的資料總量（32）的有界通道。我的目標是使用多個發送執行緒 (8) 和一定數量 (4) 的數字，每個執行緒通過橫梁通道發送到單個接收器，并讓所有這些并行發生以優化效率。這只是我試圖解決的一個更大問題的一個小原型。但是，我擁有的代碼導致程式無休止地卡住并且永遠不會退出/超時。我也知道它為什么會發生 -r.iter()阻塞直到發送者被丟棄，這只發生在范圍之外。我嘗試了各種不起作用的方法：

1. 使用每個執行緒克隆發件人，然后洗掉它們（如您在評論中所見）

2. 使接收器代碼超出范圍，但這只是讓最終向量包含 20 長度而不是所需的 32。

fn main() {
    use crossbeam_channel::{unbounded, bounded};
    use crossbeam_utils::thread::scope;
    use itertools::Itertools;

    let (s, r) = bounded(20);
    let mut v = vec![];

    scope(|scope| {
        scope.spawn(|_| {
            for data in r.iter() {
                v.push(data);
            }
        });
        let _sender_threads = (0..8)
            .into_iter()
            .map(|_| {
                scope.spawn(|_| {
                    // let s_clone = s.clone();
                    for i in 0..4 {
                        // std::thread::sleep(std::time::Duration::from_millis(100));
                        match s.send(i) {
                            Ok(_) => {
                                // println!("sent i {:?}", i);
                                ()
                            },
                            Err(_)=> {
                                // println!("error sending i {:?}", i);
                                ()
                            }
                        };
                    }
                    // drop(s_clone);
                })
            })
            .collect_vec();
    }).expect("scope error.");
    drop(s);
    println!("{:?}", v);
}

uj5u.com熱心網友回復：

發生這種情況是因為s在作用域結束之前不會被洗掉，但是在所有執行緒退出之前，作用域不會結束，并且執行緒呼叫在被洗掉r.iter()之前不會退出s。這是典型的死鎖場景。

您需要s進入范圍，但只有在發送者執行緒全部退出后才能這樣做，因此您不能drop(s);以當前撰寫的方式進入范圍。

解決這個問題的最簡單方法是s為每個發送者執行緒克隆并將克隆移動到執行緒的閉包中，然后s放入主范圍。

fn main() {
    use crossbeam_channel::{unbounded, bounded};
    use crossbeam_utils::thread::scope;
    use itertools::Itertools;

    let (s, r) = bounded(20);
    let mut v = vec![];

    scope(|scope| {
        scope.spawn(|_| {
            for data in r.iter() {
                v.push(data);
            }
        });
        let _sender_threads = (0..8)
            .into_iter()
            .map(|_| {
                // ** Clone the sender, move it into the thread:
                let s = s.clone();
                scope.spawn(move |_| {
                    // let s_clone = s.clone();
                    for i in 0..4 {
                        // std::thread::sleep(std::time::Duration::from_millis(100));
                        match s.send(i) {
                            Ok(_) => {
                                // println!("sent i {:?}", i);
                                ()
                            },
                            Err(_)=> {
                                // println!("error sending i {:?}", i);
                                ()
                            }
                        };
                    }
                    // drop(s_clone);
                })
            })
            .collect_vec();

        // ** Drop the remaining sender.
        drop(s);
    }).expect("scope error.");
    println!("{:?}", v);
}

通過添加來注意let s = s.clone();對以下閉包的添加和更改，move以便閉包獲得克隆的所有權。然后我們drop(s)進入范圍。現在，一旦所有發送者執行緒都退出，通道的發送端將關閉，接收者的for回圈將終止。

（游樂場）

標籤：多线程 锈 横梁

上一篇：如何在特定執行緒上獲取物件？

下一篇：固定速率的ScheduledExecutorService計劃沒有像預期的那樣準確