본문 바로가기

네트워크 프로그래밍/Java NIO

[박혜웅] java.util.concurrent.ThreadPoolExecutor


Executor 에 의해서 생성되는 ExecutorService의 하나로 태스크를 처리 할 때, 내부의 스레드 풀을 이용한다.
스레드풀을 생성할 때 Factory Method 패턴을 이용한다.

생성자와 인자는 아래와 같다.
corePoolSize, maximumPoolSize, keepAliveTime + unit 은 setXXXXX()메소드를 이용하여 동적으로 수정이 가능하다.
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) 
아래 내용에서 poolSize 는 내부변수로서 현재의 스레드풀 크기(스레드 수)를 나타낸다.
 
corePoolSize, maximumPoolSize: 아래와 같은 경우에 따라, 새로운 태스크가 execute되면 스레드 생성 여부가 결정된다.
if( poolSize < corePoolSize): idle인 다른 스레드가 있더라도, 새 스레드가 생성된다.
if( poolSize > corePoolSize && poolSize < maximumPoolSize ): 큐가 가득 찼을 때만 새 스레드가 생성된다.

On-demand construction, Creating new Threads: 
기본적으로 새 태스크가 실행될 때, 스레드 풀도 생성된다. 하지만 prestartCoreThread() 또는 prestartAllCoreThreads()를 오버라이딩하여 미리 스레드풀을 생성할 수 있다.
기본적으로 Excutors.defaultThreadFactory()를 이용하여 스레드가 생성되는데, 이 대 스레드는 같은 ThreadGroup을 갖으며, NORM_PRIORITY 중요도를 갖고 daemon 이 아닌 상태로 실행된다. 이를 수정하기 위해 newFixedThreadPool(), newSingleThreadExecutor()등을 실행할 때, 뒤에 ThreadFactory를 인자로 지정할 수 있다. 

KeepAliveTime:
if( poolSize > corePoolSize ): keepAliveTime보다 오랫동안 idle인 스레드는 제거된다.(terminate)
if( poolSize < corePoolSize )인 경우에도 allowCoreThreadTimeOut(boolean)을 통해 스레드를 제거할 수 있다. (하지만 그다지 필요하지 않을 것 같다.)

Queuing:
기본적인 큐는 BlockingQueue를 사용한다.
if( poolSize < corePoolSize ): 새 스레드를 생성한다.
if( poolSize > corePoolSize ): 요청(태스크)를 큐잉한다.
만약 큐잉을 할 수 없고 새 스레드를 생성해서 poolSize가 maximumPoolSize를 넘는 경우, 해당 태스크는 거절(reject)된다. 하지만 BlockingQueue를 생성할 때 기본적으로 큐의 크기를 Integer.MAX_VALUE로 선언하므로, 큐잉을 할 수 없는 경우는 거의 없을 것이다.

Rejected tasks:
새로운 태스크는 Executor가 종료(shut down)되거나, 스레드 또는 큐 제한에 의해 execute(java.lang.Runnable)에서 거절될 수 있다.
거절되는 경우, execute 메소드는 RejectedExecutionhandler.rejectedExeution() 메소드를 호출하며, 이 때 4가지 기정의된 핸들러 정책을 사용할 수 있다.
1. ThreadPoolExecutor.AbortPolicy(기본): RejectedExecutionException 예외를 발생시킵니다.
2. ThreadPoolExecutor.CallerRunsPolicy: execute를 호출한 스레드에서 태스크를 수행합니다.
3. ThreadPoolExecutor.DiscardPolicy:  태스크를 버립니다.
4. ThreadPoolExecutor.DiscardOlestPolicy: executor가 종료된 것이 아니라면, 큐의 가장 앞에 태스크를 넣고 다시 수행합니다.

Hook Methods
beforeExecute() 와 afterExecute()를 오버라이드하여, 각 태스크의 수행 전 후에 특정일을 수행할 수 있습니다. 스레드를 초기화하거나, 통계 정보를 수집, 로그를 추가하는 등의 작업을 할 수 있습니다.
terminated()를 오버라이드 하여, Executor가 종료될 때 특정한 업무를 수행할 수 있습니다.
하지만 이러한 메소드를 수행하다가 예외가 발생하면, 스레드가 실패(fail)되거나 종료될 수 있습니다.

Queue maintenance
getQueue() 메소드로 큐를 모니터링하거나 디버깅할 수 있습니다. 하지만 다른 용도로는 사용하지 마십시오.

Finalization
프로그램에 의해서 참조되지 않거나, 스레드가 없는 풀은 자동적으로 종료됩니다.
사용자가 shutdown()을 호출하는 것을 잊더라도, 참조되지 않는 풀이 재생성되는 것을 막으려면, 사용하지 않는 스레드를 결국 죽도록 해야 합니다. keepAliveTime을 설정하거나, corePoolSize를 0으로 설정하거나 allowCoreThreadTimeOut()을 이용하면 됩니다.