Javaの並列プログラミング

並列問題とは、マルチスレッド環境において、オペレーティングシステムのスレッドスケジューリング、JITコンパイラによる最適化、プロセッサの命令再順序化、およびマルチレベルキャッシュの影響により、プログラムの実行結果が期待される挙動と一致しない問題を指します。

• 並列処理とは、同一時間内に複数のタスクが交互に実行されること
• 同時実行とは、同一瞬間に複数のタスクが実際に同時に実行されること

Javaのスレッドには6つの状態があり、以下のように分類できます：

• 新規(New): スレッドオブジェクトは作成されたがstart()が呼び出されておらず、スケジューリングには入っていない状態
• 実行可能(Runnable): start()が呼び出された後、実行中かまたはCPUスケジューリングを待機中の状態
• ブロック(Blocked): synchronizedロックの競合によりブロックされている状態
• 待機(Waiting): タイムアウトなしの待機状態（wait()など）、他のスレッドに明示的に起動される必要がある
• タイムド待機(Timed Waiting): タイムアウト付きの待機状態（sleep(1000)など）、タイムアウトまたは起動後に復帰する
• 終了(Terminated): run()の実行完了または異常終了により、ライフサイクルが終了した状態

JMM（Java Memory Model）

JUCが登場する前、Javaはvolatileやsynchronizedなどの同期プリミティブを提供し、開発者が並列問題を解決する手段としていました。これらの同期プリミティブの実装を理解することは、スレッドの並列問題解決の考え方を整理し、JUCの実装を理解する上で役立ちます。

Javaメモリモデル（JMM、Java Memory Model）はJava仮想マシン仕様で定義された抽象的なメモリモデルであり、その核心的な目標は異なるハードウェアやオペレーティングシステムのメモリアクセスの差異を隠蔽し、マルチスレッドプログラムに一貫したメモリの可視性、順序性、および原子性を保証することです。これにより、開発者は異なるプラットフォームで書いた並列コードが予測可能な動作を示すことができます。

happens-beforeはJMMが定義する操作間の可視性と順序性の制約関係であり、JMMを理解する上で鍵となります。その核心的な定義は、操作Aが操作B「happens-before」の場合、Aの実行結果はBに可視であり、Aの論理順序はBの前にある（物理的な実行順序とは無関係）ということです。

1. プログラム順序ルール: 同じスレッド内では、前の操作は後の操作にhappens-before
2. volatileルール: volatile変数への書き込み操作は、後続の読み取り操作にhappens-before
3. ロックルール: ロックの解放操作は、後続の同一ロックの取得操作にhappens-before

volatileの実装

• 特性: volatile変数の読み書き命令がマルチスレッド間で可視性を保証する
• JMM仕様: volatile変数への書き込み操作は、その変数の読み取り操作にhappens-before
• JMM実装:
• volatile書き込み操作後にメモリバリアを挿入:
• StoreStoreバリア: 通常の書き込み（非volatile変数の書き込み）がvolatile書き込み後に再順序化されるのを禁止し、volatile書き込み前のすべての変数変更が完了することを保証
• StoreLoadバリア: volatile変数の新しい値をCPUキャッシュからメインメモリに強制的にフラッシュし、他のCPUコアでのその変数のキャッシュを無効化する。これにより他のスレッドが最新値を読み取ることができる（可視性の保証）
• volatile読み込み操作前にメモリバリアを挿入:
• LoadLoadバリア: 他の読み込み操作（非volatile変数の読み込み）がvolatile読み込み前に再順序化されるのを禁止し、volatile読み込みが優先的に実行されることを保証
• LoadStoreバリア: 書き込み操作がvolatile読み込み前に再順序化されるのを禁止し、volatile読み込みの結果が後続の書き込み操作によって上書きされるのを防ぐ
• X86での実装:
• 共有データ変更の可視性を保証し、キャッシュ一貫性プロトコルを使用。現在のCPUでの変数キャッシュ書き込みが物理メインメモリに書き込まれる前に、他のCPUはその変数をメインメモリに書き込むことができず、操作終了後には他のCPUのキャッシュを無効化して物理メインメモリからデータを読み取らせることで、CPUキャッシュとメインメモリを同期させる。以前はバスロック方式もありましたが、これは他のプロセッサにlock信号を送信し、自身がメインメモリを独占することを宣言する方法でした。
• 原子操作の順序性を保証し、lock接頭辞命令前後の命令の再順序化を禁止します。

CAS（Compare-And-Swap）

CAS（Compare-And-Swap、比較と交換）はハードウェアレベルの原子操作メカニズムで、並列環境下での共有変数の安全な変更を実現するために使用されます。その核心的な考え方は、共有変数を変更する前に、まず変数の現在値が期待値と一致するかを確認することです。一致した場合は新しい値に更新し、一致しない場合は変更を行いません（通常は再試行または放棄します）。このプロセスは原子性を持つ（他のスレッドによって中断されることがない）ため、マルチスレッドによる並行変更がもたらすデータ不整合問題を回避できます。

synchronizedロックのエスカレーション

重いロック（重量級ロック）の使用コストが高いため、Javaではロックのエスカレーションメカニズムが導入されています。synchronizedには4つのロック状態があり、無锁、偏向锁、轻量级锁、重量级锁があります。

synchronizedのロック状態（無锁、偏向锁、轻量级锁、重量级锁）はオブジェクトヘッダ（Object Header）に保存されます。オブジェクトヘッダはJavaオブジェクトのメモリ内の先頭情報であり、その中のMark Word（マークワード）がロック状態を記録するために特化されています。構造はロック状態に応じて動的に変化します（64ビットJVMの場合）：

• 偏向锁の状態では、スレッドが初めてロックを取得する際にCASを使用して自身のスレッドIDをオブジェクトヘッダのMark Wordに書き込みます。2回目にスレッドが入った際に、現在のスレッドIDとMark Wordに記録されたスレッドIDを比較し、同じであればリソースを取得します。異なる場合は、スレッド競合が発生したことを示し、エスカレートして轻量级锁になります。
• 轻量级锁の状態では、スレッドがクリティカルセクションに入る際に、現在スレッドのプライベートメモリ内でLock Recordを作成し、競合するリソース情報を記録します。その後、CASを使用してMark Wordのポインタを現在スレッドのLock Recordに更新し、成功した場合は現在スレッドがリソースを取得したことを示し、失敗した場合はスピン状態に入ります。スピン状態ではスレッドが複数回のCAS操作を試み、成功すればリソースを取得し、失敗すればエスカレートして重量级锁になります。
• 重量级锁の実装では、オブジェクトがsynchronizedでロックされると、オブジェクトヘッダのMark WordはObjectMonitorを指します。ObjectMonitorはJava言語におけるモニター概念の具体的な実装であり、ObjectMonitorにはリソースの競合プロセスを維持するための一連の変数とデータ構造が含まれています。ObjectMonitorのOwnerフィールドはObjectMonitorを保持するスレッドを指しており、スレッドによるリソース（ロック）の競合はCAS同期プリミティブを使用してOwnerフィールドを自身に指し示すプロセス（スピン）と考えることができます。ObjectMonitorのEntryListフィールドはクリティカルセクションに入ったがリソースを取得できなかったスレッドを管理しており、これらのスレッドはブロック状態です。ロックが解放されると、1つまたは複数のブロックされたスレッドを起動し、再びロックを競合させます（新しい競合は非公平戦略に従い、新しいスレッドが割り込む可能性があります）。ObjectMonitorのWaitSetフィールドはリソースを取得したが、何らかの理由でリソースを解放する必要があるスレッドを管理します。waitメソッドを使用してこれらのスレッドをWaitSetに入れ、notify/notifyAllでスレッドを起動してEntryListに追加し、再びリソースを競合させます。

synchronizedロックの最適化

• ロックの除去: **特定のロックオブジェクトが複数のスレッドからアクセスされない（つまり並行競合がない）**と検出された場合、JVMはコンパイル時にそのロックの同期操作を自動的に削除し、不要なロックのオーバーヘッドを回避します。
• ロックの粗大化: 複数の連続したロック取得と解放の操作は、1回のロック取得と解放に簡略化できます。例えばStringBufferオブジェクトの連続した複数回のappendは、最初のappendメソッドでロックを取得し、最後のappendメソッドでロックを解放するように最適化できます。

JUC（java.util.concurrent）

ReentrantLockとSynchronizedの違い

1. SynchronizedはJavaのキーワードであるのに対し、ReentrantLockはJUCの同期コンポーネントです。
2. Synchronizedはブロッキング方式でのみロックを取得できます（一度ブロック状態に入ると、ロックを取得するまで待機し続けます）が、ReentrantLockはブロッキング方式でのロック取得（lock）、試行によるロック取得（tryLock）、割り込み可能なロック取得（lockInterruptibly - スレッドがロック待機中に割り込みを受けると、即座に待機を終了して例外をスロー）、タイムアウト付きのロック取得（tryLock（timeout））などの機能を提供します。
3. Synchronizedは非公平ロックですが、ReentrantLockは公平ロックと非公平ロックの両方を実装しています。
4. ReentrantLockの複数のConditionキューにより、スレッドを条件ごとにグループ化して待機させ、起動時に特定の条件のスレッドのみを起動することができ、細かい制御を実現します。一方、Synchronizedは1つの待機キューに対応するのみです。

AQS（AbstractQueuedSynchronizer）

AQSの核心は2つの部分から構成され、同期メカニズムの実現を支えています：

1. 同期状態（State）

• 定義: volatile int state変数が同期状態を表し（スレッド競争の核心リソース）、その意味はサブクラスがシナジオに応じてカスタマイズします。
• 特性: volatileは状態の可視性を保証し、AQSはgetState()、setState(int)、compareAndSetState(int expect, int update)メソッド（CASベース）を提供してstateを操作し、スレッドセーフティを保証します。
• 例示的な意味:
• ReentrantLockでは、stateは「再入回数」（0は未ロック、>0はロック中、値は再入回数）を表します。
• Semaphoreでは、stateは「残り許可数」（スレッドが許可を取得すると減少し、解放すると増加）を表します。
• CountDownLatchでは、stateは「カウントダウン回数」（0になるとすべての待機スレッドを起動）を表します。

2. 同期キュー（FIFO双方向リンクリスト）

• スレッドが同期リソースの競争に失敗した場合、AQSはスレッドをNodeノードとしてラップし、FIFO双方向リンクリストに追加します。

Semaphore

特定のリソースに同時にアクセスするスレッド数を制御します

// データベースに同時にアクセスできるスレッドを最大3に制限するセマフォ
Semaphore dbSemaphore = new Semaphore(3);

// 10個のスレッドが競合してアクセス
for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        try {
            dbSemaphore.acquire(); // 許可を取得（満杯の場合は待機）
            System.out.println("スレッド " + Thread.currentThread().getId() + " がデータベースにアクセス中");
            Thread.sleep(1000); // データベース操作をシミュレート
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("スレッド " + Thread.currentThread().getId() + " が割り込まれました");
        } finally {
            dbSemaphore.release(); // 許可を解放（他のスレッドが取得可能に）
            System.out.println("スレッド " + Thread.currentThread().getId() + " が接続を解放しました");
        }
    }).start();
}

CountDownLatch

1つまたは複数のスレッド（待機側）が他のスレッド（実行側）が指定された操作を完了するのを待ってから実行を続けます

// 3つのイベントを待機するカウントダウンロケーター
TaskCounter counter = new TaskCounter(3);

// 3つの子スレッドを起動
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            System.out.println("子スレッド " + Thread.currentThread().getId() + " がタスクを開始");
            Thread.sleep(1000); // タスクをシミュレート
            System.out.println("子スレッド " + Thread.currentThread().getId() + " がタスクを完了");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("子スレッド " + Thread.currentThread().getId() + " が割り込まれました");
        } finally {
            counter.decrement(); // イベント完了、カウントを減らす
        }
    }).start();
}

try {
    counter.await(); // 主スレッドが待機、3つの子スレッドが完了するまで（state == 0）
    System.out.println("すべての子スレッドが完了し、主スレッドが実行を再開");
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    System.out.println("主スレッドが割り込まれました");
}

CyclicBarrier

一組のスレッドが互いに待ち合わせ、すべてのスレッドが特定の「障壁点」に到達したら、その後の操作を一緒に実行します

// 3つのスレッドが到達したらアクションを実行する障壁
PhaseBarrier barrier = new PhaseBarrier(3, () -> {
    System.out.println("すべてのスレッドが障壁に到達、次のフェーズに移行します...");
});

for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            // 第一フェーズの作業
            System.out.println("スレッド " + Thread.currentThread().getId() + " が第一フェーズを完了");
            barrier.await(); // 障壁に到達、他のスレッドを待つ

            // すべてのスレッドが到達したら、第二フェーズを一緒に開始
            System.out.println("スレッド " + Thread.currentThread().getId() + " が第二フェーズを開始");
        } catch (Exception e) {
            Thread.currentThread().interrupt();
            System.out.println("スレッド " + Thread.currentThread().getId() + " が割り込まれました");
        }
    }).start();
}

スレッドプール

Javaのスレッドプールはスレッド管理メカニズムであり、あらかじめ一定数のスレッドを作成して再利用することで、スレッドの頻繁な作成と破棄によるリソースオーバーヘッド（メモリ割り当て、CPUスケジューリングコストなど）を回避し、同時に並列スレッド数の制御を実現してシステムの安定性とリソース利用率を向上させます。

• corePoolSize（コアスレッド数）: スレッドプールが長期的に保持するスレッド数（スレッドがアイドル状態でも破棄されませんが、allowCoreThreadTimeOutがtrueに設定されている場合は除く）。 タスクが提出された際に、コアスレッドが満たされていない場合は優先的にコアスレッドを作成してタスクを実行します。

• maximumPoolSize（最大スレッド数）: スレッドプールが作成を許可する最大スレッド数（コアスレッド + 非コアスレッド）。 コアスレッドが満杯でかつワークキューも満杯の場合は、最大スレッド数に達するまで非コアスレッドを作成してタスクを実行します。

• keepAliveTime+unit（非コアスレッドのアイドル時間）: 非コアスレッド（corePoolSizeを超えるスレッド）が指定時間以上アイドル状態の場合は破棄され、リソースが解放されます。 allowCoreThreadTimeOut(true)を設定することで、コアスレッドもこのルールに従うようにできます。

• workQueue（ワークキュー）: 「コアスレッドが満杯」だが「最大スレッド数に達していない」場合の待機タスクを格納するために使用され、**ブロッキングキュー（BlockingQueue）**でなければなりません。 主な実装: LinkedBlockingQueue（無限キュー）、ArrayBlockingQueue（有限キュー）、SynchronousQueue（同期キュー、タスクを直接スレッドに渡す、タスクを保存しない）。

• threadFactory（スレッドファクトリ）: スレッドの作成方法（スレッド名、デーモンスレッドかどうかなど）を定義します。デフォルトではExecutors.defaultThreadFactory()が使用されます。

• handler（拒否ポリシー）: 「ワークキューが満杯で最大スレッド数に達している」場合に、新たに提出されたタスクが拒否ポリシーをトリガーします（例外スロー、破棄など）。

スレッドプールにタスク（execute(Runnable)またはsubmit(Callable)）を提出する際の実行フローは以下の通りです：

1. コアスレッドがアイドルかどうかを判断: コアスレッドが満たされていない場合（現在のスレッド数 < corePoolSize）、コアスレッドを作成してタスクを実行し、即座に戻ります。

2. コアスレッドが満杯、ワークキューが満杯でないかを判断: ワークキューが満杯でない場合、タスクをキューに追加して待機させ、コアスレッドがキューからタスクを取得して実行します（スレッドの再利用）。

3. キューが満杯、最大スレッド数に達していないかを判断: 現在のスレッド数 < maximumPoolSizeの場合、非コアスレッドを作成してタスクを実行します。

4. 最大スレッド数に達した場合、拒否ポリシーを実行: RejectedExecutionHandlerが新しいタスクを処理します（例外スロー、破棄など）。