CompletableFutureのスレッド数について

はじめに

CompletableFuture調べててスレッド数について調べてたら設定をしないと立てるスレッド数について上限があるようなので、設定を探して上限突破を行いました。

とりあえずCompletableFutureのリストを処理させてみる

とりあえず何も考えずにリスト作って処理させてみます。
やらせていることがしょっぱいのは目をつむります。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class Main {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        List<CompletableFuture<String>> list = new ArrayList<>();
        for(int i = 0; i < 100; i++) {
            list.add(CompletableFuture.supplyAsync(() -> {
                try {
                    System.out.println(Thread.currentThread().getName());
                    Thread.sleep(1000);
                } catch (InterruptedException e) {

                }
                return "a";
            }));
        }

        CompletableFuture<Void> cf = CompletableFuture.allOf(
                list.toArray(new CompletableFuture[list.size()]));

        cf.whenComplete((ret, ex) -> {
            System.out.println(ret);
            System.out.println(ex);
          });

        Thread.sleep(20000);
        System.out.println("end");
    }
}

これを実行させてみると分かるのですが、3スレッド(4コアPC)までしか起動されていません。
これではコードでは100とか起動させたいのに思った通りの動きになりません。

ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-3
ForkJoinPool.commonPool-worker-1
ForkJoinPool.commonPool-worker-2
null
null
end

スレッド数を増やしてみる

ということで起動可能なスレッド数を増やします。
supplyAsyncはExecutorを引数に取ることが出来、java.util.concurrent.Executorsで色々な設定をすることが可能となっています。

まずはExecutors.newFixedThreadPool(100)でばりばりっと欲しい数のタスクを作れるようにして動かしてみます。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService pool = Executors.newFixedThreadPool(100);

        List<CompletableFuture<String>> list = new ArrayList<>();
        for(int i = 0; i < 100; i++) {
            list.add(CompletableFuture.supplyAsync(() -> {
                try {
                    System.out.println(Thread.currentThread().getName());
                    Thread.sleep(10000);
                } catch (InterruptedException e) {

                }
                return "a";
            }, pool));
        }

        CompletableFuture<Void> cf = CompletableFuture.allOf(
                list.toArray(new CompletableFuture[list.size()]));

        cf.whenComplete((ret, ex) -> {
            System.out.println(ret);
            System.out.println(ex);
          });

        Thread.sleep(20000);
        System.out.println("end");
    }
}

これを実行すると100スレッド起動していることが分かります。

pool-1-thread-1
pool-1-thread-4
pool-1-thread-3
pool-1-thread-2
pool-1-thread-6
pool-1-thread-5
pool-1-thread-7
pool-1-thread-10
pool-1-thread-11
pool-1-thread-12
pool-1-thread-13
pool-1-thread-15
pool-1-thread-14
pool-1-thread-9
pool-1-thread-16
pool-1-thread-17
pool-1-thread-18
pool-1-thread-20
pool-1-thread-19
pool-1-thread-8
pool-1-thread-22
pool-1-thread-23
pool-1-thread-25
pool-1-thread-21
pool-1-thread-24
pool-1-thread-26
pool-1-thread-27
pool-1-thread-29
pool-1-thread-28
pool-1-thread-30
pool-1-thread-31
pool-1-thread-32
pool-1-thread-34
pool-1-thread-33
pool-1-thread-36
pool-1-thread-37
pool-1-thread-35
pool-1-thread-38
pool-1-thread-40
pool-1-thread-39
pool-1-thread-41
pool-1-thread-42
pool-1-thread-43
pool-1-thread-64
pool-1-thread-60
pool-1-thread-57
pool-1-thread-62
pool-1-thread-59
pool-1-thread-58
pool-1-thread-65
pool-1-thread-71
pool-1-thread-63
pool-1-thread-69
pool-1-thread-70
pool-1-thread-61
pool-1-thread-52
pool-1-thread-53
pool-1-thread-51
pool-1-thread-50
pool-1-thread-49
pool-1-thread-47
pool-1-thread-45
pool-1-thread-46
pool-1-thread-44
pool-1-thread-54
pool-1-thread-55
pool-1-thread-56
pool-1-thread-68
pool-1-thread-66
pool-1-thread-48
pool-1-thread-72
pool-1-thread-73
pool-1-thread-75
pool-1-thread-76
pool-1-thread-74
pool-1-thread-77
pool-1-thread-78
pool-1-thread-80
pool-1-thread-79
pool-1-thread-82
pool-1-thread-81
pool-1-thread-67
pool-1-thread-83
pool-1-thread-84
pool-1-thread-85
pool-1-thread-86
pool-1-thread-87
pool-1-thread-89
pool-1-thread-88
pool-1-thread-90
pool-1-thread-91
pool-1-thread-92
pool-1-thread-94
pool-1-thread-93
pool-1-thread-95
pool-1-thread-96
pool-1-thread-98
pool-1-thread-97
pool-1-thread-99
pool-1-thread-100
null
null
end

他のExecutorsで作れる処理について

今回はとりあえず100スレッド起動させたかっただけなのでnewFixedThreadPoolで指定しましたが、使いまわしたい場合はnewCachedThreadPoolを使ったり(上のコードからsleepとると使いまわされていることが簡単に分かります)、実はスレッドは1つ固定したいとかならnewSingleThreadExecutorがあったりするので、用途に合わせてスレッドの起動の制御が可能となっています。
これが正解というのは使う用途で違うため、使いながら勘所をつかむ必要がありそうです。

ただ、ExecutorsにはnewScheduledThreadPoolと遅延させるExecutorも作れますが、これを設定すると遅延0秒扱いなのであまり意味はなさそうです。
動きを見る限りはnewCachedThreadPoolと似た動きになる感じです。
もしかしたらうまく遅延させることが出来るかもしれませんが、良く分かってません。