9절. 스레드 풀(ExecutorService)

스레드 폭증

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

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

• → 애플리케이션 성능 저하

스레드 풀(Thread Pool)

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

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

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

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

ExecutorService 인터페이스/ Executor 클래스

• 스레드풀을 생성, 사용

• 스레드 풀 = ExecutorService 객체

ExecutorService

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

스레드풀 생성

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

스레드풀 종료

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

• 스레드풀 종료 메소드

작업 생성

• 하나의 작업은 Runnable 또는 Callable 객체로 표현
• Runnable - 리턴 없음
• Callable<T> - 리턴 있음 (T 타입)

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

작업 처리 요청

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

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

• 작업 처리 도중 예외가 발생할 경우
• execute( ) :
• 스레드가 종료되고 해당 스레드는 제거된다. 따라서 스레드풀은 다른 작업 처리를 위해 새로운 스레드를 생성한다.
• submit( ) :
• 스레드가 종료되지 않고 다음 작업을 위해 재사용된다.

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

• 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( ) - > 예외 발생

/**
 *작업 처리 완료 후 결과가 없는 작업의 완료 처리를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

작업 완료 순으로 통보 받기

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

• 스레드풀에서 작업 처리가 완료된 것만 통보 받는 방법
• CompletionService 는 처리 완료된 작업을 가져오는 poll( ) 과 take( ) 메소드를 제공

• CompletionService 객체 얻기

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

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

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

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

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

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

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

• 작업 완료 통보 Blocking vs Callback

• 콜백 객체와 콜백하기
• 콜백 객체 - 콜백 메소드를 가지고 있는 객체
• java.nio.channels.CompletionHandler : 인터페이스 활용
• 콜백 하기 - 스레드에서 콜백 객체의 메소드 호출

@참고)

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

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

• 그림 출처 - ExecutorService 관련 공부