Java : ExecutorService - API使用例

ExecutorService (Java SE 19 & JDK 19) の使用例まとめです。
だいたいのメソッドを網羅済みです。
API仕様のおともにどうぞ。

概要

終了を管理するメソッド、および1つ以上の非同期タスクの進行状況を追跡するFutureを生成できるメソッドを提供するExecutorです。

クラス構成

ExecutorService を使うと、タスクを非同期に実行できます。
また、タスクの実行状況は Future として追跡できます。

ExecutorService を生成するには、Executors ユーティリティクラスを使います。

非同期の処理をしたい場合は、Threadクラスを直接使うより ExecutorService を使うことをおすすめします。

ノート

Java 19 から、ExecutorService は AutoCloseable を実装するようになりました。これにより try-with-resources文でリソースを自動で解放できます。

try (final var executorService = Executors.newSingleThreadExecutor()) {
    final Callable<String> task = () -> {
        try {
            System.out.println("task start");
            TimeUnit.SECONDS.sleep(1);
            return "abcd";
        } finally {
            System.out.println("task end");
        }
    };

    final var future = executorService.submit(task);

    final var ret = future.get();
    System.out.println("future get : " + ret);
}

// 結果
// ↓
//task start
//task end
//future get : abcd

同期(synchronized)の注意点については「synchronizedの多いコードは危険信号」もご参照いただけたら幸いです。

メソッド

boolean awaitTermination (long timeout, TimeUnit unit)

シャットダウン要求後にすべてのタスクが実行を完了していたか、タイム・アウトが発生するか、現在のスレッドで割込みが発生するか、そのいずれかが最初に発生するまでブロックします。

final var executorService = Executors.newSingleThreadExecutor();
try {
    executorService.submit(() -> {
        try {
            System.out.println("task start");
            TimeUnit.SECONDS.sleep(1);
            System.out.println("Run!");

        } catch (InterruptedException e) {
            System.out.println("Interrupted!");
        } finally {
            System.out.println("task end");
        }
    });
} finally {
    // shutdownでは割込み(InterruptedException)は発生しません。
    executorService.shutdown();
}

final var ret = executorService.awaitTermination(2, TimeUnit.SECONDS);
System.out.println("term : " + ret);

// 結果
// ↓
//task start
//Run!
//task end
//term : true

final var executorService = Executors.newSingleThreadExecutor();
try {
    executorService.submit(() -> {
        try {
            System.out.println("task start");
            TimeUnit.SECONDS.sleep(2);
            System.out.println("Run!");

        } catch (InterruptedException e) {
            System.out.println("Interrupted!");
        } finally {
            System.out.println("task end");
        }
    });
} finally {
    executorService.shutdown();
}

// 1秒だけ待ちます。実行中のタスクは2秒スリープしているので、結果としてfalseを返します。
final var ret = executorService.awaitTermination(1, TimeUnit.SECONDS);
System.out.println("term 1 : " + ret);

// さらに2秒待ちます。タスクは完了するのでtrueを返します。
final var ret2 = executorService.awaitTermination(2, TimeUnit.SECONDS);
System.out.println("term 2 : " + ret2);

// 結果
// ↓
//task start
//term 1 : false
//Run!
//task end
//term 2 : true

default void close ()

順序正しくシャットダウンを開始します。以前に送信されたタスクが実行されますが、新規タスクは受け入れられません。

可能であれば try-with-resources文を使うことをおすすめします。

System.out.println("-- start --");

try (final var executorService = Executors.newSingleThreadExecutor()) {
    final Callable<Void> task = () -> {
        try {
            System.out.println("  task start");
            System.out.println("  sleep...");
            TimeUnit.SECONDS.sleep(1);
        } finally {
            System.out.println("  task end");
        }
        return null;
    };

    executorService.submit(task);
    System.out.println("submit end");
}

System.out.println("-- end --");

// 結果
// ↓
//-- start --
//submit end
//  task start
//  sleep...
//  task end
//-- end --

try-with-resources文を使わない例です。

System.out.println("-- start --");

final var executorService = Executors.newSingleThreadExecutor();
try {
    final Callable<Void> task = () -> {
        try {
            System.out.println("  task start");
            System.out.println("  sleep...");
            TimeUnit.SECONDS.sleep(1);
        } finally {
            System.out.println("  task end");
        }
        return null;
    };

    executorService.submit(task);
    System.out.println("submit end");
} finally {
    executorService.close();
}

System.out.println("-- end --");

// 結果
// ↓
//-- start --
//submit end
//  task start
//  sleep...
//  task end
//-- end --

<T> List<Future<T>> invokeAll (Collection<? extends Callable<T>> tasks)

指定されたタスクを実行し、すべて完了すると、ステータスと結果を含むFutureのリストを返します。

final var tasks = List.<Callable<String>>of(
        () -> {
            try {
                System.out.println("task 1 : start");
                TimeUnit.SECONDS.sleep(1);
                return "aaa";
            } finally {
                System.out.println("task 1 : end");
            }
        },
        () -> {
            try {
                System.out.println("task 2 : start");
                TimeUnit.SECONDS.sleep(1);
                return "bbb";
            } finally {
                System.out.println("task 2 : end");
            }
        },
        () -> {
            try {
                System.out.println("task 3 : start");
                TimeUnit.SECONDS.sleep(1);
                return "ccc";
            } finally {
                System.out.println("task 3 : end");
            }
        }
);

try (final var executorService = Executors.newFixedThreadPool(3)) {
    System.out.println("-- invoke start --");

    final var futures = executorService.invokeAll(tasks);

    System.out.println("-- invoke end --");

    for (final var future : futures) {
        System.out.println("get : " + future.get());
    }
}

// 結果
// ↓
//-- invoke start --
//task 1 : start
//task 2 : start
//task 3 : start
//task 1 : end
//task 2 : end
//task 3 : end
//-- invoke end --
//get : aaa
//get : bbb
//get : ccc

<T> List<Future<T>> invokeAll (Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)

指定されたタスクを実行し、すべてが完了するか時間切れになるか、そのいずれかが最初に発生した時点で、ステータスと結果を含むFutureのリストを返します。

final var tasks = List.<Callable<String>>of(
        () -> {
            try {
                System.out.println("task 1 : start");
                TimeUnit.SECONDS.sleep(1);
                return "aaa";
            } catch (InterruptedException e) {
                System.out.println("task 1 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 1 : end");
            }
        }, () -> {
            try {
                System.out.println("task 2 : start");
                TimeUnit.SECONDS.sleep(3);
                return "bbb";
            } catch (InterruptedException e) {
                System.out.println("task 2 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 2 : end");
            }
        }, () -> {
            try {
                System.out.println("task 3 : start");
                TimeUnit.SECONDS.sleep(5);
                return "ccc";
            } catch (InterruptedException e) {
                System.out.println("task 3 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 3 : end");
            }
        });

try (final var executorService = Executors.newFixedThreadPool(3)) {
    System.out.println("-- invoke start --");

    // 4秒で完了するのは、1番目 "aaa" と 2番目の "bbb" タスク。
    final var futures = executorService.invokeAll(tasks, 4, TimeUnit.SECONDS);

    System.out.println("-- invoke end --");

    System.out.println("futures size : " + futures.size());

    for (final var future : futures) {
        if (future.isCancelled()) {
            System.out.println("Cancelled!");
            continue;
        }

        System.out.println("future get : " + future.get());
    }

}

// 結果
// ↓
//-- invoke start --
//task 1 : start
//task 2 : start
//task 3 : start
//task 1 : end
//task 2 : end
//-- invoke end --
//task 3 : Interrupted!
//task 3 : end
//futures size : 3
//future get : aaa
//future get : bbb
//Cancelled!

<T> T invokeAny (Collection<? extends Callable<T>> tasks)

指定されたタスクを実行し、例外をスローせずに正常に完了したタスクが存在する場合は、その結果を返します。

final var tasks = List.<Callable<String>>of(
        () -> {
            System.out.println("task 1 : start");
            TimeUnit.SECONDS.sleep(1);
            System.out.println("task 1 : throw IllegalStateException!");
            throw new IllegalStateException();
        }, () -> {
            try {
                System.out.println("task 2 : start");
                TimeUnit.SECONDS.sleep(2);
                return "bbb";
            } catch (InterruptedException e) {
                System.out.println("task 2 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 2 : end");
            }
        }, () -> {
            try {
                System.out.println("task 3 : start");
                TimeUnit.SECONDS.sleep(3);
                return "ccc";
            } catch (InterruptedException e) {
                System.out.println("task 3 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 3 : end");
            }
        });

try (final var executorService = Executors.newFixedThreadPool(3)) {
    System.out.println("-- invoke start --");

    final var ret = executorService.invokeAny(tasks);

    System.out.println("-- invoke end --");

    System.out.println("task ret : " + ret);

}

// 結果
// ↓
//-- invoke start --
//task 1 : start
//task 2 : start
//task 3 : start
//task 1 : throw IllegalStateException!
//task 2 : end
//-- invoke end --
//task 3 : Interrupted!
//task 3 : end
//task ret : bbb

<T> T invokeAny (Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)

指定されたタスクを実行し、タイム・アウトが経過する前に例外をスローせずに正常に完了したタスクが存在する場合は、その結果を返します。

final var tasks = List.<Callable<String>>of(
        () -> {
            System.out.println("task 1 : start");
            TimeUnit.SECONDS.sleep(1);
            System.out.println("task 1 : throw IllegalStateException!");
            throw new IllegalStateException();
        }, () -> {
            try {
                System.out.println("task 2 : start");
                TimeUnit.SECONDS.sleep(2);
                return "bbb";
            } catch (InterruptedException e) {
                System.out.println("task 2 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 2 : end");
            }
        }, () -> {
            try {
                System.out.println("task 3 : start");
                TimeUnit.SECONDS.sleep(5);
                return "ccc";
            } catch (InterruptedException e) {
                System.out.println("task 3 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 3 : end");
            }
        });

try (final var executorService = Executors.newFixedThreadPool(3)) {
    System.out.println("-- invoke start --");

    final var ret = executorService.invokeAny(tasks, 3, TimeUnit.SECONDS);

    System.out.println("-- invoke end --");

    System.out.println("task ret : " + ret);

} catch (TimeoutException e) {
    System.out.println("Timeout!");
}

// 結果
// ↓
//-- invoke start --
//task 1 : start
//task 2 : start
//task 3 : start
//task 1 : throw IllegalStateException!
//task 2 : end
//-- invoke end --
//task 3 : Interrupted!
//task 3 : end
//task ret : bbb

final var tasks = List.<Callable<String>>of(
        () -> {
            try {
                System.out.println("task 1 : start");
                TimeUnit.SECONDS.sleep(2);
                return "aaa";
            } catch (InterruptedException e) {
                System.out.println("task 1 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 1 : end");
            }
        }, () -> {
            try {
                System.out.println("task 2 : start");
                TimeUnit.SECONDS.sleep(3);
                return "bbb";
            } catch (InterruptedException e) {
                System.out.println("task 2 : Interrupted!");
                return null;
            } finally {
                System.out.println("task 2 : end");
            }
        });

try (final var executorService = Executors.newFixedThreadPool(2)) {
    System.out.println("-- invoke start --");

    final var ret = executorService.invokeAny(tasks, 1, TimeUnit.SECONDS);

    System.out.println("-- invoke end --");

    System.out.println("task ret : " + ret);

} catch (TimeoutException e) {
    System.out.println("Timeout!");
}

// 結果
// ↓
//-- invoke start --
//task 1 : start
//task 2 : start
//task 1 : Interrupted!
//task 2 : Interrupted!
//task 1 : end
//task 2 : end
//Timeout!

boolean isShutdown ()

このexecutorがシャットダウンしていた場合、trueを返します。

try (final var executorService = Executors.newWorkStealingPool()) {
    final var future = executorService.submit(() -> {
        System.out.println("task start");
        return "abcd";
    });

    final var ret = future.get();
    System.out.println("future get : " + ret);

    System.out.println("isShutdown : " + executorService.isShutdown());
    System.out.println("isTerminated : " + executorService.isTerminated());
}

// 結果
// ↓
//task start
//future get : abcd
//isShutdown : false
//isTerminated : false

final var executorService = Executors.newSingleThreadExecutor();
try {
    executorService.submit(() -> {
        try {
            System.out.println("task start");
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            System.out.println("Interrupted!");
        } finally {
            System.out.println("task end");
        }
    });

    TimeUnit.SECONDS.sleep(1);

} finally {

    System.out.println("isShutdown : " + executorService.isShutdown());
    System.out.println("isTerminated : " + executorService.isTerminated());

    executorService.shutdown();
    System.out.println("--- shutdown ---");

    System.out.println("isShutdown : " + executorService.isShutdown());
    System.out.println("isTerminated : " + executorService.isTerminated());
}

final var ret = executorService.awaitTermination(10, TimeUnit.SECONDS);
System.out.println("term : " + ret);

System.out.println("--- awaitTermination ---");

System.out.println("isTerminated : " + executorService.isTerminated());

// 結果
// ↓
//task start
//isShutdown : false
//isTerminated : false
//--- shutdown ---
//isShutdown : true
//isTerminated : false
//task end
//term : true
//--- awaitTermination ---
//isTerminated : true

boolean isTerminated ()

シャットダウンに続いてすべてのタスクが完了していた場合、trueを返します。

このメソッドの使用例は、isShutdown() にまとめて記載しました。
そちらのAPI使用例をご参照ください。

void shutdown ()

順序正しくシャットダウンを開始します。以前に送信されたタスクが実行されますが、新規タスクは受け入れられません。

final var executorService = Executors.newSingleThreadExecutor();
try {
    final Runnable task1 = () -> {
        try {
            System.out.println("task1 start");
            TimeUnit.SECONDS.sleep(2);

        } catch (InterruptedException e) {
            System.out.println("Interrupted!");
        } finally {
            System.out.println("task1 end");
        }
    };

    executorService.submit(task1);
    TimeUnit.SECONDS.sleep(1);

} finally {
    // 実行中のタスクはそのまま実行を継続します。
    executorService.shutdown();
    System.out.println("--- shutdown ---");
}

try {
    // shutdown後は新規のタスクを受け付けません。(RejectedExecutionException発生)
    executorService.submit(() -> System.out.println("task2"));

} catch (RejectedExecutionException e) {
    System.out.println("submit : Reject!");
}

final var ret = executorService.awaitTermination(10, TimeUnit.SECONDS);
System.out.println("term : " + ret);

// 結果
// ↓
//task1 start
//--- shutdown ---
//submit : Reject!
//task1 end
//term : true

List<Runnable> shutdownNow ()

実行中のアクティブなタスクすべての停止を試み、待機中のタスクの処理を停止し、実行を待機していたタスクのリストを返します。

final var executorService = Executors.newSingleThreadExecutor();
try {
    executorService.submit(() -> {
        try {
            System.out.println("task1 : start");
            TimeUnit.SECONDS.sleep(2);
            System.out.println("task1 : run");

        } catch (InterruptedException e) {
            System.out.println("task1 : Interrupted!");
        } finally {
            System.out.println("task1 : end");
        }
    });
    executorService.submit(() -> {
        System.out.println("task2 : run");
    });
    executorService.submit(() -> {
        System.out.println("task3 : run");
    });

    TimeUnit.SECONDS.sleep(1);

} finally {
    System.out.println("--- shutdownNow ---");
    final var notExecutedTasks = executorService.shutdownNow();

    TimeUnit.SECONDS.sleep(1);

    System.out.println("--- not executed tasks ---");
    System.out.println("size : " + notExecutedTasks.size());

    for (final var task : notExecutedTasks) {
        task.run();
    }
}

final var ret = executorService.awaitTermination(10, TimeUnit.SECONDS);
System.out.println("term : " + ret);

// 結果
// ↓
//task1 : start
//--- shutdownNow ---
//task1 : Interrupted!
//task1 : end
//--- not executed tasks ---
//size : 2
//task2 : run
//task3 : run
//term : true

final var executorService = Executors.newSingleThreadExecutor();
try {
    executorService.submit(() -> {
        try {
            System.out.println("task1 : start");
            TimeUnit.SECONDS.sleep(2);
            System.out.println("task1 : run");

        } catch (InterruptedException e) {
            System.out.println("task1 : Interrupted!");
        } finally {
            System.out.println("task1 : end");
        }
    });

    TimeUnit.SECONDS.sleep(1);

} finally {
    System.out.println("--- shutdownNow ---");
    executorService.shutdownNow();
}

TimeUnit.SECONDS.sleep(1);

try {
    // shutdownNow後は新規のタスクを受け付けません。(RejectedExecutionException発生)
    executorService.submit(() -> {
        System.out.println("task2 : run");
    });

} catch (RejectedExecutionException e) {
    System.out.println("submit : Reject!");
}

final var ret = executorService.awaitTermination(10, TimeUnit.SECONDS);
System.out.println("term : " + ret);

// 結果
// ↓
//task1 : start
//--- shutdownNow ---
//task1 : Interrupted!
//task1 : end
//submit : Reject!
//term : true

Future<?> submit (Runnable task)

実行用のRunnableタスクを送信し、そのタスクを表すFutureを返します。

try (final var executorService = Executors.newSingleThreadExecutor()) {
    final Runnable task = () -> {
        try {
            System.out.println("task start");
            TimeUnit.SECONDS.sleep(2);
            System.out.println("task run");

        } catch (InterruptedException e) {
            System.out.println("Interrupted!");
        } finally {
            System.out.println("task end");
        }
    };

    final var future = executorService.submit(task);

    TimeUnit.SECONDS.sleep(1);

    // タスクが終了するまで待機。
    System.out.println("-- future get --");
    future.get();

    System.out.println("future isDone : " + future.isDone());
}

// 結果
// ↓
//task start
//-- future get --
//task run
//task end
//future isDone : true

<T> Future<T> submit (Runnable task, T result)