題目與適用情境
實作 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 更新仍需要互斥；取得多個同步原語時還要仔細處理例外、中斷和許可回滾。它可以實作正確方案，卻不會自然比「一鎖兩條件」更短。選擇任何方案都要證明計數許可與陣列真實狀態始終一致。