JAVA ThreadPoolExecutor을 이용한 스레드 풀(Thread Pool)

스레드 풀

스레드 풀(Thread Pool)은 스레드를 미리 생성하고 관리하는 기법입니다. 

병렬 작업 처리가 많아지면 스레드의 개수가 증가하게 되고 그에 따른 스레드 생성과 스케줄링으로 인해 어플리케이션 성능이 저하가 됩니다. 스레드 풀을 사용하면 스레드 생성 및 제거에 따른 오버헤드를 줄이고, 스레드의 재사용성을 높여 성능을 향상시킬 수 있습니다. 일반적으로 스레드 풀은 고정된 크기의 스레드 집합을 가지며, 작업을 수행하기 위해 해당 스레드를 사용합니다.

 

스레드 풀을 생성하고 관리하며 스레드를 처리하기 위해 Java에서는 `Executor` 인터페이스를 제공해줍니다.

개발자는 Runnable을 이용하여 작성만 해주면 스레드를 생성해서 작업을 처리하고, 처리가 완료되면 스레드를 제거하고 종료하는 작업을  `Executor` 인터페이스가 해줍니다. 즉 개발자는 `Executor`를 사용하여 스레드 작업을 간편하게 실행할 수 있습니다. 예시로, 작업을 `Executor` 인터페이스의 `execute()` 메서드로 제출하면 스레드 풀 내에서 작업이 실행됩니다. 

 

`Executor` 인터페이스를 확장한 서브인터페이스로 `ExecutorService`가 있습니다. 추가된 기능으로는 스레드 풀을 생성하고 관리하는 기능, 작업 완료 여부 확인, 작업 취소 등과 같은 기능이 있습니다. `Executor` 인터페이스는 작업 실행을 담당하였지만, 작업의 결과를 반환하지 않았습니다. `ExecutorService`는 `submit()` 메서드를 통해 작업을 제출하고, 작업의 결과를 나타내는 `Future`객체를 반환합니다.

 

`ThreadPoolExecutor` 클래스는 `ExecutorService` 인터페이스를 구현한 클래스입니다. `ThreadPoolExecutor`는 스레드 풀의 생성과 관리, 작업 스케줄링을 담당하고, 다양한 설정 옵션을 사용하여 스레드 풀의 동작을 조정할 수 있습니다.

---

ThreadPoolExecutor

정리하자면 `ThreadPoolExecutor`는 스레드를 효율적으로 재사용하여 작업을 처리하는 데 사용되는 스레드 풀을 관리하기 위한 클래스입니다. ThreadPoolExecutor는 ExecutorService 인터페이스를 구현하며, 작업을 제출하고 스레드 풀에서 실행할 수 있도록 합니다.

ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 
		keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue);

ThreadPoolExecutor 매개변수

• corePoolSize: 생성할 개수
  스레드 풀의 스레드 수 입니다. 핵심 스레드는 대기 중인 작업이 있을 때 스레드 풀에 계속해서 유지됩니다.
• maximumPoolSize: 생성할 최대 개수
  스레드 풀이 허용하는 최대 스레드 수입니다. 대기 중인 작업이 많아지면 최대 스레드 수까지 스레드를 동적으로 생성할 수 있습니다.
• keepAliveTime: 유지 시간
  핵심 스레드 이외의 추가 스레드가 유휴 상태 유지시간 입니다. 유휴 시간이 경과하면 추가 스레드는 종료됩니다.
• unit
  keepAliveTime의 시간 단위
• workQueue: 작업 큐
  대기 중인 작업을 유지하는 작업 큐를 지정합니다. 이 작업 큐는 ThreadPoolExecutor가 처리해야 할 작업을 보관합니다.
• threadFactory 
  새로운 스레드를 생성하는 데 사용되는 스레드 팩토리를 지정합니다.
• handler 
  작업 큐가 가득 차거나 실행할 수 없는 경우 어떻게 처리할지를 지정하는 RejectedExecutionHandler입니다.

새로운 작업이 발생하면 `corePoolSize`보다 적은 Thread가 수행되고 있는 경우 Thread를 즉시 실행합니다. `corePoolSize`보다 많은 Thread가 수행되고 있을 때 작업이 발생한 경우 즉시 실행하지 않고 `작업 큐(workQueue)`에 넣습니다. `corePoolSize` 이내의 스레드가 모두 사용중이고 `workQueue`가 가득차게 되면, 스레드 풀의 최대 크기가 `maximumPoolSize` 만큼 커지게하고 커진만큼 스레드가 추가로 생성하여 처리합니다. 이렇게 추가로 생성된 스레드는 `keepAliveTime` 동안 할 일이 없으면 자동으로 제거됩니다.

 

즉, `corePoolSize` 만큼의 스레드가 모두 사용 중인 경우, `maximumPoolSize`만큼 스레드 풀이 확장되는 것보다 `workQueue`에 작업이 채워지는 작업이 우선시 됩니다.

 

---

ThreadPoolExecutor 메서드

스레드 풀에 작업을 요청하는 방식은 다음 메서드들을 이용하면 됩니다.

• execute(Runnable task)
  작업 처리 중에 예외가 발생하면 해당 스레드가 종료되고 스레드 풀에서 제거한 뒤, 새로운 스레드를 생성하여 다른 작업을 처리합니다. 
• submit(Callable<T> task)
  작업 처리 중에 예외가 발생하더라도 스레드가 종료되지 않고 다음 작업에 사용됩니다. 또한 처리 결과를 Future<?>로 반환합니다.
• shutdown()
  스레드 풀을 종료합니다. 대기 중인 작업은 완료됩니다.
• shutdownNow()
  스레드 풀을 즉시 종료하며, 실행 중인 작업도 취소합니다.

 

---

`execute` VS `submit`

execute() 메서드와 submit() 메서드는 모두 Java의 ThreadPoolExecutor에서 작업을 제출하는 메서드이지만 몇 가지 용도에 따른 차이점이 있습니다.

• 반환값
  
  • `execute()`: void를 반환합니다. 작업이 스레드 풀에 제출되면 바로 실행되고 작업 처리 중 예외 발생시 스레드가 종료되고 스레드 풀에서 제거한 뒤, 새로운 스레드를 생성하여 다른 작업을 처리합니다.
  • `submit()`: `Future` 객체를 반환합니다. 작업이 스레드 풀에 제출되어 실행되고 작업 처리 중 예외 발생시 스레드가 종료되지 않고 다음 작업에 사용됩니다. 처리 결과가 `Future` 객체로 반환되어 작업의 상태와 결과를 추적할 수 있습니다. 

• 예외 처리
  
  • `execute()`: 작업이 실행되는 동안 발생하는 예외를 메인 스레드로 전파하지 않습니다. 작업이 예외를 발생시키면 스레드 풀 내에서 처리됩니다.
  • `submit()`: 작업이 실행되는 동안 발생하는 예외를 `Future` 객체의 `get()` 메서드 호출 시 예외로 던집니다. 작업의 결과를 얻을 때 예외 처리를 위해 `try-catch` 블록을 사용할 수 있습니다.

• 인터페이스
  
  • `execute()`: `Runnable` 인터페이스를 구현한 작업 객체를 받습니다.
  • `submit()`: `Runnable` 또는 `Callable` 인터페이스를 구현한 작업 객체를 받습니다. `submit()` 메서드는 반환값을 가질 수 있는 `Callable` 작업을 실행하는 데 더 적합합니다.

• 활용
  
  • `execute()`: 작업을 제출하고 실행 결과를 받지 않아도 되는 경우에 사용됩니다. 예를 들어, 단순한 로그 메시지 출력이나 비동기적인 작업 등에 적합합니다.
  • `submit()`: 작업의 결과가 필요하거나 작업이 예외를 발생시킬 수 있는 경우에 사용됩니다. 예를 들어, 계산 작업의 결과를 얻거나 데이터베이스에서 데이터를 조회하는 작업 등에 적합합니다.

 

요약하면, `execute()` 메서드는 간단한 작업을 제출하고 실행 결과를 신경쓰지 않을 때 사용하며, `submit()` 메서드는 작업의 결과를 추적하고 예외 처리를 위해 `Future` 객체를 반환받을 때 사용합니다. `submit()` 메서드는 `Callable` 작업과 `Runnable` 작업을 모두 처리할 수 있습니다.

 

`Runnable` VS `Callable`

Java에서 스레드 풀에서 작업을 처리하기 위해 `Runnable` 또는 `Callable` 객체를사용합니다. `Runnable`은 가장 기본적인 형태의 작업을 표현하고, `Callable`은 작업의 결과를 반환하면서 추가적인 기능을 제공합니다. 

Runnable

• `Runnable`은 `java.lang.Runnable` 인터페이스를 구현한 객체로, 스레드에서 실행되는 작업을 표현합니다.
• `Runnable`은 매개변수가 없고 반환값도 없는 작업을 수행합니다.
• `Runnable`은 `run()` 메서드를 구현해야 하며, 이 메서드 안에 작업의 로직을 구현합니다.
  예를 들면, 파일을 읽고 처리하는 작업이나 네트워크 연결을 수행하는 작업 등이 있습니다.
• `Runnable`은 스레드 풀에서 사용되는 작업의 형태를 나타내는 가장 기본적인 인터페이스입니다.

 

Callable

• `Callable`은 `java.util.concurrent.Callable` 인터페이스를 구현한 객체로, 스레드에서 실행되는 작업을 표현합니다.
• `Callable`은 매개변수가 없지만 반환값이 있는 작업을 수행합니다.
• `Callable`은 `call()` 메서드를 구현해야 하며, 이 메서드 안에 작업의 로직을 구현합니다.
  `call()` 메서드는 작업의 결과를 반환하며, 작업이 완료될 때까지 대기하는 기능도 제공합니다.
  예를 들면, 계산 작업이나 데이터베이스에서 데이터를 조회하는 작업 등이 있습니다.

---

execute() 예제

아래는 `execute()` 메서드를 사용해서 작업을 스레드 풀에 제출한 코드입니다. `execute()`는 작업을 비동기적으로 실행하고, 작업이 완료될 때 까지 기다리지 않습니다.

import java.util.concurrent.*;

public class ThreadPoolExecutorExample {
    public static void main(String[] args) {
        // ThreadPoolExecutor 생성
        int corePoolSize = 2;
        int maxPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue);

        // 작업 제출
        for (int i = 1; i <= 10; i++) {
            executor.execute(() -> {
          		System.out.println("Before_Task: " + i + ", ThreadName: " + Thread.currentThread().getName());
                int result = i * 5;
                System.out.println("After_Task: " + i + ", ThreadName: " + Thread.currentThread().getName());
            });
        }

        // 스레드 풀 종료
        executor.shutdown();
    }
}

 

submit() 예제

아래는 `submit()` 메서드를 사용하여 작업을 제출하고 작업 결과를 처리하는 예제입니다.

`submit()` 메서드는 `Callable` 객체를 매개변수로 받아 작업을 제출하고, 작업에 대한 결과를 `Future<Integer>`를 통해 얻을 수 있습니다. `future.get()` 메서드를 통해 작업이 완료된 후 작업에 대한 결과를 얻을 수 있습니다.

import java.util.concurrent.*;

public class ThreadPoolExecutorExample {
    public static void main(String[] args) {
    
        // ThreadPoolExecutor 생성
        int corePoolSize = 2;
        int maxPoolSize = 4;
        long keepAliveTime = 10;
        TimeUnit unit = TimeUnit.SECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue);

        // 작업 제출
        for (int i = 1; i <= 10; i++) {
            Future<Integer> future = executor.submit(new Callable<Integer>() {
                public Integer call() throws Exception {
                    System.out.println("Before_Task: " + i + ", ThreadName: " + Thread.currentThread().getName());
                    int result = i * 5;
                    System.out.println("After_Task: " + i + ", ThreadName: " + Thread.currentThread().getName());
                    return result;
                }
            });

            try {
                // 작업 결과 처리
                int result = future.get();
                System.out.println("결과: " + result);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }

        // 스레드 풀 종료
        executor.shutdown();
    }
}