ExecutorService

ExecutorService는 재사용이 가능한 ThreadPool로 Executor 인터페이스를 확장하여 Thread의 라이프사이클을 제어합니다. ExecutorService 인터페이스는 종료를 관리하는 방법과 비동기 작업의 진행을 추적하기 위한 실행자를 구현하기 위해 필요한 인터페이스입니다.

기본적으로 Thread를 활용하여 다수의 작업을 비동기적으로 수행한다는 것은 매우 어렵습니다. 왜냐하면 우선순위의 문제, 생성과 제거에 대한 핸들링과 같은 문제를 직접 관여하는 것이 굉장히 어렵고 불편한 작업이기 때문입니다. 하지만 이러한 부분을 신경 쓰지않고 편하게 추상화한 것이 ExecutorService입니다.

ExecutorService에 Task를 지정해주면 가진 ThreadPool을 활용하여 Task를 실행하고 이 Task는 Blocking Queue로 관리되기 때문에 Thread보다 많은 수의 Task가 들어오더라도 사용이 가능한 Thread가 생기는 시점까지 기다립니다.

ExecutorService를 생성하는 방법은 2가지가 있습니다. new 키워드를 사용하여 생성하는 방법과 Executors 클래스에서 제공하는 Factory Method를 사용하는 방법이 있습니다.

 

new를 이용하여 생성하는 방법

ExecutorService를 구현한 구현체 중 하나를 new 키워드로 초기화하여 생성합니다. 대표적으로 ThreadPoolExecutor 클래스를 초기화하여 리턴하는 경우가 있습니다.

ExecutorService executorService1 = new ThreadPoolExecutor(10, 10, 1000L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()):

첫번째 인자는 corePoolSize를 의미하며 기본적으로 관리하는 Thread의 숫자를 지정합니다.

두번째 인자는 maximumPoolSize를 의미하며 corePoolsize를 초과하여 최대한으로 만들 수 있는 Thread의 숫자를 의미합니다.

세번째 인자와 네번째 인자는 keepAliveTime과 TimeUnit으로 corePoolSize를 초과하여 생성된 Thread가 작업을 대기할 시간을 지정할 수 있습니다. 이 시간을 초과하면 corePoolSize를 초과한 Thread들을 정리합니다.

다섯번째 인자는 workQueue로 Task를 보관하는 Blocking Queue를 의미합니다.

여기서 주의할점은 corePoolSize에 모든 Thread가 사용중이라고해서 바로 maximumPoolSize만큼 Thread를 생성하는 것이 아니라 workQueue에 지정한 사이즈 이상의 Task가 쌓일 때, maximumPoolSize만큼 Thread를 늘리기 시작합니다.

 

Executors를 이용하는 방법

Executors 클래스에는 newSingleThreadExecutor(), newFixedThreadPool(), newCachedThreadPool()을 비롯한 다양한 Factory Method를 통해서 ExecutorService의 구현체를 생성할 수 있도록 도와줍니다. (newSingleThreadScheduledExecutor(), newScheduledThreadPool(), newWorkStealingPool()은 추후에 따로 다루도록 하겠습니다.)

 

newSingleThreadExecutor()

SingleThread를 기반으로 하는 ExecutorService를 생성합니다. 내부적으로 보면 corePoolSize와 maximumPoolSize와 모든 값이 최소로 지정되어 있음을 알 수 있습니다. 인자로 ThreadFactory를 넘겨주어 Thread 생성에 직접 관여할 수 있습니다.

ExecutorService singleThread = Executors.newSingleThreadExecutor();

 

newFixedThreadPool()

말 그대로 인자로 전달하는 n개의 Thread를 기반으로 동작하는 ExecutorService를 생성합니다. corePoolSize와 maxumumPoolSize가 인자로 넘은 n개 만큼 고정되고, keepAliveTime이 0 mills로 고정됩니다. 마찬가지로 ThreadFactory를 넘겨줄 수 있습니다.

ExecutorService fixedThread = Executors.newFixedThreadPool(5);

 

newCachedThreadPool()

corePoolSize가 0개이고 maximumPoolSize가 Integer.MAX_VALUE입니다. 필요할 때마다 필요한 만큼 Thread를 생성합니다. 이미 생성된 Thread를 재활용 할 수 있고 이론상 21억개까지 생성될 수 있습니다. 마찬가지로 ThreadFactory도 인자로 넘길 수 있습니다.

ExecutorService cachedThread = Executors.newCachedThreadPool();

 

Function

execute(Runnable command)

Executor 인터페이스를 상속하고 있기 때문에 execute를 사용할 수 있습니다. 인자로 받은 Runnable 객체를 단순히 실행시키는 함수입니다. Return 값이 없는 void형 함수로 Task의 실행 결과나 Task의 상태를 알 수 없습니다.

 

submit(Callable<T> task)

인자로 받은 Callable 객체인 task를 를 실행시키고 Return으로 T 타입의 Future를 발행합니다. Future는 비동기적 처리에 대한 값을 받을 수 있는 함수형 인터페이스입니다.

 

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

인자로 Callable<T> Collections을 넘겨받고, 실행하여 성공적으로 완료된 최초의 값을 Task의 결과를 반환합니다. 정상 또는 예외적으로 반환되는 경우 완료되지 않은 작업은 취소됩니다. 이 작업이 진행되는 중 지정된 Collection이 수정되면 이 함수의 결과가 정의되지 않습니다.

 

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

인자로 Callable<T> Collections을 넘겨받고, 실행하는 것은 invokeAny와 동일하지만 모든 Task의 리턴값을 List<Future<T>> 타입으로 반환합니다. invokeAny와 마찬가지로 동작중에 Collection의 변경이 일어나면 결과를 보장할 수 없으며 Future를 Collection의 순서대로 List로 반환하고 전부 끝나는 시점까지 Holding(Blocking) 됩니다. 또 주의할점으로 Collection에 들어있는 Task의 중 예외가 발생하여도 완료된 것으로 간주합니다. 한마리도 완료된 결과 Future에서 get()을 할 때, Exception이 발생합니다. timeOut을 설정하여 취소되는 경우에는 isCancelled()의 값이 true로 반환되며 Exception에 대한 isCancelled는 false가 반환됩니다.

 

shutdown()

실행한 Task가 모두 수행되어도 ExecutorService는 자동으로 종료되지 않고 wait 상태로 대기하게 됩니다. 이때 shutdown() 함수를 이용하여 실행중인 모든 Task가 수행되면 종료되도록하고 새로운 작업을 받지 않습니다.

 

shutdownNow()

이 함수는 실행중인 모든 Thread를 즉시 종료시키도록 하지만 모든 Thread가 동시에 종료되는 것을 보장할 수 없습니다.

 

awaitTermination(long timeout, TimeUnit unit)

이 함수는 shutdown이 불리고 난 뒤 주어진 timeout 내에 작업을 모두 완료했으면 true, 그렇지 않으면 작업중인 Thread를 interrupt시키고 false를 반환합니다.

 

isShutdown()

종료 요청이 호출된 다음 true를 반환합니다.

 

isTerminated()

작업이 완료되었는지를 체크합니다. 완료 되었다면 true를 반환합니다.

 

GitHub - dev-donghwan/donghwan-study: donghwan-personal-study-repository