スレッドプールは、リソースの効率的な再利用とコンテキストスイッチングのオーバーヘッド削減を目的として設計されています。本稿では、JUC（java.util.concurrent）におけるThreadPoolExecutorの実装と活用方法、およびスレッド数の最適化について解説します。
カスタムスレッドプールの構築
基本的なスレッドプールは、タスクキューとワーカースレッド群で構成されます。タスクが投入されると、空きスレッドがあれば即時実行、なければキューに格納して待機します。
class TaskQueue<T> {
private final Deque<T> tasks = new ArrayDeque<>();
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
private final int capacity;
public TaskQueue(int cap) { this.capacity = cap; }
public void offer(T task) throws InterruptedException {
lock.lock();
try {
while (tasks.size() >= capacity) notFull.await();
tasks.addLast(task);
notEmpty.signal();
} finally { lock.unlock(); }
}
public T poll() throws InterruptedException {
lock.lock();
try {
while (tasks.isEmpty()) notEmpty.await();
T task = tasks.removeFirst();
notFull.signal();
return task;
} finally { lock.unlock(); }
}
}
class SimplePool {
private final Set<Worker> workers = ConcurrentHashMap.newKeySet();
private final TaskQueue<Runnable> queue;
private final int maxThreads;
public SimplePool(int size, int qCap) {
this.maxThreads = size;
this.queue = new TaskQueue<>(qCap);
}
public void submit(Runnable task) {
synchronized (workers) {
if (workers.size() < maxThreads) {
Worker w = new Worker(task);
workers.add(w);
w.start();
} else {
queue.offer(task);
}
}
}
class Worker extends Thread {
private Runnable current;
Worker(Runnable initial) { this.current = initial; }
@Override
public void run() {
while (current != null || (current = queue.poll()) != null) {
try { current.run(); } finally { current = null; }
}
synchronized (workers) { workers.remove(this); }
}
}
}
拒否ポリシーの柔軟な実装
キューが満杯になった場合の挙動を、戦略パターンで柔軟に切り替え可能にします。
@FunctionalInterface
interface RejectionHandler<T> {
void handle(TaskQueue<T> q, T task);
}
// 使用例: タイムアウト付き投入
pool.submit(task, (q, t) -> {
if (!q.offerWithTimeout(t, 1, TimeUnit.SECONDS)) {
System.err.println("Task rejected: " + t);
}
});
ThreadPoolExecutorの詳細
JDK標準のThreadPoolExecutorは、状態管理とスレッド数を32ビット整数で同時に管理しています。上位3ビットが状態（RUNNING/SHUTDOWN等）、下位29ビットがアクティブスレッド数です。これにより、状態遷移時のアトミック性を保証します。
主要コンストラクタパラメータ
- corePoolSize: 常駐スレッド数
- maximumPoolSize: 最大スレッド数（救急スレッド含む）
- keepAliveTime: 救急スレッドの生存時間
- workQueue: 待機タスクキュー
- threadFactory: スレッド生成ファクトリ
- handler: 拒否ハンドラ
事前定義されたファクトリメソッド
// 固定サイズ（常駐スレッドのみ）
Executors.newFixedThreadPool(4);
// キャッシュ型（0常駐、無制限救急スレッド）
Executors.newCachedThreadPool();
// 単一スレッド（直列実行保証）
Executors.newSingleThreadExecutor();
// スケジュール対応
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
scheduler.scheduleAtFixedRate(task, 0, 1, TimeUnit.SECONDS);
タスク結果の取得と例外処理
Futureインタフェース経由で非同期結果を取得できます。例外はget()呼び出し時にExecutionExceptionとして伝播されます。
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() -> {
if (Math.random() < 0.5) throw new RuntimeException("Oops");
return "Success";
});
try {
String result = future.get(); // 成功時のみ値が返る
} catch (ExecutionException e) {
Throwable cause = e.getCause(); // 元の例外を取得
}
シャットダウン制御
shutdown(): 新規タスク受付停止。既存タスクは完了まで実行。shutdownNow(): 実行中タスクをinterrupt()で中断。未実行タスクを返却。awaitTermination(): 全タスク完了を待機。
スレッド数の最適化
適切なスレッド数はワークロード特性により異なります。
CPUバウンド処理
int optimal = Runtime.getRuntime().availableProcessors() + 1;
I/Oバウンド処理
理論値 = コア数 × 目標CPU使用率 × (計算時間 + 待機時間) / 計算時間
// 計算時間が10%、待機90%の場合
int threads = cores * 1.0 * (1.0) / 0.1; // コア数の10倍
デッドロック回避のための設計原則
異なる性質のタスク（例：受注処理と調理処理）は、別々のスレッドプールで処理すべきです。
ExecutorService orderPool = Executors.newFixedThreadPool(2);
ExecutorService cookPool = Executors.newFixedThreadPool(4);
orderPool.execute(() -> {
// 注文受付
Future<Dish> dish = cookPool.submit(() -> prepareDish());
serveCustomer(dish.get()); // 調理完了を待機
});