Executor は Runnable の送信と実行場所の決定を分離します。ExecutorService は
ライフサイクル管理と submit を追加します。Callable<V> は値を返すか検査例外を
スローでき、Future<V> は最終的な結果を表します。Future.get() は、タスクが完了
する前に行ったアクションに必要な可視性を確立します。
Future<Result> future = pool.submit(() -> compute(input));
try {
return future.get();
} catch (ExecutionException e) {
throw new ServiceException(e.getCause());
}synchronized、Lock、アトミック、並行コレクション、メッセージパッシングを使って
happens-before 関係を確立します。エグゼキューターが提供するのはスケジューリングで
あり、可変共有オブジェクトの自動的な安全性ではありません。
AQS と同期化機構 #
AbstractQueuedSynchronizer（AQS）は、ロック、セマフォ、ラッチなどの同期化機構向け
フレームワークです。サブクラスはアトミックな整数状態を保持し、取得・解放の試行を
実装します。競合時、AQS は待機スレッドを FIFO 形式の同期キューに追加し、
LockSupport で park します。解放時には適格な後続スレッドを unpark します。
Condition は関連する条件キューを使い、通知時に待機者を同期キューへ戻します。
キューの順序だけでは、普遍的な公平性は保証されません。
ThreadPoolExecutor の受け入れ #
ThreadPoolExecutor はまず corePoolSize までワーカーを作成し、その後タスクを
キューに入れます。キューがタスクを拒否した場合にだけ maximumPoolSize まで増加し、
その後は拒否します。そのため、上限のない LinkedBlockingQueue は実質的にコア
スレッドを超える増加を防ぎます。一方、SynchronousQueue は最大数まで積極的に
ワーカーを作成します。コアサイズを超えるアイドルワーカーは keepAliveTime 後に
終了し、明示的に有効にすればコアスレッドもタイムアウトします。
有界キューを選び、ワークロードの実測、メモリ制限、下流システムの容量に基づいて
サイズを決めます。ワーカーファクトリは名前、デーモン状態、キャッチされない例外の
処理を制御します。AbortPolicy は例外をスローし、CallerRunsPolicy は送信元で
実行することでバックプレッシャーを適用し、DiscardPolicy は作業を破棄し、
DiscardOldestPolicy は再試行前にキュー内の古い作業を削除します。呼び出し側が
許容できる場合にだけ、無言の破棄を認めます。
ライフサイクル、キャンセル、可視性 #
cancel(true) は Future をキャンセル済みとしてマークし、タスクが実行中なら割り
込みを要求します。割り込みは協調的であるため、ブロッキング操作は割り込み状態を
伝播または復元し、CPU バウンドのタスクは状態をポーリングする必要があります。
キャンセルによって副作用がロールバックされることはありません。
shutdown() は新しいタスクを拒否し、キュー内のタスクは完了させます。
shutdownNow() はワーカーへの割り込みを試み、開始されていないタスクを返します。
アプリケーションの境界で awaitTermination を呼び出します。現在の JDK の
Executors ファクトリメソッドは、固定、単一スレッド、スケジュール、ワーク
スティーリング、仮想スレッドのエグゼキューターを提供しますが、ThreadPoolExecutor
を明示的に構築すればキューイングと拒否ポリシーを確認できます。ExecutorService
の寿命がレキシカルな場合は、必ず閉じてください。
コメント
GitHub で閲覧・コメントする（投稿にはログインが必要です）