题干与适用场景
实现 BoundedBlockingQueue<E>。构造器接收正整数容量；put(item) 按 FIFO 入队，队列满时等待；take() 从队头出队，队列空时等待。多个生产者和消费者可以同时调用这两个方法，任何元素不能丢失、重复取出或越过更早入队的元素。等待获取锁或条件等待期间收到中断时，方法抛出 InterruptedException。
本题不允许直接包装 ArrayBlockingQueue。基础接口不包含非阻塞 offer/poll、超时、批量出队和关闭语义，也不承诺等待线程严格公平；这些适合作为追问。实现像 Java BlockingQueue 一样拒绝 null，因为 null 常被队列接口用来表示“没有元素”。
这是一道并发数据结构编码题。难点不在数组下标，而在能否给出可检查的状态契约：共享状态由谁保护、线程为什么不会丢失唤醒、醒来后为什么必须重新检查条件，以及一次操作究竟在哪一刻对其他线程生效。
面试官考察点
第一项信号是能否把“线程安全”拆成安全性与活性。安全性要求容量始终在 [0, capacity]，同一元素最多出队一次，出队顺序符合入队顺序。活性要求队列从满变为非满时生产者有机会继续，队列从空变为非空时消费者有机会继续；中断还必须能取消等待。
第二项信号是同步原语是否与状态谓词一一对应。一把锁保护数组、队头、队尾和元素数，使检查条件与修改状态处于同一个临界区。notFull 代表 size < capacity，notEmpty 代表 size > 0。生产者只等待前者，消费者只等待后者，状态跨过边界后唤醒相反角色。
第三项信号是能否解释 while，而不是把它当模板。Condition 允许虚假唤醒；即使醒来来自真实 signal，另一个竞争线程也可能先拿到锁并再次填满或取空队列。线程重新获得锁后只能重新验证谓词，不能把“被叫醒”当作“条件仍成立”。
最后看实现能否承受追问：环形下标如何保持 FIFO，更新 size 的位置为什么是线性化点，为什么一把锁没有锁顺序死锁，signal 何时足够，以及超时和关闭为什么需要重新定义接口契约。
回答前需要澄清的问题
- 容量和元素是否合法？ 容量必须大于 0，
null元素被拒绝。 - 满和空时需要忙等吗？ 不需要。线程应释放锁并进入条件等待，不能自旋占用 CPU，也不能持锁睡眠。
- 中断如何处理？
put和take都向调用者传播InterruptedException，不吞掉中断，也不在失败后修改队列。 - 是否要求严格公平？ 基础题不要求。默认的非公平
ReentrantLock允许后来线程先获得锁；公平锁会改变吞吐和调度特性。 - 需要关闭队列吗？ 基础题不需要。若需要，必须明确关闭后是否允许排空、等待者收到异常还是特殊返回值，以及谁负责唤醒全部等待线程。
- 线性化 FIFO 与等待线程 FIFO 是一回事吗？ 不是。队列元素按成功入队的线性化顺序出队，不等于多个等待生产者或消费者会按到达顺序获准执行。
- 能直接用库吗？ 生产代码通常应优先用经过验证的
ArrayBlockingQueue；本题要求手写，是为了检查并发不变量与条件等待语义。
30 秒回答框架
“我用固定数组做环形缓冲区，以 head、tail 和 size 表示队头、下一次写入位置和当前元素数。一把 ReentrantLock 保护全部共享状态，两个 Condition 分别表示非空与非满。put 在 while (size == capacity) 中等待，写入并增加 size 后唤醒一个消费者；take 对称地等待非空，清除队头并减少 size 后唤醒一个生产者。所有检查和状态转换都在同一把锁内，await 会原子释放并在返回前重获锁，所以不会出现检查与入睡之间的丢失唤醒。单次成功操作为 O(1)，空间为 O(capacity)。”
分步骤深入解答
先从朴素方案的瓶颈推导表示法。普通数组若每次出队都把剩余元素左移，单次 take 会变成 O(n)；只保留一个不断增长的读取下标又会浪费数组前部空间。固定环形数组复用已释放槽位：head 指向下一次读取位置，tail 指向下一次写入位置，size 是当前元素数。下标走到数组末端后回到 0，因此入队和出队都不需要移动已有元素。
实现始终维护四条不变量：
0 <= size <= items.length。- 从
head开始沿环形方向的前size个槽位，依次保存尚未取出的 FIFO 元素。 tail == (head + size) % items.length；当队列满时head == tail，必须由size区分“满”与“空”。items、head、tail和size的每次读取或修改都发生在同一把锁内。
核心实现如下：
import java.util.Objects;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public final class BoundedBlockingQueue<E> {
private final Object[] items;
private final ReentrantLock lock = new ReentrantLock();
private final Condition notEmpty = lock.newCondition();
private final Condition notFull = lock.newCondition();
private int head;
private int tail;
private int size;
public BoundedBlockingQueue(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("capacity must be positive");
}
items = new Object[capacity];
}
public void put(E item) throws InterruptedException {
Objects.requireNonNull(item, "item");
lock.lockInterruptibly();
try {
while (size == items.length) {
notFull.await();
}
items[tail] = item;
tail = (tail + 1) % items.length;
size++;
notEmpty.signal();
} finally {
lock.unlock();
}
}
@SuppressWarnings("unchecked")
public E take() throws InterruptedException {
lock.lockInterruptibly();
try {
while (size == 0) {
notEmpty.await();
}
E item = (E) items[head];
items[head] = null;
head = (head + 1) % items.length;
size--;
notFull.signal();
return item;
} finally {
lock.unlock();
}
}
}Objects.requireNonNull 放在加锁前，因为它只检查调用参数，不依赖共享状态。lockInterruptibly() 让等待获取锁本身也能被中断；进入 try 后无论正常返回、条件等待被中断还是发生运行时异常，finally 都会释放当前线程持有的锁。
await() 的关键语义是原子地释放关联锁并等待，返回前重新取得这把锁。生产者不能先手动解锁、再把自己登记为等待者；那会留下一个窗口：消费者可能在窗口内释放空间并发送信号，而生产者尚未开始等待，信号就永久丢失。条件变量把“释放锁并进入等待”合成一个同步动作，消除了这个窗口。
while 同时处理两类情况。其一是规范允许的虚假唤醒；其二是正常竞争：两个消费者都可能被唤醒，其中一个先取得锁并取走唯一元素，另一个取得锁时队列又为空。条件通知只表示“状态可能变化”，状态谓词才是继续执行的依据。
成功 put 的线性化点位于锁内元素加入队列、size 从 k 变为 k + 1 的状态转换；成功 take 对应元素被移除、size 从 k 变为 k - 1。锁让其他线程只能看到转换前或转换后的完整状态，不能看到数组已经写入而 size 尚未更新的中间态。unlock 与后续获取同一把锁还建立内存可见性；这与标准并发队列对跨线程元素传递的 happens-before 目标一致。
为什么使用两个条件？若只有一个等待集合，take 之后可能唤醒另一个消费者，而队列仍为空；真正需要空间的生产者继续沉睡。分开 notEmpty 和 notFull 后，每次状态转换只通知可能因此变得可运行的角色。基础的单元素 put/take 每次只创造一个新槽位或一个新元素，因此 signal() 足够；醒来的线程仍必须用 while 复查。若一次操作改变多个槽位，或关闭操作使所有等待者都必须观察新状态，就需要重新评估 signalAll()。
正确性可以用不变量归纳。初始化时 head = tail = size = 0。入队只能在 size < capacity 时执行，它写入 tail、推进尾指针并增加一次大小，所以不越界且追加在所有未取元素之后。出队只能在 size > 0 时执行，它读取 head、清空槽位、推进头指针并减少一次大小，所以返回最早未取元素。锁把每次转换串行化，多个生产者和消费者的任意调度都等价于某个合法的顺序执行。
每次成功 put 或 take 只做固定次数的数组、整数与同步操作，排除等待时间后为 O(1)；固定数组占用 O(capacity) 空间。竞争越高，锁等待和上下文切换可能越多，复杂度符号不能表达这部分延迟。单锁设计也意味着只有一个锁，不存在多锁循环等待，但调用方若持有其他锁再调用阻塞方法，仍可能在更大的系统里制造锁顺序问题。
测试不能只跑单线程示例。至少应覆盖容量 1 的满/空交替、环形下标多次回绕、重复的相等元素被分别保存和取出、空字符串可入队而 null 被拒绝、多个生产者生成不重复编号并由多个消费者取尽、最终集合无丢失无重复、每个单生产者流内部顺序不逆转，以及分别中断等待入队和等待出队的线程。极大容量会在构造时直接分配同等长度数组，可能因内存不足失败；基础实现不做懒分配，也不能把分配失败伪装成空队列。并发测试应使用屏障或锁存器协调开始，并设置测试框架级截止时间来发现永久阻塞；不要靠短暂 sleep 猜测线程已经进入等待。
高质量示范回答
“我先把基础契约限定为正容量、非空元素、阻塞 put/take、FIFO、多生产者多消费者和可中断等待。超时、关闭与严格公平需要额外返回值和状态语义，先不混进核心实现。
表示上用固定环形数组，head 指向下次读取，tail 指向下次写入，size 同时消除 head == tail 时满与空的歧义。四个共享字段全部由同一把 ReentrantLock 保护。notEmpty 的谓词是 size > 0，notFull 的谓词是 size < capacity。
put 可中断地获取锁，在 while 中等待非满，写入尾部并增加 size，再通知一个消费者；take 对称地等待非空，取出头部、清掉引用并减少 size，再通知一个生产者。使用 while 是因为条件等待允许虚假唤醒，而且被唤醒线程重新竞争锁时，谓词可能已被别的线程改回 false。
await 原子完成释放锁与进入等待，避免检查条件后、真正睡眠前漏掉通知。成功操作的线性化点是锁内改变队列成员与 size 的状态转换；同一把锁使其他线程只能观察完整的前态或后态。由容量、环形 FIFO 和同锁保护三条核心不变量归纳，可以证明不会越界、重复取出或乱序。
排除阻塞时间后，两个方法都是 O(1)，空间是 O(capacity)。我会用锁存器同时启动多生产者和多消费者，核对唯一编号集合与每个生产者的内部顺序，再单独验证满队列、空队列和中断路径。”
常见错误
- 用
if检查满或空 → 虚假唤醒或重新竞争后条件可能仍不成立 → 在while中反复检查状态谓词。 - 检查条件后手动解锁再等待 → 状态变化与登记等待之间出现窗口，通知可能丢失 → 使用关联同一把锁的
Condition.await()。 - 数组、下标和
size分别同步 → 其他线程能看到互相矛盾的中间态 → 一把锁覆盖完整的检查与状态转换。 - 一个条件集合配合随意
signal→ 可能唤醒无法推进的同类线程 → 分别维护notEmpty与notFull。 - 在持锁时忙等或
sleep→ 能改变条件的线程拿不到锁 → 条件等待必须释放锁。 - 吞掉
InterruptedException→ 上层无法取消任务，线程可能永久滞留 → 声明并传播中断，且用finally解锁。 - 只用
head == tail判断空满 → 环形数组的两种状态不可区分 → 同时维护受锁保护的size。 - 出队后不清空数组槽位 → 已消费对象仍被数组引用，延长其生命周期 → 读取后将槽位置为
null。 - 把元素 FIFO 等同于线程公平 → 默认锁不保证等待调用者按到达顺序完成 → 分别描述元素顺序与调度策略。
- 为基础实现声称支持关闭 → 等待线程没有可观察的关闭状态，可能永远不醒 → 先定义关闭契约，再增加状态与全量通知。
追问及应对
追问一：怎样增加带超时的 offer 和 poll？
把剩余时长转换为纳秒，并在同一个 while 谓词循环中调用 awaitNanos(remaining)。每次返回都用它给出的剩余时间继续等待；若剩余值不大于 0 且谓词仍为 false，则返回失败。不能在每次虚假唤醒后重新使用完整超时，否则实际等待可能无限延长。接口还要区分“超时未完成”与“元素值为空”，这也是拒绝 null 的原因之一。
追问二：怎样实现 shutdown()？
先定义状态机。例如关闭后拒绝新 put，已入队元素允许继续排空；关闭且为空时 take 抛出专用异常或返回明确结果。shutdown 必须在同一把锁内改变关闭标记，并对两个条件执行 signalAll()，让所有等待者重新获得锁后观察关闭状态。每个等待循环的谓词都要加入关闭分支，不能只增加一个布尔字段。
追问三：为什么这里用 signal()，什么时候改用 signalAll()？
一次基础入队只新增一个可消费元素，一次出队只新增一个空槽，所以唤醒一个相反角色即可推进，并减少无效竞争。批量插入、批量取出、容量动态变化或关闭会一次改变多个线程的可执行性，通常需要 signalAll() 或与变化数量匹配的通知策略。无论唤醒多少线程，while 复查都不能省略。
追问四：怎样提供公平性？
可以用 new ReentrantLock(true) 让锁倾向等待最久的线程，但这仍不等于对 put/take 完成顺序给出绝对实时保证，中断和调度也会影响结果。公平策略通常减少插队和饥饿风险，却可能降低吞吐。只有业务明确需要等待者顺序时，才值得承担并验证该成本。
追问五：能否做成无锁队列？
无锁有界 MPMC 队列通常需要原子序号、CAS 与更复杂的内存序证明；“阻塞”部分仍需停车和唤醒机制，不能只靠 CAS 自旋。它会引入 ABA、伪共享、进度保证和平台内存模型等问题。除非性能测量表明单锁是瓶颈，并且团队能维护严格证明与压力测试，标准库或清晰的锁实现更稳妥。
追问六：为什么不直接使用两个信号量？
一个计数信号量可以表示空槽数，另一个表示元素数，但环形数组的 head/tail 更新仍需要互斥；获取多个同步原语时还要仔细处理异常、中断和许可回滚。它可以实现正确方案，却不天然比“一锁两条件”更短。选择任何方案都要证明计数许可与数组真实状态始终一致。