[Java] 23. スレッドプール(Threadpool)を使う方法

こんにちは。明月です。

この投稿はJavaでスレッドプール(Threadpool)を使う方法に関する説明です。

以前の投稿でJavaでスレッドを使う方法に関する説明したことがあります。

link - [Java] 22.スレッド(Thread)を使う方法

スレッドは別に制限がなく、生成するたびに生成されます。例えばfor文で1から100まで繰り返しを作ってスレッドを生成すればスレッドは100個まで生成されます。

適切なスレッド個数で並列処理するとかなり早い演算処理をしますが、我々のハードウェアは物理的な容量の限界があるので、無限にスレッドを生成することでプログラムのパフォーマンスが速くなることではないです。つまり容量の限界が届くとスレッドの管理するリソースのせいでメモリやシステムリソースがいっぱいになってシングルスレッド(main threadだけ）より遅くなる結果になります。

Javaではプログラムではなく、システムによってスレッド個数を管理して運用するライブラリがあり、それがスレッドプール(Threadpool)ということです。

スレッドプール(Threadpool)とはプール(pool)の中でスレッドを生成してそのプールのなかでスレッドの個数やリソース、運用メモリなどを管理することです。

または、スレッド個数を管理することもありますが、スレッド再使用率を管理して全般的なパフォーマンスも管理します。プログラム中でスレッドを生成、消滅することでかなり時間がかかります。実はスレッドだけではなく、リソース系オブジェクトは生成や消滅が時間がかかりますね。

でも、スレッドプールではスレッドを先に生成してスレッドを呼ばれたら生成されたオブジェクトに載せて使うことでスレッド生成、消滅時間が守ります。つまり、ただのスレッドを使うことよりスレッドプールを利用してスレッド再使用率は上げることで全般的にシステムパフォーマンスが改善することができます。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Example {
  // スレッドsleep関数(sleep関数のException取り除く用)
  private static void sleep() {
    try {
      // スレッド1秒待機
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  // 実行関数
  public static void main(String[] main) {
    // singleスレッドプール -> スレッドプール中でスレッドが一つだけ動く。
    ExecutorService service = Executors.newSingleThreadExecutor();
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプール中のスレッドがすべて正常終了ならスレッドプールを終了。
    service.shutdown();
  }
}

上はスレッドプールの中で一つのスレッドだけあるスレッドプールです。結果をみればスレッドを何回に呼ばれても順番でスレッドが実行されます。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Example {
  // スレッドsleep関数(sleep関数のException取り除く用)
  private static void sleep() {
    try {
      // スレッド1秒待機
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  // 実行関数
  public static void main(String[] main) {
    // スレッドの個数制限がないスレッドプール
    ExecutorService service = Executors.newCachedThreadPool();
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプール中のスレッドがすべて正常終了ならスレッドプールを終了。
    service.shutdown();
  }
}

上のスレッドはスレッド個数の制限がないスレッドプールです。スレッド個数の制限がないので、別にプールを使わなく、ただのnew Threadでインスタンス生成して使うことと同じ感じなスレッドプールです。ただ、スレッドプールの中でスレッドを実行するので、リソース管理ができることだけかな。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Example {
  // スレッドsleep関数(sleep関数のException取り除く用)
  private static void sleep() {
    try {
      // スレッド1秒待機
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  // 実行関数
  public static void main(String[] main) {
    // スレッドを二つだけ生成したスレッドプール
    ExecutorService service = Executors.newFixedThreadPool(2);
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプールにスレッドを実行する。
    service.execute(() -> {
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 1秒待機
        sleep();
      }
    });
    // スレッドプール中のスレッドがすべて正常終了ならスレッドプールを終了。
    service.shutdown();
  }
}

スレッド二つだけに使うスレッドプールを生成したので、結果をみれば始めにスレッドを2つを使います。後、スレッドの処理が終われば続けてスレッドが実行されることを確認できます。

その以外にnewScheduledThreadPool、newWorkStealingPoolがありますが、よく使わないし、私も使ったことがないので他のプールと差異はよく分かりませんね。

上の例のThreadProolではexecute関数を使います。execute関数にはパラメータがインタフェースRunnableを使うのでラムダ式で作成することができます。

でもスレッドプールの中では様々なスレッドが並列で実行されているので、スレッドのすべての結果を得るためにはスレッド同期化が必要です。Javaではsumbit関数を使ってリターンの値を得られます。

Callableインタフェースはジェネリックタイプでリターンタイプを決め、リターン値を計算できます。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class Example {
  // スレッドsleep関数(sleep関数のException取り除く用)
  private static void sleep() {
    try {
      // スレッド1秒待機
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  // 実行関数
  public static void main(String[] main) {
    // スレッドを二つだけ生成したスレッドプール
    ExecutorService service = Executors.newFixedThreadPool(2);
    // スレッドプールにスレッドを実行する。
    Future<Integer> data1 = service.submit(() -> {
      // 合算結果
      int sum = 0;
      // 0から9まで繰り返し
      for (int i = 0; i < 10; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        // 合算
        sum += i;
        // スレッド１秒待機
        sleep();
      }
      // 結果リターン
      return sum;
    });
    // スレッドプールにスレッドを実行する。
    Future<Integer> data2 = service.submit(() -> {
      // 合算結果
      int sum = 0;
      // 10から19まで繰り返し
      for (int i = 10; i < 20; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        / 合算
        sum += i;
        // スレッド１秒待機
        sleep();
      }
      // 結果リターン
      return sum;
    });
    // スレッドプールにスレッドを実行する。
    Future<Integer> data3 = service.submit(() -> {
      // 合算結果
      int sum = 0;
      // 20から29まで繰り返し
      for (int i = 20; i < 30; i++) {
        // コンソール出力
        System.out.println(Thread.currentThread().getName() + "  " + i);
        / 合算
        sum += i;
        // スレッド１秒待機
        sleep();
      }
      // 結果リターン
      return sum;
    });
    try {
      // 三つ目のスレッド結果をコンソールに出力
      System.out.println(data3.get());
      // dataのスレッドが終わるまで待つ。
      // 二つ目スレッド結果をコンソールに出力
      System.out.println(data2.get());
      // 始めのスレッド結果をコンソールに出力
      System.out.println(data1.get());
    } catch (Throwable e) {
      // エラー発生する時にコンソール出力
      e.printStackTrace();
    }
    // スレッドプール中のスレッドがすべて正常終了ならスレッドプールを終了。
    service.shutdown();
  }
}

実はexecuteやsubmitのどっちを使っても構いません。executeの関数で変数をクロージャ機能を利用して値を共有してもいいです。でもクロージャ機能がソースの中で多いなら可読性が落ちるので使い方法に合わせて使ったほうがよいです。

改めてまとめます。

そしてスレッドプールでスレッドを呼び出せる関数が二つがあります。

スレッドプール終了するために関数は三つがあります。

一般的にはshutdown関数でスレッドプール中でスレッドが終了するとスレッドプールを終了する関数です。

shutdownNowの場合はスレッドプールのスレッドの状況は関係しずに、スレッドプールを終了することです。awaitTerminationは時間の間に終了しなければ強制終了する関数です。

ここまでJavaでスレッドプール(Threadpool)を使う方法に関する説明でした。

ご不明なところや間違いところがあればコメントしてください。