ExecutorServiceとスレッドプール: 問題ごとに正しい型を選択する

1. 概要

毎回のタスクに対してnew Thread()を作成するのは簡単に書けますが、負荷が増えると制御が難しくなります。ExecutorServiceはスレッドを再利用してタスクをキューイングすることでこれを解決します。Executorsクラスは異なるプールタイプを提供しており、間違ったタイプを選択するとOutOfMemoryErrorまたはアプリケーションハングの原因になる可能性があります。

このガイドではプールの種類、ThreadPoolExecutorがタスクを処理する方法、プールのサイズを選択する方法、Spring Boot 3.xでの設定方法を説明します。読み終わったら、CPU限定タスク、I/O限定タスク、定期的なタスクに使用するプールの種類を知ることができます。

2. ExecutorService の基本

以下の例は、単純なプールを使用して結果を返すタスクを実行する方法を示しています:

import java.util.concurrent.*;

public class OrderReportApp {

    public static void main(String[] args) throws Exception {
        try (ExecutorService executor = Executors.newFixedThreadPool(4)) {
            Future<Integer> total = executor.submit(() -> countOrders("2026-10"));
            System.out.println("合計注文数: " + total.get());
        }
    }

    static int countOrders(String month) {
        return 128; // データベースクエリをシミュレート
    }
}

合計注文数: 128

コードは次のように動作します:

submit()はCallableを受け取り、すぐにFuture<Integer>を返します。タスクはプールのスレッド上で実行されます。
total.get()は結果が得られるまで現在のスレッドをブロックします。
Java 19以降、ExecutorServiceはAutoCloseableを実装しています。tryブロックを出ると、close()がshutdown()を呼び出し、すべてのタスクが完了するまで待機します。

注意: executorを閉じない場合、非デーモンスレッドがJVMを生き続けさせます。mainプログラムは終了しません。

3. `Executors`のプール型

3.1. `newFixedThreadPool(n)`

このプールは正確にn個のスレッドを持ちます。余ったタスクは制限なしの LinkedBlockingQueueに入れられます。

ExecutorService invoiceExecutor = Executors.newFixedThreadPool(8);

このタイプは、実行中のタスク数を制限したいとき（例えばCPU限定タスク）に適しています。リスクはキューにあります。producerがタスクを処理速度より速く投入すると、キューが拡大し続けヒープが枯渇するまで続きます。

3.2. `newCachedThreadPool()`

このプールは固定スレッド数がありません。SynchronousQueueを使用しており、実際のキューがありません。新しいタスクは、空いているスレッドがあればそれに割り当てられ、なければプールが新しいスレッドを作成します。60秒以上アイドルのスレッドは破棄されます。

ExecutorService notificationExecutor = Executors.newCachedThreadPool();

このプールは多くの短いタスクと低く不均等な負荷に適しています。負荷が突然増加すると、数千のプラットフォームスレッドを作成する可能性があります。各スレッドはデフォルトでスタックサイズが約1 MBなので、アプリケーションはOutOfMemoryError: unable to create native threadが発生しやすくなります。

3.3. `newSingleThreadExecutor()`

このプールは1つのスレッドのみを持ち、タスクは投入順に厳密に順序付けされて実行されます。

ExecutorService auditLogExecutor = Executors.newSingleThreadExecutor();
auditLogExecutor.submit(() -> System.out.println("Order #1 created"));
auditLogExecutor.submit(() -> System.out.println("Order #1 paid"));

Order #1 created
Order #1 paid

2つのログ行は常に正しい順序で出力されます。このタイプは、ロックを使わずに順序を保つ必要があるログ記録、ファイル書き込み、または状態更新に便利です。スレッドが例外で死亡する場合、プールは置換用の新しいスレッドを作成します。そのキューも制限なしです。

3.4. `newScheduledThreadPool(n)`

このタイプはスケジュール済みまたは定期的なタスク用です:

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

scheduler.scheduleAtFixedRate(
        () -> System.out.println("在庫同期 @ " + java.time.LocalTime.now()),
        0, 5, TimeUnit.SECONDS);

scheduleAtFixedRateは、前回の実行の開始から5秒ごとにタスクを実行します。対照的に、scheduleWithFixedDelayは前回の実行が終了してから5秒をカウントします。タスクの実行時間が不安定な場合は、scheduleWithFixedDelayを選択すべきです。

注意: 定期的なタスクが例外をスローする場合、後続のすべての実行はサイレントにキャンセルされ、ログはありません。常にタスク本体をtry/catchでラップしてください。

3.5. `newWorkStealingPool()`

このメソッドはCPU数と等しいスレッド数でForkJoinPoolを返します。各スレッドは独自のキューを持ちます。アイドルスレッドは他のスレッドのキューから仕事を「盗む」ことができます。

ExecutorService pricingExecutor = Executors.newWorkStealingPool();

このタイプはCPU限定タスクに適しており、再帰的に分割できます（例えば大規模データセットの並列計算）。プールのスレッドはデーモンであり、プールは実行順序を保証しません。

3.6. `newVirtualThreadPerTaskExecutor()` (Java 21)

このタイプでは、各タスクは新しいvirtual threadで実行されます。Virtual threadは非常に安価であり、数十万個を作成できます。

try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        int customerId = i;
        executor.submit(() -> {
            Thread.sleep(100); // CustomerServiceへのHTTP呼び出しをシミュレート
            return customerId;
        });
    }
}

上記のコードは10,000個のタスク（各100 msブロック）を数百ミリ秒で完了します。10スレッドの固定プールを使用した場合、同じ量の作業には約100秒かかります。

これはI/O限定タスク（HTTP呼び出し、データベースクエリ、ファイル読み取りなど）のデフォルトの良い選択です。3つのことを覚えておく必要があります:

Virtual threadをプールしないでください。 それらは1回の使用目的で作成されます。
外部リソースの負荷を制限したい場合は、Semaphoreを使用してください。 たとえば、データベースが20接続のみをサポートする場合、プールサイズに依存してはいけません。
Java 21ではブロッキング操作を持つsynchronizedブロックはvirtual threadをcarrier threadに「ピン留め」します。 その場合、virtual threadの利点が失われます。そのような場所ではReentrantLockを優先してください。

Semaphoreで同時クエリ数を制限する例:

Semaphore dbPermits = new Semaphore(20);

executor.submit(() -> {
    dbPermits.acquire();
    try {
        return orderRepository.findByCustomerId(customerId);
    } finally {
        dbPermits.release();
    }
});

4. ThreadPoolExecutor の自己設定

Executorsのファクトリメソッドは通常、危険なパラメータを隠しています: 制限なしのキューまたは制限なしのスレッド数。本番環境のシステムでは、ThreadPoolExecutorを自分で初期化してすべてを制御する必要があります:

ThreadFactory factory = Thread.ofPlatform().name("payment-", 0).factory();

ThreadPoolExecutor paymentExecutor = new ThreadPoolExecutor(
        4,                                   // corePoolSize
        8,                                   // maximumPoolSize
        60, TimeUnit.SECONDS,                // core超過スレッドの保持期間
        new ArrayBlockingQueue<>(100),       // 制限付きキュー
        factory,                             // 名前付け: payment-0, payment-1...
        new ThreadPoolExecutor.CallerRunsPolicy());

新しいタスクが到着すると、ThreadPoolExecutorは正確な順序で処理します:

スレッド数がcorePoolSize未満の場合、プールは新しいスレッドを作成します。
そうでなければ、プールはタスクをキューに入れます。
キューがいっぱいで、スレッド数がmaximumPoolSize未満の場合、プールは追加スレッドを作成します。
両方の制限に達する場合、プールはRejectedExecutionHandlerを呼び出します。

その結果、maximumPoolSizeはキューがいっぱいの場合のみ有効になります。制限なしのキューを使用する場合、プールはcorePoolSizeを超えることはありません。これは非常に一般的な誤解の原因です。

Javaは4つのrejection policyを提供します:

ポリシー	動作
`AbortPolicy` (デフォルト)	`RejectedExecutionException`をスロー
`CallerRunsPolicy`	`execute()`を呼び出したスレッドがタスクを自分で実行し、back-pressureを作成します
`DiscardPolicy`	タスクをドロップ、通知なし
`DiscardOldestPolicy`	キューの最古のタスクをドロップして再試行

ThreadFactoryを通じてスレッドに名前を付けると、スレッドダンプとログが読みやすくなります。payment-3とpool-7-thread-3を比べるとその違いは明らかです。

5. 正しい型を選択する

5.1. プールサイズ

CPU限定タスクの場合、スレッド数はコア数に近い値にするべきです:

int cores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuExecutor = Executors.newFixedThreadPool(cores);

コア数より多くのスレッドはコンテキストスイッチのコストを増加させるだけです。

プラットフォームスレッド上で実行されるI/O限定タスクの場合、おおよその公式を使用できます:

スレッド数 ≈ コア数 × (1 + 待機時間 / 計算時間)

例: 8コアのマシン、各リクエストがI/Oで90 ms、計算で10 ms待つ。結果は8 × (1 + 9) = 80スレッド。これはスタート地点です。実際のロードテストで測定が必要です。

5.2. 決定表

問題	選択肢
CPU集約的な計算	`newFixedThreadPool(cores)`または`ForkJoinPool`
大量のAPI呼び出し、DB クエリ	`newVirtualThreadPerTaskExecutor()` + `Semaphore`
順序が重要	`newSingleThreadExecutor()`
定期的なタスク	`newScheduledThreadPool(n)`
本番環境、back-pressure必要	制限付きキューの`ThreadPoolExecutor`
突発的な負荷、制限なし	`newCachedThreadPool()`を回避

6. Spring Boot 3.x のスレッドプール

まず、Maven依存関係が必要です:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

次に、メール送信用の専用executorを宣言します:

@Configuration
@EnableAsync
public class AsyncConfig {

    @Bean(name = "emailExecutor")
    public ThreadPoolTaskExecutor emailExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("email-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        return executor;
    }
}

@Service
public class EmailService {

    @Async("emailExecutor")
    public void sendOrderConfirmation(String customerEmail) {
        System.out.println(Thread.currentThread().getName() + " メール送信先 " + customerEmail);
    }
}

email-1 メール送信先 an.nguyen@example.com

いくつか重要な注意点:

ThreadPoolTaskExecutorのデフォルトqueueCapacityはInteger.MAX_VALUEです。この値を設定しない場合、maxPoolSizeは無意味になります（セクション4で説明したように）。
setWaitForTasksToCompleteOnShutdown(true)により、アプリケーションシャットダウン時に送信中のメールが中断されません。
@Async、Tomcat、@Scheduledでvirtual threadを使用したい場合、設定ファイルに1行追加するだけです:

spring.threads.virtual.enabled=true

7. プールの動作をテストする

以下のテストは、セクション4で説明した正確な処理順序を確認しています: コアスレッド、その後キュー、その後追加スレッド、最後にリジェクション。

import org.junit.jupiter.api.Test;
import java.util.concurrent.*;
import static org.assertj.core.api.Assertions.*;

class PaymentExecutorTest {

    @Test
    void shouldRejectWhenThreadsAndQueueAreFull() throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 2, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.AbortPolicy());

        Runnable blockingPayment = () -> {
            try {
                release.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        try {
            executor.execute(blockingPayment); // 1: コアスレッドを作成
            executor.execute(blockingPayment); // 2: キューに入る
            executor.execute(blockingPayment); // 3: キューがいっぱい -> 2番目のスレッドを作成

            assertThat(executor.getPoolSize()).isEqualTo(2);
            assertThat(executor.getQueue()).hasSize(1);
            assertThatThrownBy(() -> executor.execute(blockingPayment))
                    .isInstanceOf(RejectedExecutionException.class);
        } finally {
            release.countDown();
            executor.shutdown();
            assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
        }
    }
}

CountDownLatchはすべてのタスクをブロック状態に保つため、プール状態の確認時に静止しています。4番目のタスクは場所がなく、AbortPolicyは例外をスローします。finallyブロックはlatchを解放してプールを閉じるため、テストは残骸スレッドを残しません。

8. 一般的なエラー

制御されていないproducerのExecutors.newFixedThreadPoolを使用。 制限なしのキューは膨張してOOMを引き起こします。制限付きのArrayBlockingQueueを使用してください。
リクエストごとに新しいプールを作成。 プールは共有する必要があります（例えばbeanまたはstatic finalフィールド）。作成して忘れると、スレッドがリークします。
例外を飲み込む。 submit()の場合、例外はFutureに保持されます。get()を呼び出さない場合、エラーが見えません。get()を呼び出すか、タスク内のエラーをすぐにキャッチしてください。
同じ小さなプール上でタスクが子タスクを待つことによるデッドロック。 例: 2スレッドプール、両親タスクがキューに入っている子タスクでget()を呼び出す。子タスクを実行するスレッドがなく、システムが停止します。
Virtual threadに重いThreadLocalを設定。 数百万個のvirtual threadが可能で、各々はコピーを保持するため、メモリが急速に増加します。
タスクがinterruptを処理しないshutdownNow()を呼び出す。 その場合、タスクは実行を続けます。長いタスクはThread.currentThread().isInterrupted()をチェックすべきです。

注意: 複数ステップの複雑な処理フローにExecutorServiceを使用しないでください。その場合はCompletableFutureを使用すべきで、自己設定のexecutorで実行できます。

9. 結論

各スレッドプール型は異なる問題を解決します:

Fixed poolまたはForkJoinPool はCPU限定タスク用。
Virtual thread + Semaphore はJava 21でのI/O限定タスク用。
Single thread executor は順序が必要な場合。
Scheduled pool は定期的なタスク用。

本番環境では、ThreadPoolExecutorまたはThreadPoolTaskExecutorを制限付きキュー、明確なrejection policy、識別可能なスレッド名で自分で設定してください。最後に、公式のみに依存するのではなく、ロードテストでプールサイズを検証してください。

ExecutorServiceとスレッドプール: 問題ごとに正しい型を選択する

1. 概要

2. ExecutorService の基本

3. Executorsのプール型

3.1. newFixedThreadPool(n)

3.2. newCachedThreadPool()

3.3. newSingleThreadExecutor()

3.4. newScheduledThreadPool(n)

3.5. newWorkStealingPool()

3.6. newVirtualThreadPerTaskExecutor() (Java 21)