자바 멀티스레드 프로그래밍
Executor 프레임워크
BlockingQueue
자바의 생성자 소비자 문제를 해결해줄 수 있는 인터페이스이다. BlockingQueue의 API를 살펴보자
추가
인자를 큐에 추가하는 API이다. 큐가 가득 찼을 때의 로직이 다르다.
- add(e): 큐가 가득 차면
IllegalStateException예외를 던진다. - offer(e): 큐가 가득 차면
false를 반환한다. - put(e): 큐가 가득 차면 공간이 생길 때까지 대기한다.
- offer(e, time, unit): 큐가 가득 차면 입력한 시간 동안 큐가 비워지기를 기다리다가 시간이 초과되면
false를 반환한다.
취득
인자를 큐에서 제거하고 반환하는 API이다. 큐가 비었을 때의 로직이 다르다.
- remove(): 큐가 비어 있으면
NoSuchElementException예외를 던진 다. - poll(): 큐가 비어 있으면
null을 반환한다. - take(): 큐가 비어 있으면 값이 들어올 때 까지 대기한다.
- poll(time, unit): 큐가 비어 있으면 입력한 시간 동안 요소가 준비되기를 기다리다가 시간이 초과되면
null을 반환한다.
확인
제일 앞에 있는 인자를 확인하는 API이다. 취득과는 다르게 인자를 큐에서 제거하지 않는다. 큐가 비었을 때의 로직이 다르다.
- element(): 큐가 비어 있으면
NoSuchElementException예외를 던진다. - peek(): 큐가 비어 있으면
null을 반환한다.
ExecutorService
Executor 인터페이스를 확장한 인터페이스이다. Task를 제출하고 제어하는 기능을 포함한다. ExecutorService 인터페이스의 기본 구현체는 ThreadPoolExecutor 이다. ExecutorService 는 Task를 저장하고, 제어하기 위해 내부적으로 BlockingQueue를 사용한다.
참고로 ThreadPoolExecutor의 구현체들은 스레드를 지연 로딩한다. 인스턴스를 생성하는 시점에 기본 스레드를 모두 다 만들어 두는 것이 아니라, 요청이 오면 스레드를 만든다.
고정 스레드 풀 전략
고정된 스레드 개수를 사용하는 전략이다. 스레드 수가 고정되어서 리소스 사용량이 예측 가능하다. 따라서 안정적인 서비스를 운영할 수 있게된다.
아래 코드와 같이 Executors의 정적 메서드를 사용하면 쉽게 생성 가능하다. 주의해야 할 점은 Executors.newFixedThreadPool() 메서드로 생성한 ExecutorService는 용량 제한이 없는 LinkedBlockingQueue를 사용한다. 따라서 메모리를 많이 차지하거나, Task 처리까지 오래걸릴 수 있다.
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(3);
}
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}캐시드 스레드 풀 전략
유동적으로 스레드를 사용하는 전략이다. 요청이 오면 그 때 마다 즉시 스레드를 만들어서 처리한다. 작업 요청량이 많아지면 스레드 개수를 늘려서 빠르게 처리할 수 있다. 하지만 작업 요청량이 많아지면 리소스 사용량이 많아져서 자칫하면 서비스를 다운시킬 수 있다.
기본 스레드는 0이고, 최대 개수는 21억개까지 만들어질 수 있다. 생성된 스레드는 60초 동안 추가적으로 task를 처리하지 않으면 소멸된다.
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool(3);
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}커스텀 스레드 풀 전략
직접 ThreadPoolExecutor 생성자로 ExecutorService를 만들어서 사용하는 방법이다.
- corePoolSize: 기본 풀 사이즈이다. 스레드 풀에 기본적으로 유지되는 스레드의 개수이다.
- maximumPoolSize: 큐 공간에 task가 꽉차고, 모든 thread도 task를 처리하는 중이면 maximumPoolSize 까지 스레드를 새로 만들어서 처리할 수 있다. 예를 들어, corePoolSize가 2이고 queue의 사이즈가 2인데 task가 4개가 들어와서 core 스레드가 모두 task를 처리하는 중이고, queue도 꽉 찼을 때 maximumPoolSize 까지 새로운 스레드를 만들어서 task를 처리할 수 있다. 만약 maximumPoolSize 까지 새로운 스레드를 만들어서 task를 처리하는 중에 또 새로운 task가 들어오면
RejectedExecutionException을 던진다. - keepAliveTime: corePoolSize 이상으로 스레드가 생성됐을 때 추가적으로 생성된 스레드가 살아있는 시간이다. 마지막 task 처리 이후 keepAliveTime 동안 새로운 task를 처리하지 않으면 소멸된다.
- TimeUnit: keepAliveTime의 시간 단위이다.
- BlockingQueue: task를 저장하는 큐이다.
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}Runnable과 Callable
Runnable
자바 Thread 클래스 생성자의 인자로 넘겨주는 인터페이스이다. Thread의 start 메서드를 호출하면, 스레드는 Runnable의 run을 실행한다. 반환 값은 없고 check 예외를 던지지 못한다.
@FunctionalInterface
public interface Runnable {
/**
* Runs this operation.
*/
void run();
}Callable
Executor 프레임워크에서 사용되는 인터페이스이다. Runnable과 동일하게 Thread가 Callable의 call을 실행한다. 반환 값을 정의해서 사용할 수 있는 제네릭 인터페이스이다. check 예외를 던질 수 있다.
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}Future
Future 는 전달한 작업의 미래 결과를 담고 있다. es.submit() 메서드를 통해 task를 전달하면, task는 queue에 담겨서 차례가 오면 실행된다. future.get() 메서드로 task의 결과 값을 받을 때, task가 실행 완료된 상태이면 즉시 값을 반환하고 아니면 task가 완료될 때 까지 기다렸다가 완료되면 값을 반환 받는다.
public class CallableMainV2 {
public static void main(String[] args) throws ExecutionException,
InterruptedException {
ExecutorService es = Executors.newFixedThreadPool(1);
Future<Integer> future = es.submit(new MyCallable());
Integer num = future.get();
}
static class MyCallable implements Callable<Integer> {
@Override
public Integer call() {
int value = new Random().nextInt(10); log("create value = " + value); log("Callable 완료");
return value;
}
}
}
GracefulShutdown
한국말로 우아한 종료라고 한다. ExecutorService를 종료할 때 queue에 남아있는 task를 모두 다 처리한 뒤에 종료하는 것을 의미한다. ExecutorService의 shutdown() 메서드를 호출하면, 그 뒤로 들어오는 task는 거절하고, queue에 남은 task는 마저 처리한다.
우아한 종료를 하지 않고 바로 종료하는 방법도 있다. shutdownNow() 메서드를 호출하면 된다.
스프링에서 우아한 종료 구현하기
스프링에서 프로세스가 종료될 때 ExecutorService 인스턴스의 queue에 담긴 task를 모두 처리한 뒤에 프로세스가 종료되도록 하려면 아래와 같이 구현하면 된다.
awaitTermination() 메서드는 작업이 모두 끝날 때 까지 입력된 시간 만큼 대기한다. 입력된 시간 안에 작업이 모두 완료되면 true를 반환하고, 작업이 모두 완료되지 못하면 false를 반환한다. 모든 작업이 끝날 때 까지 막연히 기다릴 수 없기 때문에 아래와 같은 과정이 필요하다.
@Component
@Slf4j
public class EventPublisher {
private final ExecutorService es = Executors.newFixedThreadPool(3);
@PreDestroy
public void shutdown() {
es.shutdown();
try {
// 기존 작업들이 끝날 때까지 최대 60초 대기
if (!es.awaitTermination(60, TimeUnit.SECONDS)) {
log.info("Forcing shutdown as tasks did not finish in time...");
es.shutdownNow(); // 실행 중인 작업 강제 종료
}
} catch (InterruptedException e) {
log.error("Interrupted while shutting down", e);
es.shutdownNow(); // 강제 종료
}
log.info("Executor service shutdown complete.");
}
}