12. 스레드 풀과 Executor 프레임워크1

[YONGHYNG]

스레드를 직접 사용할 때의 문제점

실무에서 스레드를 직접 생성해서 사용하면 다음과 같은 3가지 문제가 있다.

1. 스레드 생성 시간으로 인한 성능 문제

2. 스레드 관리 문제

3. Runnable 인터페이스의 불편함

4. 스레드 생성 비용으로 인한 성능 문제
   스레드를 사용하려면 먼저 스레드를 생성해야 한다. 그런데 스레드는 다음과 같은 이유로 매우 무겁다.

• 메모리 할당: 각 스레드는 자신만의 호출 스택(call stack)을 가지고 있어야 한다. 이 호출 스택은 스레드가 실행되는 동안 사용하는 메모리 공간이다. 따라서 스레드를 생성할 때는 이 호출 스택을 위한 메모리를 할당해야 한다.
• 운영체제 자원 사용: 스레드를 생성하는 작업은 운영체제 커널 수준에서 이루어지며, 시스템 콜(system call)을 통해 처리된다. 이는 CPU와 메모리 리소스를 소모하는 작업이다.
• 운영체제 스케줄러 설정: 새로운 스레드가 생성되면 운영체제의 스케줄러는 이 스레드를 관리하고 실행 순서를 조정해야 한다. 이는 운영체제의 스케줄링 알고리즘에 따라 추가적인 오버헤드가 발생할 수 있다. 참고로 스레드 하나는 보통 1MB 이상의 메모리를 사용한다.

스레드를 생성하는 작업은 상대적으로 무겁다. 단순히 자바 객체를 하나 생성하는 것과는 비교할 수 없을 정도로 큰 작업이다.

예를 들어서 어떤 작업 하나를 수행할 때 마다 스레드를 각각 생성하고 실행한다면, 스레드의 생성 비용 때문에, 이미 많은 시간이 소모된다. 아주 가벼운 작업이라면, 작업의 실행 시간보다 스레드의 생성 시간이 더 오래 걸릴 수도 있다.

이런 문제를 해결하려면 생성한 스레드를 재사용하는 방법을 고려할 수 있다. 스레드를 재사용하면 처음 생성할 때를 제외하고는 생성을 위한 시간이 들지 않는다. 따라서 스레드가 아주 빠르게 작업을 수행할 수 있다.

2. 스레드 관리 문제

서버의 CPU, 메모리 자원은 한정되어 있기 때문에, 스레드는 무한하게 만들 수 없다. 예를 들어서, 사용자의 주문을 처리하는 서비스라고 가정하자. 그리고 사용자의 주문이 들어올 때 마다 스레드를 만들어서 요청을 처리한다고 가정하겠다. 서비스 마케팅을 위해 선착순 할인 이벤트를 진행한다고 가정해보자. 그러면 사용자가 갑자기 몰려들 수 있다. 평소 동시에 100개 정도의 스레드면 충분했는데, 갑자기 10000개의 스레드가 필요한 상황이 된다면 CPU, 메모리 자원이 버티지 못할 것이다.

이런 문제를 해결하려면 우리 시스템이 버틸 수 있는, 최대 스레드의 수 까지만 스레드를 생성할 수 있게 관리해야 한다. 또한 이런 문제도 있다. 예를 들어 애플리케이션을 종료한다고 가정해보자.

이때 안전한 종료를 위해 실행 중인 스레드가 남은 작업은 모두 수행한 다음에 프로그램을 종료하고 싶다거나, 또는 급하게 종료해야 해서 인터럽트 등의 신호를 주고 스레드를 종료하고 싶다고 가정해보자. 이런 경우에도 스레드가 어딘가에 관리가 되어 있어야한다.

3. Runnable 인터페이스의 불편함

public interface Runnable {
        void run()
}

Runnable 인터페이스는 다음과 같은 이유로 불편하다.

• 반환 값이 없다: run() 메서드는 반환 값을 가지지 않는다. 따라서 실행 결과를 얻기 위해서는 별도의 메커니즘을 사용해야 한다. 쉽게 이야기해서 스레드의 실행 결과를 직접 받을 수 없다.

앞에서 공부한 SumTask 의 예를 생각해보자. 스레드가 실행한 결과를 멤버 변수에 넣어두고, join() 등을 사용해서 스레드가 종료되길 기다린 다음에 멤버 변수에 보관한 값을 받아야 한다.

• 예외 처리: run() 메서드는 체크 예외(checked exception)를 던질 수 없다. 체크 예외의 처리는 메서드 내부에서 처리해야 한다.

이런 문제를 해결하려면 반환 값도 받을 수 있고, 예외도 좀 더 쉽게 처리할 수 있는 방법이 필요하다. 추가로 반환 값 뿐만 아니라 해당 스레드에서 발생한 예외도 받을 수 있다면 더 좋을 것이다.

해결

지금까지 설명한 1번, 2번 문제를 해결하려면 스레드를 생성하고 관리하는 풀(Pool)이 필요하다.

• 스레드를 관리하는 스레드 풀(스레드가 모여서 대기하는 수영장 풀 같은 개념)에 스레드를 미리 필요한 만큼 만들어둔다.
• 스레드는 스레드 풀에서 대기하며 쉰다.
• 작업 요청이 온다.

• 스레드 풀에서 이미 만들어진 스레드를 하나 조회한다.
• 조회한 스레드1로 작업을 처리한다.

• 스레드1은 작업을 완료한다.
• 작업을 완료한 스레드는 종료하는게 아니라, 다시 스레드 풀에 반납한다. 스레드1은 이후에 다시 재사용 될 수 있다.

Executor 프레임워크 소개

자바의 Executor 프레임워크는 멀티스레딩 및 병렬 처리를 쉽게 사용할 수 있도록 돕는 기능의 모음이다. 이 프레임워크는 작업 실행의 관리 및 스레드 풀 관리를 효율적으로 처리해서 개발자가 직접 스레드를 생성하고 관리하는 복잡함을 줄여준다.

Executor 프레임워크의 주요 구성 요소

Executor 인터페이스

package java.util.concurrent;

public interface Executor {
     void execute(Runnable command);
}

• 가장 단순한 작업 실행 인터페이스로, execute(Runnable command) 메서드 하나를 가지고 있다.

ExecutorService 인터페이스 - 주요 메서드

public interface ExecutorService extends Executor, AutoCloseable {
            
            <T> Future<T> submit(Callable<T> task);
            
            @Override
            default void close() {...}
            ...
}

• Executor 인터페이스를 확장해서 작업 제출과 제어 기능을 추가로 제공한다.
• 주요 메서드로는 submit() , close() 가 있다.

로그 출력 유틸리티 만들기

package thread.executor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;

import static util.MyLogger.log;

public abstract class ExecutorUtils {

    public static void printState(ExecutorService executorService) {
        if (executorService instanceof ThreadPoolExecutor poolExecutor) {
            int pool = poolExecutor.getPoolSize();
            int active = poolExecutor.getActiveCount();
            int queuedTasks = poolExecutor.getQueue().size();
            long completedTask = poolExecutor.getCompletedTaskCount();
            log("[pool=" + pool + ", active=" + active + ", queuedTasks=" +
                    queuedTasks + ", completedTasks=" + completedTask + "]");
        } else {
            log(executorService);
        }
    }
}

• pool : 스레드 풀에서 관리되는 스레드의 숫자
• active : 작업을 수행하는 스레드의 숫자
• queuedTasks : 큐에 대기중인 작업의 숫자
• completedTask : 완료된 작업의 숫자

ExecutorService 코드로 시작하기

package thread.executor;

import static util.MyLogger.log;
import static util.ThreadUtils.sleep;

public class RunnableTask implements Runnable {

    private final String name;
    private int sleepMs = 1000;

    public RunnableTask(String name){
        this.name = name;
    }

    public RunnableTask(String name, int sleepMs){
        this.name = name;
        this.sleepMs = sleepMs;
    }

    @Override
    public void run() {
        log(name + " 시작");
        sleep(sleepMs);
        log(name + "완료");
    }
}

• Runnable 인터페이스를 구현한다. 1초의 작업이 걸리는 간단한 작업으로 가정하자.

package thread.executor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static thread.executor.ExecutorUtils.*;
import static util.MyLogger.log;
import static util.ThreadUtils.sleep;

public class ExecutorBasicMain {

    public static void main(String[] args) {
        ExecutorService es = new ThreadPoolExecutor(2, 2, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        log("== 초기 상태 ==");
        printState(es);
        es.execute(new RunnableTask("taskA"));
        es.execute(new RunnableTask("taskB"));
        es.execute(new RunnableTask("taskC"));
        es.execute(new RunnableTask("taskD"));
        log("== 작업 수행 중 ==");
        printState(es);

        sleep(3000);
        log("== 작업 수행 완료 ==");
        printState(es);

        es.close();
        log("== showdown 완료 ==");
        printState(es);
    }
}

• ExecutorService 의 가장 대표적인 구현체는 ThreadPoolExecutor 이다.

ThreadPoolExecutor(ExecutorService) 는 크게 2가지 요소로 구성되어 있다.

• 스레드 풀: 스레드를 관리한다.
• BlockingQueue : 작업을 보관한다. 생산자 소비자 문제를 해결하기 위해 단순한 큐가 아니라,
  BlockingQueue 를 사용한다.

생산자가 es.execute(new RunnableTask("taskA")) 를 호출하면, RunnableTask("taskA") 인스턴스
가 BlockingQueue 에 보관된다.

• 생산자: es.execute(작업) 를 호출하면 내부에서 BlockingQueue 에 작업을 보관한다. main 스레드가 생산자가 된다.
• 소비자: 스레드 풀에 있는 스레드가 소비자이다. 이후에 소비자 중에 하나가 BlockingQueue 에 들어있는 작업을 받아서 처리한다.

ThreadPoolExecutor 생성자
ThreadPoolExecutor 의 생성자는 다음 속성을 사용한다.

• corePoolSize : 스레드 풀에서 관리되는 기본 스레드의 수
• maximumPoolSize : 스레드 풀에서 관리되는 최대 스레드 수
• keepAliveTime , TimeUnit unit : 기본 스레드 수를 초과해서 만들어진 스레드가 생존할 수 있는 대기 시간이다. 이 시간 동안 처리할 작업이 없다면 초과 스레드는 제거된다.
• BlockingQueue workQueue : 작업을 보관할 블로킹 큐

new ThreadPoolExecutor(2,2,0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

• 작업을 보관할 블로킹 큐의 구현체로 LinkedBlockingQueue 를 사용했다. 참고로 이 블로킹 큐는 작업을 무한대로 저장할 수 있다.

실행 결과

12:10:54.451 [ main] == 초기 상태 ==
12:10:54.461 [ main] [pool=0, active=0, queuedTasks=0, completedTasks=0]
12:10:54.461 [ main] == 작업 수행 중 ==
12:10:54.461 [ main] [pool=2, active=2, queuedTasks=2, completedTasks=0]
12:10:54.461 [pool-1-thread-1] taskA 시작
12:10:54.461 [pool-1-thread-2] taskB 시작
12:10:55.467 [pool-1-thread-1] taskA 완료
12:10:55.467 [pool-1-thread-2] taskB 완료
12:10:55.468 [pool-1-thread-1] taskC 시작
12:10:55.468 [pool-1-thread-2] taskD 시작
12:10:56.471 [pool-1-thread-2] taskD 완료
12:10:56.474 [pool-1-thread-1] taskC 완료
12:10:57.465 [ main] == 작업 수행 완료 ==
12:10:57.466 [ main] [pool=2, active=0, queuedTasks=0, completedTasks=4]
12:10:57.468 [ main] == shutdown 완료 ==
12:10:57.469 [ main] [pool=0, active=0, queuedTasks=0, completedTasks=4]

• ThreadPoolExecutor는 처음엔 스레드를 만들지 않다가, 작업이 들어올 때마다 corePoolSize까지 스레드를 생성하고, 그 이후에는 생성하지 않고 기존 스레드를 재사용한다.

작업이 끝난 스레드는 대기 상태로 풀에 남아 있다가 다시 쓰이며, shutdown 시 모두 제거된다.

• 작업이 들어올 때 스레드를 만들고, 다 쓰면 다시 써먹고, 끝나면 정리한다.

  • 필요할 때만 스레드를 만들고, 최대한 재사용해서 비용을 줄이는 구조

작업 요청 → 스레드 생성 → 작업 처리 → 대기 → 재사용
                                  ↓
                              shutdown → 제거

Runnable의 불편함

public interface Runnable {
       void run()
}

• 반환 값이 없다: run() 메서드는 반환 값을 가지지 않는다. 따라서 실행 결과를 얻기 위해서는 별도의 메커니즘을 사용해야 한다. 쉽게 이야기해서 스레드의 실행 결과를 직접 받을 수 없다. 앞에서 공부한 SumTask 의 예를 생각해보자.

스레드가 실행한 결과를 멤버 변수에 넣어두고, join() 등을 사용해서 스레드가 종료되길 기다린 다음에 멤버 변수를 통해 값을 받아야 한다

package thread.executor.future;

import java.util.Random;

import static util.MyLogger.log;
import static util.ThreadUtils.sleep;

public class RunnableMain {
    public static void main(String[] args) throws InterruptedException {

        MyRunnable task = new MyRunnable();
        Thread thread = new Thread(task, "Thread-1");
        thread.start();
        thread.join();
        int result = task.value;
        log("result value = " + result);
    }

    static class MyRunnable implements Runnable {
        int value;

        @Override
        public void run() {
            log("Runnable 시작");
            sleep(2000);
            value = new Random().nextInt(10);
            log("create value = " + value);
            log("Runnable 완료");
        }
    }
}

• run(): 0 ~ 9 사이의 무작위 값을 조회한다. 작업에 2초가 걸린다고 가정한다.

실행 결과

14:22:06.675 [ Thread-1] Runnable 시작
14:22:08.685 [ Thread-1] create value = 1
14:22:08.686 [ Thread-1] Runnable 완료
14:22:08.686 [ main] result value = 1

• 작위 값이므로 숫자의 결과는 다를 수 있다.
• 프로그램이 시작되면 Thread-1 이라는 별도의 스레드를 하나 만든다.
• Thread-1 이 수행하는 MyRunnable 은 무작위 값을 하나 구한 다음에 value 필드에 보관한다.
• 클라이언트인 main 스레드가 이 별도의 스레드에서 만든 무작위 값을 얻어오려면 Thread-1 스레드가 종료될 때까지 기다려야 한다. 그래서 main 스레드는 join() 을 호출해서 대기한다.
• 이후에 main 스레드에서 MyRunnable 인스턴스의 value 필드를 통해 최종 무작위 값을 획득한다.

별도의 스레드에서 만든 무작위 값 하나를 받아오는 과정이 이렇게 복잡하다. 작업 스레드( Thread-1 )는 값을 어딘가에 보관해두어야 하고, 요청 스레드( main )는 작업 스레드의 작업이 끝날 때 까지 join() 을 호출해서 대기한 다음에, 어딘가에 보관된 값을 찾아서 꺼내야 한다.

작업 스레드는 간단히 값을 return 을 통해 반환하고, 요청 스레드는 그 반환 값을 바로 받을 수 있다면 코드가 훨씬 더 간결해질 것이다. 이런 문제를 해결하기 위해 Executor 프레임워크는 Callable 과 Future 라는 인터페이스를 도입했다.