아이템 80. 스레드보다는 실행자, 태스크, 스트림을 애용하라

[chikeem90]

실행자 프레임워크 (Executor Framework)

• 수행 정책 설정과 해야할 태스크를 지정하면, 그 설정에 맞게 알아서 수행해서 처리해주는 프레임워크.
• 실행자 서비스는 스레드 풀을 생성하고 수행하고자 하는 태스크를 설정에 따라 각 스레드에 위임하는 역할을 함.

실행자 프레임워크의 주요 인터페이스

• Executor : 새로운 태스크 시작을 지원하는 간단한 인터페이스.
• ExecutorService : 개별 작업과 실행자 자체의 수명 주기를 관리하는 데 도움이 되는 기능을 추가한 Executor의 하위 인터페이스. (실질적으로 사용)
• ScheduledExecutorService는 Executor의 하위 인터페이스로, 태스크의 이후 실행 또는 주기적 실행을 지원함

실행자 서비스 (ExecutorService) 에서 제공하는 기능

• 태스크를 전달받아 수행 (execute)
• 실행자 graceful shutdown (shutdown)
• 특정 태스크가 완료되기를 기다림 (get)
• 태스크 모음 중 아무것 하나 또는 모든 태스크가 완료되기를 기다림 (invokeAny, invokeAll)
• 실행자 서비스가 종료하기를 기다림 (awaitTermination)
• 완료된 태스크들의 결과를 차례로 받음 (ExecutorCompletionService)
• 태스크를 특정 시간에 혹은 주기적으로 실행하게 함 (ScheduledThreadPoolExecutor)

실행자 서비스 사용 방법

• 실행자 대부분은 java.util.concurrent.Executors의 정적 팩토리를 이용해 생성할 수 있으며, 여러 설정을 주어서 제어하고 싶을 경우 ThreadPoolExecutor 클래스를 직접 사용해도 됨.
• newFixedThreadPool : 고정된 수의 스레드를 갖는 스레드 풀을 생성.
• newSingleThreadExecutor : 단일 스레드를 사용하는 ExecutorService.
• newCachedThreadPool : 필요에 따라 새 스레드를 생성하지만 이전에 생성된 스레드를 사용할 수 있을 때 재사용하는 스레드 풀을 생성.
• newScheduledThreadPool : 지정된 시간 이후 실행되거나 주기적으로 실행되도록 명령을 예약할 수 있는 스레드 풀을 생성.

예제

• newSingleThreadExecutor에 태스크를 위임한 경우

    public static void main(String[] args) {
        Callable<Integer> task = () -> {
            int num1 = 10;
            int num2 = 20;
            int num3 = num1 + num2;
            System.out.println(Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        ExecutorService exec = Executors.newSingleThreadExecutor();
        Future<Integer> result = exec.submit(task); // 실행자에 위임
        System.out.println("End " + Thread.currentThread().getName());
        exec.shutdown();
    }

End main
pool-1-thread-1 : 30

• newFixedThreadPool에 태스크를 위임한 경우

 public static void main(String[] args) {
        Callable<Integer> task1 = () -> {
            int num1 = 1;
            int num2 = 2;
            int num3 = num1 + num2;
            System.out.println("task1 : " + Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        Callable<Integer> task2 = () -> {
            int num1 = 3;
            int num2 = 4;
            int num3 = num1 + num2;
            System.out.println("task2 : " + Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        Callable<Integer> task3 = () -> {
            int num1 = 5;
            int num2 = 6;
            int num3 = num1 + num2;
            System.out.println("task3 : " + Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        Callable<Integer> task4 = () -> {
            int num1 = 7;
            int num2 = 8;
            int num3 = num1 + num2;
            System.out.println("task4 : " + Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        Callable<Integer> task5 = () -> {
            int num1 = 9;
            int num2 = 10;
            int num3 = num1 + num2;
            System.out.println("task5 : " + Thread.currentThread().getName() + " : " + num3);
            return num3;
        };

        ExecutorService exec = Executors.newFixedThreadPool(2);
        // 실행자에 위임
        exec.submit(task1);
        exec.submit(task2);
        exec.submit(task3);
        exec.submit(task4);
        exec.submit(task5);
        System.out.println("End " + Thread.currentThread().getName());
        exec.shutdown();

End main
task2 : pool-1-thread-2 : 7
task1 : pool-1-thread-1 : 3
task3 : pool-1-thread-1 : 11
task5 : pool-1-thread-1 : 19
task4 : pool-1-thread-2 : 15

실행자 서비스 사용 시 주의사항

• newCachedThreadPool의 경우 작은 프로그램이나 가벼운 서버인 경우에 사용해도 괜찮으나, 무거운 프로덕션 서버에서 사용하는 것은 적절하지 않음.
• newCachedThreadPool에서 요청받은 태스크들이 큐에 쌓이지 않고 즉시 스레드에 위임되기 때문에, 가용한 스레드가 없다면 새로 하나 생성하여 처리하도록 위임함.
• 서버가 무거운 작업을 수행하는 경우 CPU 이용률이 100%에 도달하고, 새로운 태스크가 도착하면 또 다른 스레드를 생성하여 상황이 악화될 수 있음.
• 따라서 무거운 프로덕션 서버에서는 스레드 개수를 고정한 newFixedThreadPool을 선택하거나 완전히 통제할 수 있는 ThreadPoolExecutor를 직접 사용하는 편이 나음.

실 사용처

• 메인 로직과 별개로 따로 수행되면서 처리되어야 하는 부분 (ex. 어떤 이벤트가 발생할 때마다 소비해서 그에 해당하는 처리를 해야하는 부분)
• Spring Batch에서 사용 (Spring의 taskExecutor)

추가로 보면 좋은 내용

• https://leeyh0216.github.io/posts/truth_of_threadpoolexecutor/
• https://pompitzz.github.io/blog/Java/threadPoolExecutor.html#spring%E1%84%8B%E1%85%B4-threadpooltaskexecutor
• https://hamait.tistory.com/612

참고한 내용

• https://docs.oracle.com/javase/tutorial/essential/concurrency/exinter.html
• https://zoiworld.tistory.com/623
• https://m.blog.naver.com/mals93/220743747346

[corock]

책 내용만 봐서는 실전 예시를 떠올리기가 쉽지 않았는데, 비동기 처리와 Spring Batch 가 있군요.
대부분의 IT 사관학교 자바 수업은 스레드만 언급하고 넘어가는 경우가 많은데, 실행자 프레임워크를 언급하는 것도 좋겠다고 생각했습니다.
제가 먼 훗날에 현우진 비슷한 포지션이 된다면 적극 반영해보겠습니다. 잘 읽었습니다~!

[jinan159]

newCachedThreadPool 스레드 개수를 제한을 안둔다니... 이 클래스는 무서운 녀석이군요...