multithreading - 使用Mutex防止同时处理同一类型时进行多线程列表迭代

Question

我正在编写一个需要同时在多个线程上运行的应用程序。它将处理一长串项目，其中每个项目的一个属性是user_id。我试图确保属于同一user_id的项目永远不会在同一时间处理。这意味着运行子线程的闭包需要等待，直到没有其他线程正在为同一用户处理数据为止。
我不知道该如何解决。我的简化的当前示例如下所示:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use threadpool::ThreadPool;

fn main() {
    let pool = ThreadPool::new(num_cpus::get());

    let mut locks: HashMap<String, Mutex<bool>> = HashMap::new();
    let queue = Arc::new(vec![
        "1".to_string(),
        "1".to_string(),
        "2".to_string(),
        "1".to_string(),
        "3".to_string(),
    ]);
    let count = queue.len();

    for i in 0..count {
        let user_id = queue[i].clone();
        // Problem: cannot borrow `locks` as mutable more than once at a time
        // mutable borrow starts here in previous iteration of loop
        let lock = locks.entry(user_id).or_insert(Mutex::new(true));

        pool.execute(move || {
            // Wait until the user_id becomes free.
            lock.lock().unwrap();

            // Do stuff with user_id, but never process
            // the same user_id more than once at the same time.
            println!("{:?}", user_id);
        });
    }

    pool.join();
}

我试图保留一个 Mutex列表，然后使用它来等待 user_id变得免费，但是借位检查器不允许这样做。在我正在处理的实际应用程序中，队列项目和项目处理代码要复杂得多。
我不允许更改队列中项目的顺序(但由于等待锁定，因此将允许进行一些更改)。
如何解决这种情况？

Answer 1

首先， HashMap::entry() 使用 key ，因此由于您也想在闭包中使用它，因此需要克隆它，即.entry(user_id.clone())。
由于您需要在主线程和工作线程之间共享Mutex<bool>，因此同样需要将其包装在 Arc 中。您也可以使用 Entry::or_insert_with() ，因此除非需要，否则避免不必要地创建新的 Mutex 。

let mut locks: HashMap<String, Arc<Mutex<bool>>> = HashMap::new();
// ...
let lock = locks
    .entry(user_id.clone())
    .or_insert_with(|| Arc::new(Mutex::new(true)))
    .clone();

最后，您必须存储 lock() 返回的防护，否则它将立即释放。

let _guard = lock.lock().unwrap();

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use threadpool::ThreadPool;

fn main() {
    let pool = ThreadPool::new(num_cpus::get());

    let mut locks: HashMap<String, Arc<Mutex<bool>>> = HashMap::new();
    let queue = Arc::new(vec![
        "1".to_string(),
        "1".to_string(),
        "2".to_string(),
        "1".to_string(),
        "3".to_string(),
    ]);
    let count = queue.len();

    for i in 0..count {
        let user_id = queue[i].clone();

        let lock = locks
            .entry(user_id.clone())
            .or_insert_with(|| Arc::new(Mutex::new(true)))
            .clone();

        pool.execute(move || {
            // Wait until the user_id becomes free.
            let _guard = lock.lock().unwrap();

            // Do stuff with user_id, but never process
            // the same user_id more than once at the same time.
            println!("{:?}", user_id);
        });
    }

    pool.join();
}