LeetCode 1188. 设计有限阻塞队列

2022-05-08 HGNULB 约 13 分钟

文章信息 6 个分类 1 个标签

分类

题目描述

✅ 1188. 设计有限阻塞队列

题意分析

实现一个容量固定为 capacity 的线程安全队列，支持三个方法：enqueue(x) 把元素放到队尾——如果队列已满，调用线程必须阻塞等待，直到有空位；dequeue() 取出并返回队首元素——如果队列为空，调用线程必须阻塞等待，直到有元素；size() 返回当前元素个数。题目保证生产者线程只调 enqueue、消费者线程只调 dequeue，但两侧都可能有多个线程。

「阻塞」这个词是全题的核心，它排除了两类看似可行的写法。第一类是返回失败让调用方重试——题目要求的是等待而不是失败。第二类是忙等（while (full) {} 空转）——它虽然「能跑」，但会把 CPU 烧满，在面试里等同于错误答案。必须让线程真正挂起、并在条件满足时被唤醒。

其次要看清这是一个双向阻塞：满时挡住生产者、空时挡住消费者。两个方向的等待条件不同、唤醒时机也不同，这决定了用一把锁配两个条件变量而不是一个。

数据本身没有难度：先进先出、容量固定，用一个定长数组加两个下标环形使用即可，所有操作 $O(1)$。真正被考察的是并发原语的正确使用。

边界：capacity 至少为 1；队列可能长期处于满或空的极端状态，多个线程同时被阻塞；size() 也会被并发调用，同样需要同步保护。

解法：锁 + 条件变量 + 环形数组

核心思路

先看最朴素的尝试：给三个方法都加 synchronized，满时写 while (size == cap) { } 自旋。自旋线程始终持有队列锁，消费者在任何 CPU 核上都无法进入临界区，队列永远腾不出空位；同时自旋还会持续占用 CPU。瓶颈很明确——等待期间必须同时释放锁和 CPU。

正确的原语是「锁 + 条件变量」。条件变量的 await() 做了三件不可分割的事：释放锁、挂起当前线程、被唤醒后重新抢锁。释放锁这一点是关键，它让别的线程有机会进来改变状态，从而使等待条件最终得以满足。

于是数据结构定为：一个定长数组 data，下标 head 指向队首元素、tail 指向下一个写入位置，计数 size；一把 ReentrantLock，以及从它派生的两个条件变量 notFull（生产者在此等待）与 notEmpty（消费者在此等待）。

不变量是：任何线程持锁期间观察到的 size 恒等于队列中的实际元素个数，且这 size 个元素依次位于 data[head]、data[(head+1) % cap]、……；0 <= size <= cap。 三个方法都必须在持锁状态下读写这套状态，退出前恢复不变量。

为什么要单独维护 size 而不用 head 与 tail 推算？因为环形数组里 head == tail 既可能表示空、也可能表示满，二者无法区分。常见的替代方案是「牺牲一个格子」，但那会让实际容量变成 capacity - 1，与题目要求不符。显式的 size 既解决歧义又让判断条件直白。

两个条件变量的分工要清楚：

enqueue 在 size == cap 时 notFull.await()；成功入队后 notEmpty.signal() 唤醒一个可能在等的消费者。

dequeue 在 size == 0 时 notEmpty.await()；成功出队后 notFull.signal() 唤醒一个可能在等的生产者。

等待自己的条件、唤醒对方的条件——这是生产者消费者模型的标准配对。分开等待队列后，notEmpty.signal() 只会选择消费者，notFull.signal() 只会选择生产者。若所有线程共用一个条件变量，signal() 在等待者混合时可能选中条件仍不成立的一侧，真正能推进状态的线程继续沉睡；共用条件时必须改用 signalAll()，代价是惊群。两个条件变量让单次唤醒既正确又精确。

最后一个必须讲清的点：等待条件必须循环复检，不能只用 if 判断一次。Java 的 Condition.await() 允许虚假唤醒；更普遍的原因是，signal 只让等待者重新参与锁竞争，在它拿到锁之前，另一个同侧线程可能已经取走那个空位或元素。Go 的 sync.Cond.Wait() 不会无故返回，但同样存在后一种竞争。因此 Java 用 while、Go 用 for，醒来后都要重新检查谓词。

正确性：同一把锁使每次入队、出队的状态变更不可交错；等待循环保证入队只在 size < cap 时发生、出队只在 size > 0 时发生，因此容量边界始终成立。入队只写 tail 并向后推进，出队只读 head 并向后推进，所以元素按进入顺序离开。每次改变“空/满”谓词后再唤醒对侧，使已满足条件的等待者有机会继续；所有方法退出时都恢复环形队列不变量。

解题步骤

构造函数初始化状态：分配长度为 capacity 的数组，创建 ReentrantLock，并用 lock.newCondition() 派生两个条件变量。两个条件变量必须来自同一把锁——它们共享该锁的等待机制，跨锁使用会抛异常。head、tail、size 默认为 0 即为空队列。

enqueue 加锁：lock.lock() 之后立刻 try，finally 里 unlock()。把解锁放在 finally 是硬性要求——中间任何一步抛出异常（包括 await 抛 InterruptedException）都不能让锁泄漏，否则整个队列永久卡死。

循环等待非满：while (size == data.length) notFull.await();。用 while 应对虚假唤醒与「醒来后名额被抢」两种情况。

写入并推进：data[tail] = element; tail = (tail + 1) % data.length; size++;。取模实现环形复用，让数组尾部与头部首尾相接，避免搬移数据。三行必须在锁内一起完成，否则别的线程可能看到「写了数据但 size 还没加」的中间态。

唤醒对侧：状态更新完成后调用 notEmpty.signal()。被唤醒者仍要等当前线程释放锁后才能继续，因此同一临界区内先更新还是先通知不会暴露中间态；写成「改变谓词后通知」能让状态迁移与信号的因果关系最清楚。

dequeue 对称处理：循环等待 size == 0，取出 data[head]，head 前移取模，size--，然后 notFull.signal()，最后返回。注意 return val 写在 try 块内、finally 之前——Java 会保证 finally 中的 unlock 在返回前执行。

size() 也要加锁：读取 size 同样需要同步。不加锁读一个非 volatile 的 int 存在可见性问题（可能长期读到过期值），且与 enqueue/dequeue 的更新构成数据竞争。

以 capacity = 2、两个生产者线程 P1、P2 和一个消费者线程 C 走一遍：

初始 head = tail = size = 0，数组 [_, _]。

P1 调 enqueue(1)：拿到锁，size = 0 != 2 不等待，写 data[0] = 1，tail = 1，size = 1，notEmpty.signal()（此刻没人在等，信号被丢弃，这是允许的），释放锁。

P2 调 enqueue(2)：拿锁，size = 1 != 2，写 data[1] = 2，tail = (1+1) % 2 = 0 绕回数组头部，size = 2，signal，释放锁。此时 head = 0、tail = 0、size = 2——head == tail 却是满的，这正是必须显式维护 size 的原因。

P1 再调 enqueue(3)：拿锁，size == 2 满足等待条件，进入 notFull.await()——锁被释放，P1 挂起。若这里用忙等，锁不会释放，下面的 C 永远进不来。

C 调 dequeue()：拿到（被 P1 释放的）锁，size = 2 != 0，取 val = data[0] = 1，head = 1，size = 1，notFull.signal() 唤醒 P1，释放锁并返回 1。

P1 被唤醒后重新抢锁，回到 while 顶部重新检查 size == 2？现在 size = 1，条件不成立，退出循环，写 data[0] = 3（tail = 0），tail = 1，size = 2，释放锁。

假设此时还有 P3 也在 notFull 上等待，而 C 只出队了一次：signal 唤醒的可能是 P3 而不是 P1，P3 抢到锁后填满队列；P1 后来醒来发现 size 又变回 2，于是再次进入等待——这正是 while 而非 if 的价值所在。若写成 if，P1 会直接覆盖 data[tail]，把 P3 刚写入的元素冲掉，并让 size 变成 3 超过容量，不变量彻底崩坏。

代码实现

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

class BoundedBlockingQueue {
    private final int[] data;
    private int head;
    private int tail;
    private int size;
    private final ReentrantLock lock;
    private final Condition notFull;
    private final Condition notEmpty;

    public BoundedBlockingQueue(int capacity) {
        this.data = new int[capacity];
        this.lock = new ReentrantLock();
        // 两个条件变量必须派生自同一把锁。
        this.notFull = lock.newCondition();
        this.notEmpty = lock.newCondition();
    }

    public void enqueue(int element) throws InterruptedException {
        lock.lock();
        try {
            // 必须用 while：应对虚假唤醒，以及醒来后空位被同侧线程抢走。
            while (size == data.length) {
                notFull.await();
            }

            data[tail] = element;
            tail = (tail + 1) % data.length;
            size++;

            // 状态更新完成后再唤醒对侧。
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public int dequeue() throws InterruptedException {
        lock.lock();
        try {
            while (size == 0) {
                notEmpty.await();
            }

            int val = data[head];
            head = (head + 1) % data.length;
            size--;

            notFull.signal();
            return val;
        } finally {
            lock.unlock();
        }
    }

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

import "sync"

type BoundedBlockingQueue struct {
	cap      int
	data     []int
	head     int
	tail     int
	size     int
	mu       sync.Mutex
	notFull  *sync.Cond
	notEmpty *sync.Cond
}

// 必须返回指针：Cond 持有 &q.mu，返回值拷贝会让锁与条件变量指向不同的对象。
func Constructor(capacity int) *BoundedBlockingQueue {
	q := &BoundedBlockingQueue{
		cap:  capacity,
		data: make([]int, capacity),
	}
	q.notFull = sync.NewCond(&q.mu)
	q.notEmpty = sync.NewCond(&q.mu)
	return q
}

func (q *BoundedBlockingQueue) Enqueue(element int) {
	q.mu.Lock()
	defer q.mu.Unlock()
	// 必须用 for：醒来后重新抢锁时，空位可能已被同侧 goroutine 抢走。
	for q.size == q.cap {
		q.notFull.Wait()
	}

	q.data[q.tail] = element
	q.tail = (q.tail + 1) % q.cap
	q.size++

	// 状态更新完成后再唤醒对侧。
	q.notEmpty.Signal()
}

func (q *BoundedBlockingQueue) Dequeue() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	for q.size == 0 {
		q.notEmpty.Wait()
	}

	val := q.data[q.head]
	q.head = (q.head + 1) % q.cap
	q.size--

	q.notFull.Signal()
	return val
}

func (q *BoundedBlockingQueue) Size() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return q.size
}

复杂度分析

时间复杂度：三个方法在不阻塞时均为 $O(1)$——只做常数次下标运算与赋值，环形数组避免了任何搬移。阻塞时长取决于对侧线程的调度，不计入算法复杂度。

空间复杂度：$O(capacity)$，定长数组一次性分配，不随入队出队增长；此外只有一把锁、两个条件变量和三个整型下标，都是常数开销。

关键点总结

阻塞必须靠条件变量的 await/wait 实现，它在挂起线程的同时释放锁；忙等既烧 CPU 又因不放锁而必然死锁，是这类题的第一条红线。

一把锁 + 两个条件变量：等待自己的条件、唤醒对方的条件。共用一个条件变量会因唤醒到同侧线程而丢失信号，除非退而使用 signalAll 承受无谓唤醒。

等待必须循环检查谓词：Java 用 while (...) await() 防虚假唤醒和状态竞争；Go 用 for ... { Wait() } 防重新抢锁前的状态竞争。一次 if 判断不足以保证醒来时条件仍成立。

环形数组中 head == tail 无法区分空与满，因此要么显式维护 size，要么牺牲一个格子；本题容量固定，显式 size 是更贴题的选择。

解锁必须放在 finally（Go 用 defer），保证任何异常路径都不会泄漏锁。

只读方法同样要加锁：size() 不加锁可能读到过期值，并与写操作构成数据竞争；原子地读到一个 int 不等于具备跨线程可见性。

Go 中 sync.Cond 保存的是 Locker 的指针，构造函数必须返回指针；返回结构体值会让拷贝出来的 mu 与 Cond.L 指向不同的互斥量，导致 Wait 解开的不是当前持有的那把锁。

易错点总结

错误写法：等待用 if (size == cap) notFull.await();。用例容量 1 的满队列中 P1 等待，消费者出队并唤醒 P1；P1 重新拿锁前，新生产者 P2 抢先入队填满。P1 随后若不复检就继续写入，会让 size 变成 2 并覆盖未消费元素。

错误写法：只用一个条件变量并调用 signal()。可复现调度：容量 1 且队列已满，P1、P2 入队等待；C1 出队并唤醒 P1；P1 抢锁前 C2 发现队列为空并等待；随后 P1 入队，却错误唤醒 P2。P2 复检后继续等待，C2 也仍在睡，队列虽有元素却再无活动线程，形成死锁。

错误写法：忙等 while (size == cap) { } 不释放锁。用例队列满时生产者调用 enqueue：锁被一直持有，消费者永远进不了 dequeue，队列永远不会腾出空位，死锁且 CPU 占满。

错误写法：unlock() 写在 try 块末尾而非 finally。用例 await() 抛出 InterruptedException：解锁语句被跳过，锁永久泄漏，之后所有线程在 lock() 上无限阻塞。

错误写法：size() 不加锁直接返回字段。用例高并发下持续调用：可能长期读到缓存中的过期值，且与 size++ 构成数据竞争，Go 的 -race 检测器会直接报错。

错误写法：用 head == tail 判空、(tail + 1) % cap == head 判满而不维护 size。用例 capacity = 2 连续入队两个元素：判满条件提前成立，实际只能存 1 个元素，容量比题目要求少 1。

错误写法：tail 或 head 前移时忘记取模。用例 capacity = 2 入队三次（中间有出队）：tail 递增到 2 后数组下标越界抛异常。

错误写法：状态改变后忘记唤醒对侧。用例容量 1 的空队列中消费者已在 notEmpty 等待，生产者入队后若漏掉 notEmpty.signal()，元素已经存在但消费者仍可能永久沉睡。

错误写法：两个条件变量分别用 new ReentrantLock().newCondition() 从不同的锁派生。用例任意一次 await：会抛 IllegalMonitorStateException，因为当前线程并未持有该条件变量所属的锁。

错误写法：Go 中 func Constructor(capacity int) BoundedBlockingQueue 返回结构体值。用例任意一次 Wait：Cond.L 指向构造函数内那个已被拷贝走的 mu，Wait 解开的锁与调用者持有的锁不是同一把，立刻产生数据竞争并可能永久阻塞。

错误写法：dequeue 中先解锁再取值，例如把 int val = data[head] 挪到 unlock() 之后。用例两个消费者并发出队：head 已被推进，两人可能读到同一个位置或读到已被覆盖的数据，返回值与实际出队的元素对不上。

相似题目

题目	难度	考察点
1114. 按序打印	简单	多线程同步的入门题，只需一个计数状态与条件等待，没有容量限制
1115. 交替打印 FooBar	中等	两线程严格交替，等待条件是「轮到我了」，可对照双条件变量的写法
1116. 打印零与奇偶数	中等	三方轮转，条件变量数量与状态划分的设计更复杂
1195. 多线程 Fizz Buzz	中等	四线程按数值特征分工，练习「唤醒谁」的判定逻辑
1226. 哲学家进餐	中等	多把锁的获取顺序与死锁避免，是并发题里另一条主线
622. 设计循环队列	中等	本题去掉并发后的单线程内核，专门练环形数组的空满判定
美团面试题-Java 生产者消费者模式	中等	同一模型的工程视角，涵盖 `wait`/`notifyAll`、信号量与现成容器的选型