問題と適用場面
BoundedBlockingQueue<E> を実装します。そのコンストラクタは正の容量を受け取ります。put(item) は FIFO 順で要素を追加し、キューが満杯の間は待機します。take() は先頭から要素を取り出し、キューが空の間は待機します。複数のプロデューサーおよびコンシューマーが両方のメソッドを並行して呼び出す可能性があります。要素が失われたり、2回返されたり、先に入った要素より後に返されたりしてはなりません。スレッドがロックの取得中または条件の待機中に割り込まれた場合、メソッドは InterruptedException をスローします。
この課題では ArrayBlockingQueue をラップすることは許可されていません。ベース API からは、非ブロッキングの offer/poll、タイムアウト、一括取り出し、シャットダウンのセマンティクスは除外されており、待機スレッド間の厳密な公平性（fairness）も保証されません。これらは発展的な問いです。Java の BlockingQueue と同様に、この実装でも null は拒否されます。キューの API では、利用可能な要素がないことを表すために null がよく使用されるためです。
これは並行データ構造のコーディング問題です。配列のインデックス操作は簡単な部分にすぎません。真のテストは、候補者が検証可能な契約を明確に説明できるかどうかです。すなわち、どの同期機構が共有状態を保護しているのか、なぜ通知が失われないのか、なぜ起床したスレッドが条件を再確認しなければならないのか、そしてどの瞬間に他のスレッドに対して操作が有効になるのか（効果を持つのか）です。
面接官が評価しているポイント
第1のシグナルは、候補者が安全性（safety）と生存性（liveness）を明確に区別できているかです。安全性には、サイズが [0, capacity] 内にとどまること、各要素が最大1回しか取り出されないこと、および取り出し順序が挿入順序と一致することが求められます。生存性には、満杯のキューが満杯でなくなったときにプロデューサーが進む機会を得られること、および空のキューが空でなくなったときにコンシューマーが進む機会を得られることが求められます。また、割り込みによって待機をキャンセルできる必要があります。
第2のシグナルは、同期プリミティブが状態述語と正しく対応しているかです。1つのロックが配列、head、tail、および size を保護するため、条件の確認と状態の変更は同じクリティカルセクション内で行われます。notFull は size < capacity を表し、notEmpty は size > 0 を表します。プロデューサーは前者のみを待機し、コンシューマーは後者のみを待機し、境界を超える状態変化が発生すると反対の役割にシグナルを送ります。
第3のシグナルは、候補者が while を暗記パターンとしてではなく、その理由を説明できるかです。Condition はスプリアスウェイクアップ（擬似起床）を許容します。実際の signal の後であっても、競合する別のスレッドが先にロックを再取得してキューを再び満杯または空にする可能性があります。起床したスレッドはロックを再取得した後、述語を再テストしなければなりません。通知を受けたこと自体は、条件が依然として成立していることを証明しません。
最後に、面接官は設計についてさらに掘り下げることができます。リングインデックスがどのように FIFO を維持するのか、なぜサイズ更新が線形化ポイントに属するのか、なぜ1つのロックがロック順序によるデッドロックを回避できるのか、いつ signal で十分なのか、そしてなぜタイムアウトやシャットダウンには新しい API セマンティクスが必要なのかなどです。
回答前に確認すべき質問
- 有効な容量と要素は何か？ 容量は0より大きい必要があり、
null要素は拒否されます。 - 満杯時や空時の待機はスピンすべきか？ いいえ。待機スレッドはロックを解放して条件変数で待機しなければなりません。スピンループで CPU を消費したり、ロックを保持したままスリープしたりしてはなりません。
- 割り込みはどのように振る舞うべきか？
putとtakeは両方ともInterruptedExceptionを伝播させます。割り込みを握りつぶしたり、割り込まれた操作が失敗した後にキューを変更したりしません。 - 厳密な公平性は必要か？ ベース問題では不要です。デフォルトの非公平な
ReentrantLockは、後から来たスレッドが先にロックを取得することを許容する場合があります。公平なロックはスループットとスケジューリングの挙動を変化させます。 - キューにシャットダウン機能は必要か？ ベース問題では不要です。必要な場合は、既存のアイテムを排出すべきか、待機者が例外を受け取るか特別な結果を受け取るか、そして誰がすべての待機者を起こすかを定義します。
- 線形化された要素の FIFO と待機スレッドの FIFO は同じ保証か？ いいえ。要素は成功した put の線形化順序で取り出されます。これは、ブロックされたプロデューサーやコンシューマーが到着順に受け入れられることを意味するわけではありません。
- 標準ライブラリのクラスを使用してよいか？ 本番コードでは通常、テスト済みの
ArrayBlockingQueueを優先すべきです。ここでの手動実装は、並行処理の不変条件と条件待機セマンティクスを評価することを特に目的としています。
30秒の回答フレームワーク
「固定長配列をリングバッファとして使用し、head、tail、size で次の読み取り位置、次の書き込み位置、現在の要素数を表します。1つの ReentrantLock がすべての共有状態を保護し、2つの Condition オブジェクトが not-empty と not-full を表します。put は while (size == capacity) 内で待機し、要素を挿入して size をインクリメントした後、1つのコンシューマーにシグナルを送ります。take は対称的に not-empty を待機し、head をクリアして size をデクリメントした後、1つのプロデューサーにシグナルを送ります。すべてのチェックと状態遷移は同一のロック下で行われます。await はロックをアトミックに解放し、戻る前に再取得するため、チェックとスリープの間に通知が失われるウィンドウは存在しません。成功する各操作は O(1) であり、空間計算量は O(capacity) です。」
ステップごとの詳細解説
単純なアプローチのボトルネックからデータ表現を導出します。取り出しのたびに残りの要素をシフトする単純な配列では、take が O(n) になってしまいます。単調増加するだけの読み取りインデックスを保持すると、解放された前方部分が無駄になります。固定長のリング配列は解放されたスロットを再利用します。head は次の読み取り位置を指し、tail は次の書き込み位置を指し、size は現在の要素数です。末尾に達したインデックスは0にラップアラウンドするため、挿入も取り出しも既存の要素をシフトしません。
この実装は4つの不変条件を維持します：
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() の本質的な保証は、関連付けられたロックをアトミックに解放して待機状態に入り、戻る前にそのロックを再取得することです。プロデューサーは手動でロックを解除してから待機者として登録してはなりません。そうしてしまうと、プロデューサーが実際に待機を開始する前にコンシューマーがスペースを解放して通知を送信し、その通知が永久に失われるという隙間（ウィンドウ）が生じてしまいます。条件変数は「解放して待機を開始する」という処理を1つの同期アクションにまとめ、そのウィンドウを排除します。
while は2つの異なるケースを処理します。1つは仕様で許可されているスプリアスウェイクアップです。もう1つは通常の競合です。2つのコンシューマーが起こされ、一方が先にロックを再取得して唯一の要素を取り出した場合、もう一方は最終的にロックを取得したときに空のキューを見ることになります。条件通知は状態が変化した可能性があることしか示しません。処理を進めてよいかの最終判断は状態述語が行います。
成功した put の線形化ポイントは、要素を追加して size を k から k + 1 に移行させる、ロックされた状態遷移です。take の場合は、対応する要素の取り出しと k から k - 1 への遷移です。ロックにより、他のスレッドは遷移前の完全な状態か遷移後の完全な状態のいずれかしか観測できず、サイズ更新を伴わない配列への書き込み途中状態を観測することは決してありません。また、ロックの解放とそれに続く同一ロックの取得によってメモリ可視性が確立され、標準的な並行キューで要素を受け渡す際に規定される happens-before の目標を満たします。
なぜ2つの条件変数を使用するのでしょうか？ 1つの待機セットしかない場合、キューが空のままであるにもかかわらず take が別のコンシューマーを起こしてしまい、新しいスロットを利用できるはずのプロデューサーが眠ったままになる可能性があります。notEmpty と notFull を分離することで、各遷移は今進むことができる役割のみにシグナルを送ることができます。ベースとなる単一要素の put または take は新しい要素またはスロットを1つだけ生成するため、signal() で十分です。起床したスレッドは依然として while 内で再チェックを行います。1つの操作で複数のスロットが変化する場合や、シャットダウンですべての待機スレッドが新しい状態を観測する必要がある場合は、signalAll() を検討してください。
正当性は不変条件に対する帰納法によって導かれます。初期状態では head = tail = size = 0 です。挿入は size < capacity のときにのみ実行されます。tail に書き込み、tail を進め、size を正確に1回インクリメントするため、容量内に収まり、消費されていないすべての要素の後に追加されます。取り出しは size > 0 のときにのみ実行されます。head から読み取り、スロットをクリアし、head を進め、size を正確に1回デクリメントするため、最も古い未消費要素を返します。ロックがこれらの遷移を直列化（シリアライズ）するため、多数のプロデューサーとコンシューマーの任意のスケジューリングは、何らかの正当な順次実行と等価になります。
成功した各 put または take は一定数の配列操作、整数操作、および同期操作を実行するため、待機時間を除けばその作業量は O(1) です。固定長配列は O(capacity) の空間を使用します。競合下では、ロック待ちやコンテキストスイッチがレイテンシの支配要因になることがあり、漸近記法ではそのコストを捉えられません。単一ロック設計には複数ロックによる循環はありませんが、呼び出し側が別のロックを保持したままブロッキングメソッドを呼び出すと、より広範なロック順序の問題を引き起こす可能性があります。
テストは単一スレッドの例だけで終わらせてはなりません。容量1での満杯/空の交互動作、複数回のリングラップアラウンド、重複する等しい要素が個別に格納・取り出しされること、空文字列は受け入れられつつ null は拒否されること、複数のプロデューサーによって生成され複数のコンシューマーによって排出された一意の値、最終セットに欠落や重複がないこと、各プロデューサーストリーム内での順序維持、およびブロックされた put とブロックされた take の両方に対する割り込みなどをカバーします。極端に大きな容量を指定すると、コンストラクタで同様に巨大な配列が割り当てられ、メモリ不足で失敗する可能性があります。ベース実装は遅延割り当てではないため、メモリ割り当ての失敗を空のキューとして偽装してはなりません。並行テストでは、バリアやラッチで開始を調整し、テストランナーのデッドラインを使用して永久ブロックを検出する必要があります。短い sleep は、スレッドが待機状態に達したことの証明にはなりません。
質の高い模範解答
「まず、ベースとなる契約を正の容量、非 null 要素、ブロッキングな put/take、FIFO、複数のプロデューサーとコンシューマー、および割り込み可能な待機に絞り込みます。タイムアウト、シャットダウン、厳密な公平性には追加の戻り値と状態セマンティクスが必要になるため、これらはコア実装から除外します。
内部表現は固定長のリング配列です。head は次の読み取り位置、tail は次の書き込み位置であり、size は head == tail のときの満杯と空の曖昧さを排除します。1つの ReentrantLock が4つの共有フィールドすべてを保護します。notEmpty の述語は size > 0 であり、notFull の述語は size < capacity です。
put はロックを割り込み可能に取得し、while 内で not-full を待機し、tail に書き込んで size をインクリメントした後、1つのコンシューマーにシグナルを送ります。take は対称的に not-empty を待機し、head を取り出して参照をクリアし、size をデクリメントした後、1つのプロデューサーにシグナルを送ります。条件待機が擬似起床する可能性があり、また起床したスレッドがロックの再取得を競合している間に別のスレッドが述語を再び偽にする可能性があるため、ループが必須です。
await はロックをアトミックに解放して待機を開始するため、チェック後からスリープまでの間に通知を失うことを防ぎます。成功した操作の線形化ポイントは、キューの所属要素と size を変更するロックされた状態遷移です。同一のロックを使用しているため、他のスレッドは完全な遷移前状態または遷移後状態を観測します。容量、リング FIFO、および同一ロックの不変条件に対する帰納法により、実装がオーバーフローしたり、取り出しが重複したり、要素が並び替えられたりしないことが証明されます。
ブロッキング時間を除けば、両メソッドは O(1) であり、空間計算量は O(capacity) です。ラッチを使用して複数のプロデューサーとコンシューマーを同時に開始し、一意の識別子のセットとプロデューサーごとの順序を検証し、満杯、空、割り込みの各パスを個別にテストします。」
よくある間違い
- 満杯または空のチェックに
ifを使用する → スプリアスウェイクアップやロック競合の後、条件が依然として偽である可能性がある →while内で状態述語を再テストする。 - チェック後に手動でロックを解除してから待機する → 待機者登録の前に状態変化が発生し、通知が失われる可能性がある → 同一ロックに関連付けられた
Condition.await()を使用する。 - 配列、インデックス、
sizeを個別に同期する → 他のスレッドが矛盾した中間状態を観測する可能性がある → チェックと遷移の全体を1つのロックで保護する。 - 任意の
signalを持つ単一の待機セットを使用する → シグナルが処理を進められない同役割のスレッドを起こしてしまう可能性がある →notEmptyとnotFullで別々の条件変数を維持する。 - ロックを保持したままスピンまたはスリープする → 条件を変更できるスレッドがロックを取得できなくなる → 条件待機はロックを解放しなければならない。
InterruptedExceptionを握りつぶす → 呼び出し側が作業をキャンセルできなくなり、スレッドが無期限に残る可能性がある → 割り込みを宣言して伝播させ、finallyでロックを解放する。- 両方の状態を検出するために
head == tailのみを使用する → リングの満杯状態と空状態が区別できない → ロックで保護されたsizeを維持する。 - 取り出した参照を配列内に残す → 配列が必要以上に長く消費済みオブジェクトを保持してしまう（メモリリーク） → 読み取り後にそのスロットに
nullをセットする。 - 要素の FIFO とスレッドの公平性を同一視する → デフォルトのロックは待機中の呼び出しを到着順に完了させない → 要素の順序付けとスケジューリングポリシーを明確に区別して説明する。
- ベース実装でシャットダウンをサポートしていると主張する → 待機スレッドには観測可能な closed 状態がなく、永久に起きない可能性がある → 状態とブロードキャスト通知を追加する前に、シャットダウンの契約を定義する。
発展的な質問と回答
発展質問1: タイムアウト付きの offer と poll はどのように追加しますか？
残りの期間をナノ秒に変換し、同じ述語 while ループ内で awaitNanos(remaining) を呼び出します。復帰するたびに、報告された残り時間を使ってループを継続します。値が0以下になり、かつ述語が依然として偽である場合は失敗を返します。スプリアスウェイクアップのたびにタイムアウト時間全体を再利用すると、実際の待機時間が無期限に延長される可能性があります。また、API はタイムアウトと null 要素を区別する必要があり、これも null を拒否する理由の1つです。
発展質問2: shutdown() はどのように実装しますか？
まず状態遷移モデルを定義します。たとえば、クローズ後は新しい put を拒否しますが、既存のアイテムの排出は許可します。クローズされていて空の場合、take は専用の例外をスローするか、明示的な結果を返します。shutdown は同じロック下で closed フラグを変更し、すべての待機スレッドがロックを再取得してクローズ状態を観測できるように両方の条件で signalAll() を呼び出す必要があります。すべての待機ループ述語に closed 状態の分岐が必要です。単に boolean フィールドを追加するだけでは不十分です。
発展質問3: なぜここで signal() を使用するのですか？ また signalAll() はいつ使用しますか？
ベース実装の1回の挿入は1つの消費可能要素を生成し、1回の取り出しは1つの空きスロットを生成します。反対の役割のスレッドを1つだけ起こせば処理を進めるのに十分であり、無駄な競合を減らすことができます。一括挿入、一括取り出し、動的な容量変更、またはシャットダウンでは、複数の待機スレッドが一度に対象となる可能性があるため、通常は signalAll() または状態変化に合わせた回数の通知が必要です。いくつのスレッドが起こされたとしても、while による再チェックは必須です。
発展質問4: 公平性はどのように提供しますか？
new ReentrantLock(true) は、ロック取得時に最も長く待機しているスレッドを優先させますが、それでも put/take の絶対的なリアルタイム完了順序を提供するわけではありません。割り込みやスケジューリングも影響するためです。公平なポリシーは一般にバージング（割り込み割り当て）や飢餓（starvation）のリスクを減らしますが、スループットを低下させる可能性があります。呼び出し側の契約が待機スレッドの順序付けを真に要求している場合にのみ、そのコストを受け入れて検証してください。
発展質問5: これをロックフリーキューにすることはできますか？
ロックフリーの有界 MPMC キューには通常、アトミックなシーケンス番号、CAS、およびはるかに複雑なメモリーオーダーの証明が必要です。また、その「ブロッキング」動作には依然としてスレッドのパーキングと起床のメカニズムが必要であり、CAS のスピンだけではブロッキングキューになりません。この設計は、ABA 問題、偽共有（false sharing）、進行保証、プラットフォームのメモリモデルに関する懸念を追加します。測定によって単一ロックがボトルネックであることが示され、チームが証明とストレステストを維持できる場合を除き、標準ライブラリのクラスまたは明確なロックベースの実装の方が安全です。
発展質問6: なぜ2つのセマフォを直接使用しないのですか？
1つの計数セマフォで空きスロットを表し、もう1つで利用可能な要素を表すことができますが、リングの head/tail の更新には依然として相互排他が必要です。また、複数の同期プリミティブを取得する場合は、例外、割り込み、および許可（permit）のロールバック処理を慎重に行う必要があります。セマフォによる解決策も正しく作成できますが、1つのロックと2つの条件変数を使用する構成よりも自動的にコードが短くなるわけではありません。どちらの設計でも、許可数と実際の配列状態が決して乖離しないことを証明する必要があります。