=================================
=================================
=================================
출처: http://ismydream.tistory.com/46
※ 개인적인 스터디를 위해 위 책의 내용을 정리한 내용입니다. 자세한 내용을 확인하고 싶으신 분은 위 책을 구매하셔서 확인하시기 바랍니다. http://www.yes24.com/24/goods/3015162?CategoryNumber=001001003
8. 스레드 풀 활용
8.1 작업과 실행 정책 간의 보이지 않는 연결 관계
일정한 조건을 갖춘 실행 정책이 필요한 작업에는 다음과 같은 것들이 있다.- 의존성이 있는 작업- 스레드 한정 기법을 사용하는 작업- 응답 시간이 민감한 작업- ThreadLocal 을 사용하는 작업
8.1.1 스레드 부족 데드락
8.1.2 오래 실행되는 작업
오래 실행되는 스레드의 갯수가 많을 수록 스레드풀의 응답속도는 느려지게 된다.따라서 계속해서 대기하는 기능 대신 일정시간 동안만 대기하는 메소드를 사용할 수 있다면 응답 속도를 향상시킬 수 있다.
8.2 스레드 풀 크기 조절
스레드 풀의 크기는 설정 파일이나 Runtime.availableProcessors() 값에 따라 동적으로 지정되도록 해야 한다.
8.3 ThreadPoolExecutor 설정
8.3.1 스레드 생성과 제거
public ThreadPoolExecutor(int corePoolSize,
int maxizmumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler){ ... }
corePoolSize 는 스레드 풀을 사용할때 원하는 스레드의 갯수라고 볼 수 있다.queue 가 가득차지 않는 이상 corePoolSize 를 초과하지 않는다.최초에 prestartAllCoreThreads 메소드를 실행하면 코어 크기만큼이 스레드가 미리 만들어진다.
newFixedThreadPool 의 경우 생성시 지정한 크기로 corePoolSize, maximumPoolSize 가 설정된다. 시간 제한은 무제한이다.newCachedThreadPool 의 경우 corePoolSize = 0, maximumPoolSize 는 Integer.MAX_VALUE 로 설정된다. 시간 제한은 1분이다.
8.3.2 큐에 쌓인 작업 관리
newFixedThreadPool 과 newSingleThreadExecutor 는 기본설정으로 크기가 제한되지 않은 LinkedBlockingQueue를 사용한다.크기가 제한된 큐를 사용하면 자원 사용량을 한정시킬 수 있지만 큐가 가득 찼을 때 새로운 작업을 등록하려는 상황을 처리해야 한다.큐의 크기를 제한한 상태에서는 큐의 크기와 스레드의 갯수를 동시에 튜닝해야 한다.
newCachedThreadPool 은 SynchronousQueue 를 사용한다. SynchronousQueue 는 내부적으로 스레드 간에 작업을 넘겨주는 기능을 담당한다.큐의 크기가 0 이므로 스레드의 크기가 최대값에 다다른 상태라면 작업을 거부하게 된다.
PriorityBlockingQueue 를 사용하면 우선 순위에 따라 작업이 실행되도록 할 수 있다.
(자바 6에서는 LinkedBlockingQueue 대신 SynchronousQueue 를 사용할때 자바 5 버전에 비해 세배 이상의 성능을 발휘한다.)
8.3.3 집중 대응 정책
크기가 제한된 큐에 작업이 가득 차면 집중 대응 정책 saturation policy 가 동작한다.집중 대응 정책은 ThreadPoolExecutor 에 setRejectedExecutionHandler 메소드를 사용해 설정할 수 있다.
- 중단 정책(abort) 기본적으로 사용하는 집중 대응 정책이며 execute 메소드에서 RejectedExecutionException을 던진다.- 제거 정책(discard) 큐에 추가하려던 작업을 제거해 버린다.- 호출자 실행 정책(caller runs) 작업을 프로듀서에게 거꾸로 넘겨 작업 추가 속도를 늦출 수 있도록 일종의 속도 조절 방법으로 사용한다. execute 메소드 내부에서 main 스레드가 해당 작업을 직접 처리함으로써 또 다른 새로운 작업이 추가되지 못하도록 한다. 따라서 그 시간동안 스레드는 큐의 작업을 처리할 시간을 가지게 된다.
ThreadPoolExecutor executor = new ThreadPoolExecutor(N_THREADS, N_THREADS, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(CAPACITY));
executor.setRejectedExecutionHandler( new ThreadPoolExecutor.CallerRunsPolicy());
8.3.4 스레드 팩토리
스레드풀에서 새로운 스레드를 생성할 때는 스레드 팩토리를 사용해 새로운 스레드를 생성한다.
public interface ThreadFactory{
Thread newThread(Runnable r);
}
스레드 팩토리에서 MyAppThread 를 생성하여 스레드의 이름, 현재 실행중인 스레드의 갯수 등을 확인할 수 있다.
public class MyAppThread extends Thread{
public static final String DEFAULT_NAME = "MyAppThread";
private static volatile boolean debugLifecycle = false;
private static final AtomicInteger created = new AtomicInteger();
private static final AtomicInteger alive = new AtomicInteger();
private static final Logger log = Logger.getAnonymousLogger();
public MyAppThread(Runnable r){ this(r, DEFAULT_NAME);}
public MyAppThread(Runnable runnable, String name){
super(runnable, name + "=" + created.incrementAndGet());
setUncaughtExceptionHandler(
new Thread.UncaughtExceptionHandler(){
public void uncaughtException(Thread t, Throwable e){
log.log(Level.SEVERE, "UNCAUGHT in thread " + t.getName(), e);
}
}
);
}
public void run(){
boolean debug = debugLifecycle;
if( debug) log.log(Level.FINE, "Created " + getName());
try{
alive.incrementAndGet();
super.run();
}finally{
alive.decrementAndGet();
if(debug) log.log(Level.FINE, "Exiting " + getName());
}
}
public static int getThreadsCreated(){ return created.get();}
public static int getThreadsAlive(){ return alive.get();}
public static boolean getDebug(){ return debugLifecycle;}
public static void setDebug(boolean b){debugLifecycle = b;}
}
8.3.5 ThreadPoolExecutor 생성 이후 설정 변경
Executors 로 생성한 ThreadPool 은 관련된 set 메소드를 사용해서 내부의 설정된 값(코어 스레드 갯수, 최대 스레드 갯수, 스레드 유지 시간, 스레드 팩토리, 작업 거부 처리 정책)을 변경할 수 있다.
Executors 에는 unconfigurableExecutorSevice 를 사용해서 설정을 변경할 수 없는 ThreadPool을 생성할 수도 있다.
8.4 ThreadPoolExecutor 상속
ThreadPoolExecutor 는 beforeExecute, afterExecute, terminated 와 같은 훅 메소드를 제공한다.만약 beforeExecute 메소드에서 RuntimeException 이 발생하면 해당 작업뿐 아니라 afterExecute 메소드도 실행되지 않는다.
처리 순서
beforeExecute
작업
afterExecute
terminatd
8.4.1 예제 : 스레드 풀에 통계 확인 기능 추가
public class TimingThreadPool extends ThreadPoolExecute{
private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();
private final Logger log = Logger.getLogger("TimingThreadPool");
private final AtomicLong numTasks = new AtomicLong();
private final AtomicLong totalTime = new AtomicLong();
protected void beforeExecute(Thread t, Runnable r){
super.beforeExecute(t, r);
log.fine(String.format("Thread %s: start %s", t, r);
startTime.set(System.nanoTime());
}
protected void afterExecute(Runnable r, Throwable t){
try{
long endTime = System.nanoTime();
long taskTime = endTime - startTime.get();
numTask.incrementAndGet();
totalTime.addAndGet(taskTime);
log.fine(String.format("Thread %s: end %s, time=%dns", t, r, taskTime));
}finally{
super.afterExecute(r, t);
}
}
protected void terminated(){
try{
log.info(String.format("Terminated: avg time=%dns", totalTime.get() / numTasks.get()));
}finally{
super.terminated();
}
}
}
8.5 재귀 함수 병렬화
순차적 프로그램을 병렬 프로그램으로 변경
void processSequentially(List<Element> elements){
for( Element e : elements)
process(e);
}
void processInParallel(Executor exec, List<Element> elements){
for( final Element e : elements)
exec.execute(new Runnable(){
public void run(){ process(e);}
});
}
ExecutorService.invokeAll 메소드를 사용하면 한 묶음의 작업을 한꺼번에 등록하고 그 작업들이 모두 종료될 때까지 대기하도록 할 수 있다.
public<T> void sequentialRecursive(List<Node<T>> nodes, Collection<T> results){
for(Node<T> n : nodes){
results.add(n.compute());
sequentialRecursive( n.getChildren(), results);
}
}
public<T> void paralleRecursive(final Executor exec, List<Node<T>> nodes, final Collection<T> results){
for(final Node<T> n : nodes){
exec.execute(new Runnable(){
public void run(){
results.add(n.compute());
}
});
parallelRecursive(exec, n.getChildren(), results);
}
}
public<T> Collection<T> getParallelResults(List<Node<T>> nodes) throws InterruptedException{ ExecutorService exec = Executors.newCachedThreadPool(); Queue<T> resultQueue = new ConcurrentLinkedQueue<T>(); parallelRecursive(exec, nodes, resultQueue); exec.shutdown(); exec.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS); return resultQueue;}
8.5.1 예제 : 퍼즐 프레임웍
=================================
=================================
=================================
'프로그래밍 관련 > 언어들의 코딩들 C++ JAVA C# 등..' 카테고리의 다른 글
| c++ enum보다 향상된 enum class 관련 (0) | 2017.02.20 |
|---|---|
| 프로그래밍 언어 람다식 표현 관련 (0) | 2017.02.17 |
| 자바 병렬 프로그래밍 summary - 14. 동기화 클래스 구현 (0) | 2016.11.08 |
| 프로그래밍 java MD5 변환 (0) | 2016.08.09 |
| [Java] 자바 열거형 Enum 예제 관련 ~! (0) | 2016.06.22 |