[JAVA] Executors 클래스

Executors 클래스는 직접 스레드를 다루는 것은 번거롭기 때문에 ExecutorService 및 ScheduledExecutorService 인스턴스를 생성하여 반환을 도와주는 팩토리 클래스입니다.

 

Executors 메서드

• newFixedThreadPool(int) : 고정된 개수의 스레드를 사용하여 동작하며, 공유된 무한 큐를 사용하여 작업을 처리하는 스레드 풀을 생성합니다. 파라미터로 통해 생성된 스레드 개수는 항상 활성화되어 작업을 처리합니다. 모든 스레드가 활성화된 상태에서 추가 작업이 있으면 큐에 대기합니다. 실행 중에 오류로 스레드가 종료되면 새로운 스레드가 생성됩니다.
• newFixedThreadPool(int, ThreadFactory) : 새 스레드를 생성할 때 ThreadFactory를 사용하여 생성합니다.
• newWorkStealingPool(int) : 주어진 파라미터 수만큼 병렬성을 지원하는데 필요한 스레드를 유지합니다. 여러 큐를 사용하여 스레드 풀을 생성합니다. 실제 스레드 수는 동적으로 움직입니다.
• newWorkStealingPool() : 사용 가능한 프로세서 모두를 병렬 처리로 만듭니다. 16 코어면 16개의 스레드 풀을 만듭니다.
• newSingleThreadExecutor() : 무제한 큐에서 작업을 처리하는 단일 워커 스레드를 생성합니다. 단일 스레드이기 때문에 작업은 순차적으로 실행되며 언제나 하나의 작업만 실행 중입니다. 실행 중에 오류로 스레드가 종료되면 새로운 스레드가 생성됩니다. 반환된 Executor는 변경되지 않음을 보장합니다.
• newSingleThreadExecutor(ThreadFactory) : 새 스레드를 생성할때 ThreadFactory를 사용하여 생성합니다.
• newCachedThreadPool() : 이전에 생성된 스레드를 재사용하는 스레드 풀을 생성합니다. 사용 가능한 이전 스레드가 없으면 새 스레드를 생성하고 풀에 추가됩니다. 최대 개수는 Integer.MAX_VALUE입니다. 60초 동안 사용되지 않은 스레드는 종료되고 캐시에서 삭제됩니다.
• newCachedThreadPool(ThreadFactory) : 새 스레드를 생성할 때ThreadFactory를 사용하여 생성합니다.
• newSingleScheduledThreadPool() : 주어진 시간 혹은 주기적으로 실행하는 단일 스레드 Executor를 생성합니다. 실행 중에 오류로 스레드가 종료되면 새로운 스레드가 생성됩니다. 작업은 순차적으로 실행되며 언제나 하나의 작업만 실행 중입니다. 반환된 Executor는 변경되지 않음을 보장합니다.
• newSingleScheduledThreadPool(ThreadFactory) : 새 스레드를 생성할 때ThreadFactory를 사용하여 생성합니다.
• newScheduledThreadPool(int) : int만큼 스레드를 생성하여 주어진 시간 혹은 주기적으로 실행하는 스레드 풀을 생성합니다.
• newScheduledThreadPool(int, ThreadFactory) : 새 스레드를 생성할 때ThreadFactory를 사용하여 생성합니다.
• unconfigurableExecutorService(ExecutorService) : SingleThread처럼 캐스트를 사용하여 액세스를 할 수 없습니다. 해당 스레드는 설정을 변환하지 않고 ExecutorService 메서드만 사용할 수 있습니다.
• unconfigurableExecutorService(ScheduledExecutorService) : SingleThread처럼 캐스트를 사용하여 액세스를 할 수 없습니다. 해당 스레드는 설정을 변환하지 않고 ScheduledExecutorService 메서드만 사용할 수 있습니다.
• defaultThreadFactory() : 기본 스레드 팩토리를 반환합니다. 생성되는 모든 스레드를 동일한 ThreadGroup에서 생성합니다. 새로운 스레드는 pool-N-thread-M의 이름으로 만들어지며, 여기서 N은 이 팩토리의 연속 번호이고 M은 이 팩토리에 의해 생성된 스레드의 연속 번호입니다. 우선순위가 작은 값으로 생성되며 비데몬 스레드입니다.
• callable : 해당 메서드는 작업을 실행하고 결과를 반환하는 Callable 객체를 반환합니다.

newFixedThreadPool 예제

ExecutorService es = Executors.newFixedThreadPool(1);
		
Runnable task = () -> System.out.println(Thread.currentThread().getName());	

for(int i = 0; i< 5; i++) {
    es.execute(task);
}

고정된 스레드 풀에 1개의 스레드만 생성했기 때문에 pool-1-thread-1만 5번 출력됩니다.

ExecutorService es = Executors.newFixedThreadPool(2);

Runnable task = () -> System.out.println(Thread.currentThread().getName());	

for(int i = 0; i< 5; i++) {
    es.execute(task);
}

2개의 스레드를 생성하여 실행하면 pool-1-thread-1, pool-1-thread-2 섞여서 5번 출력됩니다.

class CustomThreadFactory implements ThreadFactory {
    int count = 1;

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, "CustomThread-" + count);
        count++;
        return thread;
    }
}

ExecutorService es = Executors.newFixedThreadPool(3, new CustomThreadFactory());

Runnable task = () -> System.out.println(Thread.currentThread().getName());	

for(int i = 0; i< 5; i++) {
    es.execute(task);
}

ThreadFactory를 생성하여서 newFixedThreadPool에 담을 수 있습니다. 그리고 출력을 하게 되면 CustomThread-1,2,3 

3개의 스레드를 확인 할 수 있습니다.

ExecutorService es = Executors.newFixedThreadPool(1);

Runnable exceptionTask = () -> throw new RuntimeException("Exception " + Thread.currentThread().getName());

for(int i = 0; i< 5; i++) {
    es.execute(exceptionTask);
}

Exception이 생기면 새로운 스레드가 생성되기 때문에 Exception pool-1-thread-1 ~ Exception pool-1-thread-5 까지 출력이 됩니다.

 

newWorkStealingPool 예제

ExecutorService es = Executors.newWorkStealingPool(1);
    	
Runnable task = () -> System.out.println(Thread.currentThread().getName());	

for(int i = 0; i < 5; i++) {
    es.execute(task);
}

es.awaitTermination(1, TimeUnit.SECONDS);

병렬처리를 하기 때문에 결과를 확인하기 위해서는 awaitTermination을 이용했습니다. 1 스레드만 유지하기 때문에 ForkJoinPool-1-worker-1만 출력됩니다.

ExecutorService es = Executors.newWorkStealingPool();
    	
Runnable task = () -> System.out.println(Thread.currentThread().getName());	

for(int i = 0; i < 5; i++) {
    es.execute(task);
}

es.awaitTermination(1,TimeUnit.SECONDS);

코어 수만큼 스레드가 증가하기 때문에 ForkJoinPool-1-worker-? 출력됩니다.

 

newSingleThreadExecutor 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ExecutorService es = Executors.newSingleThreadExecutor();

for(int i = 0; i < 2; i++) {			
    es.execute(task);
}

싱글 스레드로 실행되기 때문에 pool-1-thread-1만 출력됩니다.

ExecutorService es2 = Executors.newFixedThreadPool(1);
ThreadPoolExecutor tpe2 = (ThreadPoolExecutor) es2;
System.out.println(tpe2.getCorePoolSize());
ExecutorService es = Executors.newSingleThreadExecutor();
ThreadPoolExecutor tpe = (ThreadPoolExecutor) es; // 에러

newFixedThreadPool(1)과 newSingThreadExecutor 다른 점은 Executor를 변경할 수 없습니다.

 

newCachedThreadPool 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ExecutorService es = Executors.newCachedThreadPool();

for(int i = 0; i < 10; i++) {			
    es.execute(task);
}

class CustomThreadFactory implements ThreadFactory {
    int count = 1;

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, "CustomThread-" + count);
        count++;
        return thread;
    }
}

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ExecutorService es = Executors.newCachedThreadPool(new CustomThreadFactory());

for(int i = 0; i < 10; i++) {			
    es.execute(task);
}

newSingleThreadScheduledExecutor 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor();
ses.schedule(task, 3, TimeUnit.SECONDS); // 3초 뒤에 task를 실행합니다.

class CustomThreadFactory implements ThreadFactory {
    int count = 1;

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, "CustomThread-" + count);
        count++;
        return thread;
    }
}

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor(new CustomThreadFactory());
ses.scheduleAtFixedRate(task, 1, 5, TimeUnit.SECONDS); // 1초 뒤 실행 후 5초 간격으로 task 실행

newScheduledThreadPool 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ScheduledExecutorService ses = Executors.newScheduledThreadPool(3);
ses.schedule(task, 3, TimeUnit.SECONDS); // 3초 뒤에 task 실행

class CustomThreadFactory implements ThreadFactory {
    int count = 1;

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, "CustomThread-" + count);
        count++;
        return thread;
    }
}

Runnable task = () -> System.out.println(Thread.currentThread().getName());
ScheduledExecutorService ses = Executors.newScheduledThreadPool(10, new CustomThreadFactory());
ses.scheduleAtFixedRate(task, 1, 5, TimeUnit.SECONDS); // 1초 뒤 실행 후 5초 간격 task 실행

unconfigurableExecutorService 예제

ExecutorService es = Executors.newFixedThreadPool(1);
ExecutorService unes = Executors.unconfigurableExecutorService(es);
ThreadPoolExecutor tpe = (ThreadPoolExecutor) unes; // 에러 생성

ScheduledExecutorService ses = Executors.newScheduledThreadPool(1);
ScheduledExecutorService unses = Executors.unconfigurableScheduledExecutorService(ses);
ThreadPoolExecutor tpe2 = (ThreadPoolExecutor) unses; // 에러 생성

defaultThreadFactory 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
		
ThreadFactory tf = Executors.defaultThreadFactory();

for(int i = 0; i < 5; i++) {
    Thread t = tf.newThread(task);
    t.start();
}

ExecutorService es = Executors.newFixedThreadPool(5, tf);

for(int i = 0; i < 5; i++) {
    es.execute(task);
}

defaultThreadFactory를 생성하여 ThreadFactory로 새로우 스레드 5개 생성하고 pool로 5개 스레드를 고정을 생성해서 돌리는 로직입니다. pool-1-thread 숫자가 1-10까지 찍힙니다.

 

callable 예제

Runnable task = () -> System.out.println(Thread.currentThread().getName());
Callable<Object> result = Executors.callable(task, "hello"); // task 실행 후 결과를 hello로 반환합니다.
System.out.println(result.call());