[spring] ThreadPoolTaskExecutor

ThreadPoolTaskExecutor

• 멀티스레드 구현
• 스레드 풀을 사용하는 Executor
• org.springframework.scheduling.concurrent 패키지

테스트 

@Configuration 으로 등록한 클레스에 executor @Bean 을 추가 하여구현 할 수도 있으나

@Async 를 이용하여 구현 

package com.curi.log.config;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Configuration
@EnableAsync(proxyTargetClass = true)
@RequiredArgsConstructor
public class SpringAsyncConfig implements AsyncConfigurer {

    @Bean(name = "executor")
    public Executor threadPoolItemTaskExecutor() {
        return getExecutor();
    }

    private Executor getExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(4); //기본 쓰레드 사이즈
        taskExecutor.setMaxPoolSize(4); //최대 쓰레드 사이즈
        taskExecutor.setQueueCapacity(5); //Max쓰레드가 동작하는 경우 대기하는 queue 사이즈
        taskExecutor.setThreadNamePrefix("executor_");
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true); //Queue 에 남아있는 작업이완료 될 때까지 기다림
        taskExecutor.setAwaitTerminationSeconds(60); //shutdown 최대 60초 대기
        taskExecutor.initialize();
        return taskExecutor;
    }
}

 

 

package com.curi.log.service;


import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

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

@Slf4j
@Service
@RequiredArgsConstructor
public class MultiThreadAsyncService {

    private final AsyncTask asyncTask;

    private List<CompletableFuture<Integer>> completableFutures = new ArrayList<>();

    public void indexer() {
        int i = 0;
        while (i < 10) {
            CompletableFuture<Integer> completableFuture = asyncTask.task(i);
            completableFutures.add(completableFuture);
            i++;
        }

        for (CompletableFuture<Integer> future : completableFutures) {
            try {
                future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        log.info("End info ");
    }
}

 

package com.curi.log.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

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

@Slf4j
@Component
public class AsyncTask {

    @Async("executor")
    public CompletableFuture<Integer> task(int i) {
        List<Integer> itemList = new ArrayList<>();

        if (i % 2 == 0) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        log.info("indexer : {}, count : {}", Thread.currentThread().getName(), i);
        return CompletableFuture.supplyAsync(itemList::size);
    }
}

Thread count : 2

음.. 내가 생각한 결과는 이기 아닌데 

 

Thead count : 4