SpringにあるThreadPoolTaskExecutorの勉強記録です。

ExecutorService

通常のExecutorServiceを利用した場合はexecuteの後にshutdownを置いてもエラーになりません

@GetMapping("/sample3")
public String sample3(){

    ExecutorService te = Executors.newFixedThreadPool(3);

    for(int i = 0; i < 3; i++){
        te.execute( () -> {
            for(int y = 0; y < 3; y++){
                try {
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println(
                            Thread.currentThread().getName() + " No:" + y
                    );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    te.shutdown();

    return "終わり３";
}

pool-1-thread-1 No:0

pool-1-thread-3 No:0

pool-1-thread-2 No:0

pool-1-thread-1 No:1

pool-1-thread-3 No:1

pool-1-thread-2 No:1

pool-1-thread-1 No:2

pool-1-thread-3 No:2

pool-1-thread-2 No:2

ExecutorServiceなので非同期処理として画面には即座に「終わり３」が返ってきますが、バックグラウンドで処理が流れています。

ThreadPoolTaskExecutor

次の同じようなコードをThreadPoolTaskExecutorを使って書くと割り込みエラーが発生します

@GetMapping("/sample2")
    public String sample2(){

        ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();
        te.setCorePoolSize(5);
        te.setMaxPoolSize(10);
        te.setQueueCapacity(20);
        te.initialize();

        for (int i = 0; i < 3; i++) {
            te.execute(() -> {
                for (int y = 0; y < 3; y++) {
                    try {
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println(
                                Thread.currentThread().getName() + " No:" + y
                        );
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }

        te.shutdown();

        return "処理が終わり";
    }

java.lang.InterruptedException: sleep interrupted

	at java.base/java.lang.Thread.sleep(Native Method)

	at java.base/java.lang.Thread.sleep(Thread.java:336)

	at java.base/java.util.concurrent.TimeUnit.sleep(TimeUnit.java:446)

	at com.volkruss.Executor.task.SampleController.lambda$sample2$0(SampleController.java:44)

	...(省略)

ThreadPoolTaskExecutor-1 No:1

ThreadPoolTaskExecutor-3 No:1

ThreadPoolTaskExecutor-2 No:1

ThreadPoolTaskExecutor-1 No:2

ThreadPoolTaskExecutor-3 No:2

ThreadPoolTaskExecutor-2 No:2

No:0が実行されていませんでした。

スレッドがスリープ中にシャットダウンしようとしてるからエラーなんでしょうか？No0だけが実行されていません。

setWaitForTasksToCompleteOnShutdown

このsetWaitForTasksToCompleteOnShutdownをtrueにして実行してみます。

※デフォルトではfalseになっている

@GetMapping("/sample2")
    public String sample2(){

        ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();
        te.setCorePoolSize(5);
        te.setMaxPoolSize(10);
        te.setQueueCapacity(20);
        // 追加
        te.setWaitForTasksToCompleteOnShutdown(true);
        te.initialize();

        for (int i = 0; i < 3; i++) {
            te.execute(() -> {
                for (int y = 0; y < 3; y++) {
                    try {
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println(
                                Thread.currentThread().getName() + " No:" + y
                        );
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        te.shutdown();
        return "処理が終わり";
    }

ログを見ると割り込みが発生していませんでした

ThreadPoolTaskExecutor-2 No:0

ThreadPoolTaskExecutor-1 No:0

ThreadPoolTaskExecutor-3 No:0

ThreadPoolTaskExecutor-1 No:1

ThreadPoolTaskExecutor-2 No:1

ThreadPoolTaskExecutor-3 No:1

ThreadPoolTaskExecutor-1 No:2

ThreadPoolTaskExecutor-3 No:2

ThreadPoolTaskExecutor-2 No:2

Bean登録する

今までの実装だと毎回スレッドプールを作成しています。通常はBean登録して利用するのでそのやり方を見ます

まずはBeanとして登録します

@Configuration
public class TaskExecutorConfig {

    @Bean
    ThreadPoolTaskExecutor threadPoolTaskExecutor(){
        ThreadPoolTaskExecutor th = new ThreadPoolTaskExecutor();
        // コアのスレッドプール
        th.setCorePoolSize(5);
        // 最大スレッド数
        th.setMaxPoolSize(10);
        // キューで受け付けるリクエスト数
        th.setQueueCapacity(20);
        return th;
    }
}

Beanとして登録したThreadPoolTaskExecutorを利用するクラスです

@Component
public class ExecutorSample {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public void execute(){
        System.out.println("処理開始");
        for(int i = 0; i < 2; i++){
            threadPoolTaskExecutor.execute( () -> {
                for(int y = 0; y < 3; y++){
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        System.out.println("No : " + y + " Name = " + Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

これを何度か発火させるとスレッドプールを使い回していることが確認できます

処理開始

No : 0 Name = threadPoolTaskExecutor-2

No : 0 Name = threadPoolTaskExecutor-1

No : 1 Name = threadPoolTaskExecutor-2

No : 1 Name = threadPoolTaskExecutor-1

No : 2 Name = threadPoolTaskExecutor-1

No : 2 Name = threadPoolTaskExecutor-2

処理開始

No : 0 Name = threadPoolTaskExecutor-3

No : 0 Name = threadPoolTaskExecutor-4

No : 1 Name = threadPoolTaskExecutor-4

No : 1 Name = threadPoolTaskExecutor-3

No : 2 Name = threadPoolTaskExecutor-4

No : 2 Name = threadPoolTaskExecutor-3

処理開始

No : 0 Name = threadPoolTaskExecutor-1

No : 0 Name = threadPoolTaskExecutor-5

No : 1 Name = threadPoolTaskExecutor-1

No : 1 Name = threadPoolTaskExecutor-5

No : 2 Name = threadPoolTaskExecutor-1

No : 2 Name = threadPoolTaskExecutor-5

これをシャットダウンさせてみます

まずはBean定義を修正します

@Bean
    ThreadPoolTaskExecutor threadPoolTaskExecutor(){
        ThreadPoolTaskExecutor th = new ThreadPoolTaskExecutor();
        // コアのスレッドプール
        th.setCorePoolSize(5);
        // 最大スレッド数
        th.setMaxPoolSize(10);
        // キューで受け付けるリクエスト数
        th.setQueueCapacity(20);
        //　追加
        th.setWaitForTasksToCompleteOnShutdown(true);
        return th;
    }

シャットダウンを明示的に呼び出します

@Component
public class ExecutorSample {

    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    public void execute(){
        System.out.println("処理開始");
        for(int i = 0; i < 2; i++){
            threadPoolTaskExecutor.execute( () -> {
                for(int y = 0; y < 3; y++){
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        System.out.println("No : " + y + " Name = " + Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        //シャットダウン
        threadPoolTaskExecutor.shutdown();
    }
}

2回リクエストを送ると、２回目で当然のごとくエラーになります

処理開始

No : 0 Name = threadPoolTaskExecutor-1

No : 0 Name = threadPoolTaskExecutor-2

No : 1 Name = threadPoolTaskExecutor-1

No : 1 Name = threadPoolTaskExecutor-2

No : 2 Name = threadPoolTaskExecutor-1

No : 2 Name = threadPoolTaskExecutor-2

処理開始

2022-08-02 17:07:12.563 ERROR 9732 --- [nio-8080-exec-2] o.a.c.c.C.[.[.[/].[dispatcherServlet]    : Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@1a9ad32[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2]] did not accept task: com.volkruss.Executor.task.ExecutorSample$$Lambda$641/0x0000000800f9f2d0@5963958] with root cause



java.util.concurrent.RejectedExecutionException: Task com.volkruss.Executor.task.ExecutorSample$$Lambda$641/0x0000000800f9f2d0@5963958 rejected from java.util.concurrent.ThreadPoolExecutor@1a9ad32[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 2]

	at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2057) ~[na:na]

	at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:827) ~[na:na]

	at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1357) ~[na:na]

	at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:360) ~[spring-context-5.3.22.jar:5.3.22]

	at com.volkruss.Executor.task.ExecutorSample.execute(ExecutorSample.java:18) ~[main/:na]

	at com.volkruss.Executor.task.SampleController.sample(SampleController.java:28) ~[main/:na]

	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]

	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:78) ~[na:na]

	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]

	at java.base/java.lang.reflect.Method.invoke(Method.java:567) ~[na:na]

どこでシャットダウンすればいいのか

結局どこでシャットダウンすればいいのか？

ApplicationListenerの実装クラスを作成してApplication終了時にシャットダウンするのが良さそうです。

@Component
public class ApplicationListenerImpl implements ApplicationListener<ContextClosedEvent> {

    @Autowired
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        System.out.println("ThreadPoolTaskExecutorをシャットダウンします");
        threadPoolTaskExecutor.shutdown();
        System.out.println("ThreadPoolTaskExecutorをシャットダウンしました");
    }
}

アプリケーションを明示的に終了させて処理を見てみました

@Autowired
private AbstractApplicationContext abstractApplicationContext;

@GetMapping("/close")
public void end(){
    abstractApplicationContext.close();
}

処理開始

処理開始

No : 0 Name = threadPoolTaskExecutor-2

No : 0 Name = threadPoolTaskExecutor-1

No : 0 Name = threadPoolTaskExecutor-4

No : 0 Name = threadPoolTaskExecutor-3

No : 1 Name = threadPoolTaskExecutor-2

No : 1 Name = threadPoolTaskExecutor-1

No : 1 Name = threadPoolTaskExecutor-4

No : 1 Name = threadPoolTaskExecutor-3

No : 2 Name = threadPoolTaskExecutor-2

No : 2 Name = threadPoolTaskExecutor-1

No : 2 Name = threadPoolTaskExecutor-4

No : 2 Name = threadPoolTaskExecutor-3

ThreadPoolTaskExecutorをシャットダウンします

ThreadPoolTaskExecutorをシャットダウンしました

shutdownNow

ExecutorService#shutdownNowで終了するとThreadPoolTaskExecutorでshutdownした時のような割り込みのエラーが発生します

public String sample3(){

    ExecutorService te = Executors.newFixedThreadPool(3);

    for(int i = 0; i < 3; i++){
        te.execute( () -> {
            for(int y = 0; y < 3; y++){
                try {
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println(
                            Thread.currentThread().getName() + " No:" + y
                    );
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
    te.shutdownNow();

    return "終わり３";
}

java.lang.InterruptedException: sleep interrupted

	at java.base/java.lang.Thread.sleep(Native Method)

	at java.base/java.lang.Thread.sleep(Thread.java:336)

	at java.base/java.util.concurrent.TimeUnit.sleep(TimeUnit.java:446)

	....(省略)...

pool-2-thread-3 No:1

pool-2-thread-2 No:1

pool-2-thread-3 No:2

pool-2-thread-1 No:2

pool-2-thread-2 No:2

となると、ThreadPoolTaskExecutor#shutdownが怪しいので見てみます

public void shutdown() {
	if (logger.isDebugEnabled()) {
		logger.debug("Shutting down ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
	}
	if (this.executor != null) {
		if (this.waitForTasksToCompleteOnShutdown) {
			this.executor.shutdown();
		}
		else {
			for (Runnable remainingTask : this.executor.shutdownNow()) {
				cancelRemainingTask(remainingTask);
			}
		}
		awaitTerminationIfNecessary(this.executor);
	}
}

デフォルトではwaitForTasksToCompleteOnShutdownがfalseなのでshutdownNowがコールされます。

そのためExecutorService#shutdownNowと同じ挙動になっているのでしょう

ThreadPoolTaskExecutorをアプリケーション終了時にシャットダウンさせる