9절. 스레드 풀(ExecutorService)

스레드 폭증

병렬 작업 처리가 많아지면 스레드의 개수가 증가

스레드 생성과 스케쥴링으로 인해 CPU가 바빠지고, 메모리 사용량이 늘어난다.

→ 애플리케이션 성능 저하

스레드 풀(Thread Pool)

작업 처리에 사용되는 스레드를 제한된 개수만큼 미리 생성

작업 큐에 들어오는 작업들을 하나씩 스레드가 맡아 처리

작업 처리가 끝난 스레드는 작업 결과를 애플리케이션으로 전달/ 스레드는 풀에 반납

스레드는 다시 작업큐에서 새로운 작업을 가져와 처리

쓰레드의 생성 제거 횟수 → 속도 향상!

ExecutorService 인터페이스/ Executor 클래스

스레드풀을 생성, 사용

스레드 풀 = ExecutorService 객체

ExecutorService

병렬 작업시 여러개의 작업을 효율적으로 처리하기 위해 제공되는 인터페이스

notion image

스레드풀 생성

Executors의 두 정적 메서드 중 하나로 생성

메서드 시그니처	초기 스레드수	코어 스레드수	최대 스레드수
newCachedThreadPool( )	0	0	Integer.MAX_VALUE
newFixedThreadPool(int nThreads)	0	nThreads	nThreads

코어 스레드수 - 놀고있는 스레드를 제거하였을때 최소한으로 남아 있어야하는 스레드수

newCachedThreadPool( )

1개 이상의 스레드가 추가되었을 경우 60초 동안 추가된 스레드가 아무작업을 하지 않으면 추가된 스레드를 종료하고 풀에서 제거한다.

newFixedTheadPool(int n Threads)

스레드가 작업을 처리하지 않고 놀고 있더라도 스레드 개수가 줄 지 않는다.

ThreadPoolExecutor을 이용한 직접 생성

스레드의 수를 자동으로 관리하고 싶을 경우 직접 생성해서 사용

e.g.)

코어 스레드 개수가 3, 최대 스레드 개수가 100인 스레드풀
3개를 제외한 나머지 추가된 스레드가 120초 동안 놀고 있을 경우
해당 스레드를 제거해서 스레드 수를 관리

스레드풀 종료

스레드풀의 스레드는 기본적으로 데몬 스레드가 아님

main 스레드가 종료되더라도 스레드풀의 스레드는 작업을 처리하기위해 계속 실행됨
따라서 스레드풀을 종료하여 모든 스레드를 종료시켜야함

스레드풀 종료 메소드

리턴타입	메소드명(매개변수)	설명
void	shutdown()	현재 처리 중인 작업뿐만 아니라 작업큐에 대기하고 있는 모든 작업을 처리한 뒤에 스레드풀 종료
List<Runnable>	shutdonwNow()	현재 작업 처리 중인 스레드를 interrupt 해서 작업중지를 시도하고 스레드풀 종료 작업큐의 미처리된 작업(Runnable) 목록 리턴
boolean	awaitTermination( long timeout, Timeunit unit)	shutdown( ) 메소드 호출 이후, 모든 작업처리를 timeout 시간 내에 완료하면 true, 완료하지 못하면 작업 처리 중인 스레드를 interrupt하고 false 리턴

작업 생성

하나의 작업은 Runnable 또는 Callable 객체로 표현

Runnable - 리턴 없음
Callable<T> - 리턴 있음 (T 타입)

스레드풀에서 작업 처리

작업 큐에서 Runnable 또는 Callable 객체를 가져와 스레드로 하여금 run( ), call( ) 메소드를 실행하도록 하는 것

작업 처리 요청

ExecutorService의 작업 큐에 Runnable 또는 Callable 객체를 넣는 행위를 말한다.

작업 처리 요청을 위해 ExecutorService는 다음의 두 가지의 메소드를 제공

리턴타입	메소드 시그니처	설명
void	execute(Runnable command)	- Runnable을 작업큐에 저장 - 작업 처리 결과 없음
Future<?> Future<V> Future<V>	submit(Runnable task) submit(Runnable task, V result) submit(Callable<V> task)	- Runnable 또는 Callable을 작업큐에 저장 - 리턴된 Future를 통해 작업 처리 결과를 얻을 수 있음

작업 처리 도중 예외가 발생할 경우

execute( ) :

스레드가 종료되고 해당 스레드는 제거된다. 따라서 스레드풀은 다른 작업 처리를 위해 새로운 스레드를 생성한다.

submit( ) :

스레드가 종료되지 않고 다음 작업을 위해 재사용된다.

예제

/TODO - 예제Code 삽입

블로킹 방식의 작업 완료 통보 받기

리턴타입	메서드 시그니처
Future<?>	submit(Runnable task)
Future<V>	submit(Runnable task, V result)
Future<V>	submit(Callable<V> task)

Runnable 또는 Callable 을 작업큐에 저장

리턴된 Future 를 통해 작업 처리 결과 얻음

Future

작업 결과가 아니라 지연 완료 객체 (pending completion)
작업이 완료될 때까지 기다렸다가 최종 결과를 얻기위해서 get( ) 메서드 사용

리턴타입	메서드 시그니처	설명
V	get( )	작업이 완료될 때까지 블로킹 되었다가 처리 결과 V를 리턴
V	get(long timeout, TimeUnit unit)	timeout 시간동안 작업이 완료되면 결과 V를 리턴하지만, 작업이 완료되지 않으면 TimeoutException을 발생시킴

메서드	작업 처리 완료후 리턴 타입	작업 처리 도중 예외 발생
submit(Runnable task)	future.get( ) → null	future.get( ) → 예외 발생
submit(Runnable task, Integer result)	future.get( ) → int 타입 값	future.get( ) - > 예외 발생
submit(Runnable task, String result)	future.get( ) → String 타입 값	future.get( ) - > 예외 발생

예제 - submit(Runnable task)

예제 - submit(Callable<Integer> task)

예제 - submit(Runnable task, V result)

작업 완료 순으로 통보 받기

작업 요청 순서대로 작업 처리가 완료되는 것은 아님

작업의 약과 스레드 스케쥴링에 따라 먼저 요청한 작업이 나중에 완료되는 경우도 발생
처리 결과를 순차적으로 이용할 필요가 없으면 → 작업 처리가 끝난 것 부터 이용하자

스레드풀에서 작업 처리가 완료된 것만 통보 받는 방법

CompletionService 는 처리 완료된 작업을 가져오는 poll( ) 과 take( ) 메소드를 제공

리턴타입	메소드 시그니처	설명
Future<V>	poll( )	완료된 작업의 Future를 가져옴 완료된 작업이 없다면 즉시 null 반환
Future<V>	poll(long timeout, TimeUnit unit)	완료된 작업의 Future를 가져옴 완료된 작업이 없다면 timeout 까지 blocking
Future<V>	take( )	완료된 작업의 Future를 가져옴 완료된 작업이 없다면 있을때 까지 blocking
Future<V>	submit(Callable<V> task)	스레드풀에 Callable 작업 요청
Future<V>	submit(Runnable task, V result)	스레드풀에 Runnable 작업 요청

CompletionService 객체 얻기

작업 처리 요청 방법

poll( )과 take( ) 메소드를 이용해 처리 완료된 작업의 Future를 얻으려면 CompletionService의 submit 메소드로 작업 처리 요청을 해야한다.

완료된 작업 통보 받기

take( ) 메소드를 반복 실행해 완료된 작업을 계속 통보 받을 수 있도록 한다.

예제

콜백 방식의 작업 완료 통보 받기

콜백이란

애플리케이션이 스레드에게 작업 처리를 요청한 후, 다른 기능을 수행할 동안 스레드가 작업을 관료하면 애플리케이션의 메소드를 자동 실행하는 기법
이 때 자동 실행되는 메소드를 콜백메소드라고 한다.

작업 완료 통보 Blocking vs Callback

notion image

콜백 객체와 콜백하기

콜백 객체 - 콜백 메소드를 가지고 있는 객체

java.nio.channels.CompletionHandler : 인터페이스 활용

콜백 하기 - 스레드에서 콜백 객체의 메소드 호출

@참고)

이것이 자바다 - 12.9 스레드 풀 (1)

이것이 자바다 - 12.9 스레드 풀 (2)

그림 출처 - ExecutorService 관련 공부


ExecutorService es = Excutors.newCachedThreadPool();


int nThreads = Runtime.getRuntime().availableProcessors(); //CPU의 코어의 개수
ExecutorService es = Executors.newFixedThreadPool(nThreads);


ExecutorService customEs = new ThreadPoolExecutor(
	3, //코어 스레드 개수
	100, //최대 스레드 개수
	120L, //놀고 있는 시간
	TimeUnit.SECONDS, //시간 단위
	new SynchronousQueue<Runnable>( ) //작업큐
);


/**
 *작업 처리 완료 후 결과가 없는 작업의 완료 처리를Blocking방식으로
*/
public class NoResultEx {

    public static void main(String[] args) {

        //현재 CPU 의 core 수
        int nThreads = Runtime.getRuntime().availableProcessors();
        ExecutorService es = Executors.newFixedThreadPool(nThreads);

        Runnable sumTask = () -> {
            int sum = IntStream.range(1, 10).sum();
            System.out.println("sum : " + sum);
        };

        Future<?> result = es.submit(sumTask);
        try {
            Object o = result.get(); // null
            //main thread 는 Blocking
        } catch (InterruptedException e) {
            //작업 처리 도중 스레드가 interrupt 될 결우
        } catch (ExecutionException e) {
            //작업 처리 도중 예외가 발생된 경우
        }
        System.out.println("main thread terminated ");
        es.shutdown();
    }
}


sum = 45
main thread terminated


/**
 * 작업 처리 완료 후 결과가 있는 작업의 완료 처리를 Blocking 방식으로
 */
public class ResultByCallableEx {

    public static void main(String[] args) {

        //현재 CPU 의 core 수
        int nThreads = Runtime.getRuntime().availableProcessors();
        ExecutorService es = Executors.newFixedThreadPool(nThreads);

        Callable<Integer> sumTask = () -> IntStream.range(1, 10).sum();

        Future<Integer> result = es.submit(sumTask);
        try {
            Integer sum = result.get();// sum = 45;
            System.out.println("sum = " + sum);
            //main thread 는 Blocking
        } catch (InterruptedException e) {
            //작업 처리 도중 스레드가 interrupt 될 결우
        } catch (ExecutionException e) {
            //작업 처리 도중 예외가 발생된 경우
        }
        System.out.println("main thread terminated ");
        es.shutdown();
    }
}


sum = 45
main thread terminated


public class ResultByRunnableEx {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Result result = new Result();
        Task task1 = new Task(result);
        Task task2 = new Task(result);

        ExecutorService es = Executors.newFixedThreadPool(2);
        Future<Result> future1 = es.submit(task1, result);
        Future<Result> future2 = es.submit(task2, result);

        result = future1.get();
        System.out.println("결과 = " + result.data);
        result = future2.get();
        System.out.println("결과 = " + result.data);
    }
    static class Task implements Runnable{

        Result result;

        public Task(Result result) {
            this.result = result;
        }

        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                result.add(i);
            }
        }
    }

    static class Result{
        int data;

        synchronized void add(int val){
            data += val;
        }
    }
}


결과 = 45
결과 = 90


ExecutionService es = Executors.newFixedThreadPool(2);
CompletionSerice<V> cs = new ExecutorCompletionService<V>(es);


cs.submit(Callable<V> task);
cs.submit(Runnalble task, V result);


import java.util.concurrent.*;

public class CompletionServiceEx {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(2);
        CompletionService<Integer> cs = new ExecutorCompletionService<>(es);

        System.out.println("[작업 처리 요청]");
        for (int i = 0; i < 3; i++) {
            int j = i;
            cs.submit(() -> j);
        }

        System.out.println("[처리 완료된 작업 확인]");
        es.submit(() -> {
            while(true) {
                try {
                    Future<Integer> future = cs.take();
                    int ret = future.get();
                    System.out.println("처리 결과 = " + ret);
                } catch (InterruptedException | ExecutionException e) {
                    break;
                }
            }
        });

        Thread.sleep(3000);
        es.shutdownNow();
    }
}


[작업 처리 요청]
[처리 완료된 작업 확인]
처리 결과 = 0
처리 결과 = 2
처리 결과 = 1