왜 스레드 풀(Executor)이 필요한가?
전통적으로 new Thread()를 직접 사용하는 방식은 여러 문제점을 내포합니다. 매번 새로운 스레드 객체를 생성하는 데 비용이 크게 들고, 생성된 스레드들은 관리되지 않은 '야생 스레드'가 되어 시스템 자원을 무분별하게 소비할 수 있습니다. 특히 스레드가 무제한으로 생성되면 리소스 경쟁이 심화되어 시스템이 다운될 위험이 있습니다. 또한 정기적 실행, 지연 실행, 스레드 중단과 같은 고급 기능을 구현하기 어렵습니다.
스레드 풀을 사용하면 다음과 같은 이점이 있습니다:
- 이미 생성된 스레드를 재사용하여 객체 생성 및 소멸 비용을 줄여 성능을 향상시킵니다.
- 최대 동시 스레드 수를 효과적으로 제어하여 시스템 리소스 사용률을 높이고 리소스 경쟁과 블로킹을 방지합니다.
- 지연 실행, 주기적 실행, 단일 스레드 실행, 동시성 제어 등 다양한 기능을 제공합니다.
Executor 프레임워크 소개
Java 5부터 도입된 java.util.concurrent 패키지의 Executor 프레임워크는 스레드 풀 메커니즘을 기반으로 스레드의 시작, 실행, 종료를 제어합니다. 이를 통해 기존의 Thread.start()보다 더 효율적이고 관리하기 쉬운 동시성 프로그래밍이 가능해졌습니다. 또한, 생성자 내에서 스레드를 시작할 때 발생할 수 있는 'this 탈출(this escape)' 문제를 방지하는 데도 도움이 됩니다. Executor 프레임워크는 스레드 풀, Executor, Executors, ExecutorService, CompletionService, Future, Callable 등의 핵심 요소로 구성됩니다.
Executors 유틸리티 클래스
Executors는 네 가지 주요 정적 팩토리 메서드를 제공하여 다양한 유형의 스레드 풀을 손쉽게 생성할 수 있습니다.
1. newFixedThreadPool(int nThreads)
고정된 수(nThreads)의 스레드를 가진 풀을 생성합니다. 모든 스레드가 활성 상태일 때 새로운 작업은 내부의 무제한 큐에서 대기합니다.
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 20; i++) {
executor.execute(() -> System.out.println(Thread.currentThread().getName()));
}
executor.shutdown();
위 코드는 총 5개의 스레드만 생성하며, 하나의 작업이 끝나면 동일한 스레드가 큐의 다음 작업을 재사용하여 처리합니다.
2. newCachedThreadPool()
필요에 따라 새 스레드를 생성하고, 60초 동안 사용되지 않은 유휴 스레드는 종료 및 캐시에서 제거하는 탄력적인 풀입니다. 주로 수명이 짧은 비동기 작업에 적합합니다.
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 100; i++) {
executor.execute(() -> System.out.println(Thread.currentThread().getName()));
}
executor.shutdown();
3. newScheduledThreadPool(int corePoolSize)
지연 실행 또는 주기적 실행을 지원하는 스레드 풀입니다. Timer 클래스의 대안으로 사용할 수 있습니다.
schedule(): 지연 후 한 번 실행
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
scheduler.schedule(() -> System.out.println("5초 후 실행"),
5000, TimeUnit.MILLISECONDS);
scheduler.shutdown();
scheduleAtFixedRate(): 일정한 주기로 반복 실행 (작업 시작 시간 기준)
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
Runnable task = () -> System.out.println("실행: " + System.currentTimeMillis());
scheduler.scheduleAtFixedRate(task, 2000, 3000, TimeUnit.MILLISECONDS);
scheduleWithFixedDelay(): 이전 작업 종료 후 일정 지연 후 반복 실행
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
Runnable task = () -> {
System.out.println("작업 시작");
try { Thread.sleep(1000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
};
scheduler.scheduleWithFixedDelay(task, 2000, 3000, TimeUnit.MILLISECONDS);
4. newSingleThreadExecutor()
단일 스레드로 작업을 순차적으로 처리하는 풀입니다. 작업은 FIFO, LIFO 또는 우선순위 순서로 실행됩니다.
ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
executor.execute(() -> System.out.println(Thread.currentThread().getName()));
}
executor.shutdown();
ExecutorService 인터페이스
ExecutorService는 Executor 인터페이스를 확장하며, 스레드 풀의 라이프사이클 관리 및 비동기 작업 추적 기능을 제공합니다. 주요 메서드는 다음과 같습니다:
- shutdown(): 새로운 작업 수신을 중단하고, 이미 제출된 모든 작업이 완료된 후 풀을 종료합니다.
- shutdownNow(): 실행 중인 작업을 중단하려 시도하고, 대기 중인 작업 목록을 반환합니다.
- submit(Callable task): 결과를 반환하는 Callable 작업을 제출하고, Future 객체를 반환합니다.
- submit(Runnable task): Runnable 작업을 제출하고, 완료 시 null을 반환하는 Future를 반환합니다.
- invokeAll(Collection tasks): 여러 작업을 동시에 실행하고 모든 작업이 완료되면 Future 리스트를 반환합니다.
- invokeAny(Collection tasks): 여러 작업 중 하나라도 성공적으로 완료되면 해당 결과를 반환하고 나머지 작업은 취소합니다.
ExecutorService의 라이프사이클은 '실행(Running)', '종료 중(Shutting down)', '종료(Terminated)' 세 단계로 구성됩니다. 생성 시 실행 상태이며, shutdown() 호출 후에는 종료 중 상태로 전환되어 새 작업을 거부하지만 기존 작업은 계속 실행합니다. 모든 작업이 완료되면 종료 상태가 됩니다.
Executor로 Runnable 작업 실행
execute() 메서드는 Runnable 작업을 받아 스레드 풀에서 실행합니다.
public class RunnableDemo {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
executor.execute(new SimpleTask());
System.out.println("작업 " + i + " 제출됨");
}
executor.shutdown();
}
}
class SimpleTask implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 실행됨");
}
}
Executor로 Callable 작업 실행
Callable은 Runnable과 달리 결과를 반환하고 예외를 던질 수 있습니다. submit() 메서드를 통해 실행되며, 반환된 Future 객체를 통해 결과를 조회할 수 있습니다. Future.get()은 작업이 완료될 때까지 블로킹됩니다.
public class CallableDemo {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
List futures = new ArrayList<>();
for (int i = 0; i < 5; i++) {
futures.add(executor.submit(new ComputeTask(i)));
}
for (Future<String> future : futures) {
try {
while (!future.isDone()) {
// 완료 대기
}
System.out.println(future.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
executor.shutdown();
}
}
class ComputeTask implements Callable<String> {
private final int id;
public ComputeTask(int id) { this.id = id; }
@Override
public String call() throws Exception {
System.out.println("Callable 실행: " + Thread.currentThread().getName());
return "작업 " + id + " 완료, 스레드: " + Thread.currentThread().getName();
}
}
ThreadPoolExecutor를 통한 커스텀 스레드 풀
ThreadPoolExecutor 클래스를 직접 사용하면 세부 설정(코어 스레드 수, 최대 스레드 수, 유휴 시간, 작업 큐 등)을 지정하여 맞춤형 스레드 풀을 구성할 수 있습니다.
public class CustomPoolDemo {
public static void main(String[] args) {
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(15);
ThreadPoolExecutor pool = new ThreadPoolExecutor(
3, // corePoolSize
6, // maximumPoolSize
100, // keepAliveTime
TimeUnit.MILLISECONDS,
workQueue
);
for (int i = 0; i < 10; i++) {
pool.execute(new WorkerTask());
}
pool.shutdown();
}
}
class WorkerTask implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 작업 중");
try { Thread.sleep(200); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}