并发题 1：有界阻塞队列

题目

实现一个容量为 N 的队列：

• put(x)：队列满时阻塞，直到有空位。
• take()：队列空时阻塞，直到有元素。
• size()：返回当前元素数。

这就是 LeetCode 1188，也是生产者-消费者模型的核心组件。线程池的任务队列、异步写日志、流水线式数据处理都建立在它之上。

先问清楚

• 容量固定吗？元素允许为 null 吗？
• 需要带超时的 offer(x, timeout) / poll(timeout) 吗？
• 需要关闭语义吗（关闭后 take 不再阻塞）？
• 对公平性（先等的线程先被服务）有要求吗？

先写最基础的阻塞版本，再按追问扩展。

方案一：一把锁 + 两个条件变量

存储用环形数组：head 指向下一个要取的位置，tail 指向下一个要放的位置，count 记录元素数。比链表少了每个元素的对象分配，内存也是固定的。

capacity = 5, count = 3
index:  0    1    2    3    4
       [ ]  [a]  [b]  [c]  [ ]
             ^head          ^tail

class BoundedBlockingQueue<T> {
    private final Object[] items;
    private int head, tail, count;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    BoundedBlockingQueue(int capacity) { items = new Object[capacity]; }

    public void put(T x) throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == items.length) notFull.await();
            items[tail] = x;
            tail = (tail + 1) % items.length;
            count++;
            notEmpty.signal();
        } finally { lock.unlock(); }
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (count == 0) notEmpty.await();
            T x = (T) items[head];
            items[head] = null;               // 让被取出的对象可以被 GC
            head = (head + 1) % items.length;
            count--;
            notFull.signal();
            return x;
        } finally { lock.unlock(); }
    }

    public int size() {
        lock.lock();
        try { return count; } finally { lock.unlock(); }
    }
}

逐条解释设计决策

为什么是两个条件变量？ 生产者等“不满”，消费者等“不空”，这是两个不同的条件。如果共用一个，put 之后的 signal 可能唤醒另一个生产者——它醒来发现还是满的，又睡回去——而真正能干活的消费者没被叫醒，系统可能卡住。共用一个条件就只能用 signalAll，代价是每次惊醒一大群线程。

为什么 signal 而不是 signalAll？ 每次 put 只多出一个元素，只够一个消费者用，唤醒一个就够了。

为什么用 while？ 虚假唤醒，以及从被唤醒到重新拿锁之间，状态可能已经被别的线程改掉。

为什么用 try/finally？ await 可能抛出 InterruptedException，无论如何锁都要释放。

中断：方法签名声明抛出 InterruptedException，把中断交给调用方处理，不要在内部吞掉。

为什么用 lockInterruptibly() 而不是 lock()？ 既然 put / take 声明了可中断，排队抢锁的阶段也应该能响应中断。lock() 在等锁时会无视中断，线程可能卡在抢锁上取消不掉。JDK 的 ArrayBlockingQueue 也是这么写的。size() 不会长时间阻塞，用普通的 lock() 即可。

方案二：两个信号量

把“空位数”和“元素数”看成两种资源：

slots = Semaphore(N)   // 空位数
items = Semaphore(0)   // 元素数
mutex = Lock           // 保护数组本身

put:  slots.acquire(); 持 mutex 入队; items.release()
take: items.acquire(); 持 mutex 出队; slots.release()

结构非常对称，也很好讲。注意先 acquire 信号量，再拿 mutex。如果顺序反过来，持有 mutex 时阻塞在信号量上，另一方就拿不到 mutex 去释放信号量，直接死锁。

方案三：两把锁

方案一里生产者和消费者竞争同一把锁。如果改用链表，put 只碰尾部，take 只碰头部，就可以用 putLock 和 takeLock 两把锁，加上一个原子计数器 count，让入队和出队真正并行。Java 的 LinkedBlockingQueue 就是这个思路，而 ArrayBlockingQueue 用的是方案一的单锁。

代价是代码复杂得多：比如 put 之后要唤醒消费者，就得去拿 takeLock 才能 signal notEmpty。面试中能讲清思路和取舍就够了，通常不要求写完。

方案	并行度	复杂度	面试中的定位
单锁 + 双条件	生产、消费互斥	低	主答案
双信号量 + mutex	同上	低	备选，展示对信号量的理解
双锁 + 原子计数	生产、消费可并行	高	口述优化

常见扩展

带超时：用 awaitNanos(remaining)，它返回剩余可等待的时间。循环里每次用返回值更新 remaining，降到 0 或以下就返回 false 或 null。不要每次都用原始超时值重新等，否则多次唤醒会让总等待时间远超预期。

关闭语义：加一个 closed 标志。close() 设置标志后对两个条件都 signalAll；关闭后 put 抛异常；take 在“已关闭且为空”时返回结束标记，让消费者退出循环。

怎么测

• 多个生产者各放入 K 个不同的数，多个消费者一直取，最后检查取出的集合和放入的集合完全一致，没有丢失也没有重复。
• 把容量设为 1，制造最激烈的竞争。
• 给测试加超时：阻塞类 bug 的表现往往是测试永远跑不完。

去做题

有界阻塞队列是 Core Skills 里的一道判题题，可以用 Python、Java、Go 或 C++ 把本章的单锁双条件写法实现一遍。驱动会同时跑多个生产者和消费者，检查队列有没有超出容量、取出的值有没有丢失或重复、同一个生产者的元素是否按顺序出队。同一分组里还有读写锁、哲学家就餐等并发题。