Rust多线程应用中如何使用VecDeque实现共享读写队列

我正在尝试基于VecDeque开发多线程应用，希望将其作为所有线程均具备读写权限的共享队列使用，对应的实现代码如下：

use std::collections::VecDeque;
use std::{thread, time};

fn main() {
    let mut workload = VecDeque::new();
    workload.push_back(0);

    let mut thread_1_queue = workload.clone();
    let thread_1 = thread::spawn(move || {
        let mut counter1: i32 = 0;
        let some_time = time::Duration::from_millis(50);

        loop {
            counter1 +=1;
            thread_1_queue.push_back(counter1);

            println!("Thread #1: {:?}", thread_1_queue);

            if counter1 == 10 {
                break;
            }

            thread::sleep(some_time);
        };
    });

    let mut thread_2_queue = workload.clone();
    let thread_2 = thread::spawn(move || {
        let mut counter2: i32 = 10;
        let some_time = time::Duration::from_millis(50);

        loop {
            counter2 +=1;
            thread_2_queue.push_back(counter2);

            println!("Thread #2: {:?}", thread_2_queue);

            if counter2 == 20 {
                break;
            }

            thread::sleep(some_time);
        };
    });

    let some_time = time::Duration::from_millis(50);

    loop {
        if workload.capacity() == 10 {
            break;
        }

        println!("MainQueue: {:?}", workload);

        thread::sleep(some_time);
    }

    thread_1.join();
    thread_2.join();
}

注意：该代码会无限循环运行

当前遇到的问题为：线程中克隆得到的VecDeque副本无法同步更新主队列，每个线程实际持有独立的队列实例，并未实现预期的单队列共享效果，运行输出如下所示：

Thread #1: [0, 1]
MainQueue: [0]
Thread #2: [0, 11]
Thread #1: [0, 1, 2]
Thread #2: [0, 11, 12]
MainQueue: [0]
MainQueue: [0]
Thread #2: [0, 11, 12, 13]
Thread #1: [0, 1, 2, 3]
MainQueue: [0]
Thread #2: [0, 11, 12, 13, 14]
Thread #1: [0, 1, 2, 3, 4]
MainQueue: [0]
Thread #2: [0, 11, 12, 13, 14, 15]
Thread #1: [0, 1, 2, 3, 4, 5]
MainQueue: [0]
Thread #2: [0, 11, 12, 13, 14, 15, 16]
Thread #1: [0, 1, 2, 3, 4, 5, 6]
MainQueue: [0]
Thread #2: [0, 11, 12, 13, 14, 15, 16, 17]
Thread #1: [0, 1, 2, 3, 4, 5, 6, 7]
MainQueue: [0]
Thread #2: [0, 11, 12, 13, 14, 15, 16, 17, 18]
Thread #1: [0, 1, 2, 3, 4, 5, 6, 7, 8]
MainQueue: [0]
Thread #1: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
Thread #2: [0, 11, 12, 13, 14, 15, 16, 17, 18, 19]
Thread #2: [0, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
Thread #1: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
MainQueue: [0]

请问该如何修改代码，实现多线程环境下VecDeque的跨线程共享读写？

Rust所有权规则默认不允许多个线程同时持有同一个可变对象的所有权，直接调用clone()得到的是完全独立的内存副本，修改不会同步到原始实例。要实现跨线程共享读写，需要组合使用两个标准库组件：

• Arc：原子引用计数指针，让多个线程可以安全持有同一个堆对象的共享所有权，克隆Arc只会增加引用计数，不会复制底层的队列数据
• Mutex：互斥锁，保证同一时间只有一个线程能操作队列，避免多线程同时写入导致的数据竞争

修改后的可运行代码如下：

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::{thread, time};

fn main() {
    // 用Arc<Mutex>包裹队列，同时实现跨线程共享所有权和线程安全可变访问
    let workload = Arc::new(Mutex::new(VecDeque::new()));
    workload.lock().unwrap().push_back(0);

    // 克隆Arc仅增加引用计数，所有线程指向同一份底层队列
    let queue_1 = Arc::clone(&workload);
    let thread_1 = thread::spawn(move || {
        let mut counter1: i32 = 0;
        let some_time = time::Duration::from_millis(50);

        loop {
            counter1 += 1;
            // 加锁后才能访问、修改内部队列
            let mut q = queue_1.lock().unwrap();
            q.push_back(counter1);
            println!("Thread #1: {:?}", q);
            // 锁守卫q离开作用域时会自动释放锁，无需手动解锁

            if counter1 == 10 {
                break;
            }

            thread::sleep(some_time);
        }
    });

    let queue_2 = Arc::clone(&workload);
    let thread_2 = thread::spawn(move || {
        let mut counter2: i32 = 10;
        let some_time = time::Duration::from_millis(50);

        loop {
            counter2 += 1;
            let mut q = queue_2.lock().unwrap();
            q.push_back(counter2);
            println!("Thread #2: {:?}", q);

            if counter2 == 20 {
                break;
            }

            thread::sleep(some_time);
        }
    });

    let some_time = time::Duration::from_millis(50);
    loop {
        let q = workload.lock().unwrap();
        // VecDeque扩容是翻倍增长，不一定刚好等于10，用>=判断避免无限循环
        if q.capacity() >= 10 {
            break;
        }
        println!("MainQueue: {:?}", q);
        // 主动释放锁，避免持锁睡眠阻塞其他线程
        drop(q);
        thread::sleep(some_time);
    }

    thread_1.join().unwrap();
    thread_2.join().unwrap();
}

关键修改点 #

• 移除原代码对VecDeque的直接克隆，改用Arc<Mutex<VecDeque<_>>>包裹队列实例，所有线程拿到的都是指向同一份队列的共享指针
• 所有队列读写操作前必须调用.lock().unwrap()获取互斥锁，拿到锁守卫后才能操作内部数据，锁会在守卫离开作用域时自动释放，降低死锁风险
• 主线程打印完队列后主动调用drop(q)释放锁，避免持锁睡眠导致其他线程长时间阻塞
• 调整队列容量判断逻辑为capacity() >= 10，适配VecDeque翻倍扩容的特性，避免再次触发无限循环

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