用 Mutex 与 Condvar 实现有界队列

功能说明

有界阻塞队列是生产者消费者模式的基础结构：队列满时生产者等待，队列空时消费者等待，关闭后双方都能被唤醒并退出。这段代码只用 VecDeque、Mutex 和两个 Condvar 实现，不依赖异步运行时或第三方库。

它适合线程池任务队列、日志缓冲和后台处理管线。容量上限提供天然背压，防止生产速度长期高于消费速度时内存无限增长。

关键代码

QueueState 同时保存数据队列和 closed 标志，二者必须在同一把锁下修改，否则会出现“已关闭但仍等待”的竞态。BoundedQueue 持有 ready 和 space 两个条件变量，分别唤醒等待取数据的消费者和等待空位的生产者。

push 在队列满且未关闭时循环等待 space；被唤醒后重新检查条件，避免虚假唤醒。pop 在队列空时等待 ready，取出元素后通知生产者。close 设置标志并 notify_all，让所有等待者重新检查状态。

使用注意

Condvar 必须配合 while 循环使用，不能假设一次唤醒就代表条件成立。锁中毒（线程 panic 时持锁）会导致 expect 失败，生产代码可以转换为自定义错误。close 之后 push 会静默丢弃元素，调用方需要知道这一语义。

这个队列不保证公平性，也不支持多消费者顺序以外的优先级。队列容量是元素个数，不是字节数，大对象仍可能占用大量内存。生产环境应增加超时、指标和优雅关闭流程。

验证方式

示例容量为 2，生产者写入 0 到 5 后关闭，消费者逐个取出并故意休眠 1 毫秒，最终得到 [0, 1, 2, 3, 4, 5]，说明背压和关闭唤醒都正常。可以把容量改成 1 或把消费者休眠去掉，结果仍应一致。运行 cargo test 时可加入超时测试，防止死锁。

代码示例

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

struct QueueState<T> {
    items: VecDeque<T>,
    closed: bool,
}

struct BoundedQueue<T> {
    capacity: usize,
    state: Mutex<QueueState<T>>,
    ready: Condvar,
    space: Condvar,
}

impl<T> BoundedQueue<T> {
    fn new(capacity: usize) -> Arc<Self> {
        assert!(capacity > 0, "capacity must be positive");
        Arc::new(Self {
            capacity,
            state: Mutex::new(QueueState {
                items: VecDeque::with_capacity(capacity),
                closed: false,
            }),
            ready: Condvar::new(),
            space: Condvar::new(),
        })
    }

    fn push(&self, item: T) {
        let mut state = self.state.lock().expect("queue poisoned");
        while state.items.len() == self.capacity && !state.closed {
            state = self.space.wait(state).expect("queue poisoned");
        }
        if state.closed {
            return;
        }
        state.items.push_back(item);
        self.ready.notify_one();
    }

    fn pop(&self) -> Option<T> {
        let mut state = self.state.lock().expect("queue poisoned");
        loop {
            if let Some(item) = state.items.pop_front() {
                self.space.notify_one();
                return Some(item);
            }
            if state.closed {
                return None;
            }
            state = self.ready.wait(state).expect("queue poisoned");
        }
    }

    fn close(&self) {
        let mut state = self.state.lock().expect("queue poisoned");
        state.closed = true;
        self.ready.notify_all();
        self.space.notify_all();
    }
}

fn main() {
    let queue = BoundedQueue::new(2);
    let producer = {
        let queue = Arc::clone(&queue);
        thread::spawn(move || {
            for value in 0..6 {
                queue.push(value);
            }
            queue.close();
        })
    };
    let consumer = {
        let queue = Arc::clone(&queue);
        thread::spawn(move || {
            let mut seen = Vec::new();
            while let Some(value) = queue.pop() {
                seen.push(value);
                thread::sleep(Duration::from_millis(1));
            }
            seen
        })
    };

    producer.join().expect("producer panicked");
    let seen = consumer.join().expect("consumer panicked");
    println!("{seen:?}");
    assert_eq!(seen, vec![0, 1, 2, 3, 4, 5]);
}