[인프런] 김영한의 실전 자바 - 고급 1편, 멀티스레드와 동시성 / 11. 스레드 풀과 Executor 프레임워크 2

ExecutorService 우아한 종료 - 소개

ExecutorService의 종료 메서드

 

1. 서비스 종료

void shutdown()

> 새로운 작업을 받지 않고 이미 제출된 작업을 모두 완료한 후 종료

> 논 블로킹 메서드 : 이 메서드를 호출한 스레드는 대기하지 않고 바로 다음 코드 호출

 

1) shutdown() 호출

2) ExecutorService는 새로운 요청 거절

3) 스레드 풀의 스레드는 처리중인 작업 완료

4) 스레드 풀의 스레드는 큐에 남아있는 작업도 모두 꺼내 완료

5) 모든 작업을 완료하면 자원 정리

 

List<Runnable> shutdownNow()

> 실행 중인 작업을 중단하고 대기 중인 작업을 반환해 즉시 종료

> 실행 중인 작업을 중단하기 위해 인터럽트를 발생시킴

> 논 블로킹 메서드

 

 

1) shutdownNow() 호출

2) ExecutorService는 새로운 요청 거절

3) 큐를 비우며 큐에 있는 작업을 모두 꺼내 컬렉션으로 반환 -> List<Runnable> runnables = es.shutdownNow()

4) 작업 중인 스레드에 인터럽트 발생 -> taskA, taskB는 인터럽트가 걸리고 taskC, taskD는 수행되지 않음

5) 작업을 완료하면 자원 정리

 

 

 

2. 서비스 상태 확인

boolean isShutdown()

> 서비스가 종료되었는지 확인

 

boolean isTerminated()

> shutdown(), shutdownNow() 호출 후 모든 작업이 완료되었는지 확인

 

 

 

3. 작업 완료 대기

boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException

> 서비스 종료 시 모든 작업이 완료될 때까지 대기, 지정 시간까지만 대기

> 블로킹 메서드

 

 

 

4. close()

> 자바 19부터 지원하는 서비스 종료 메서드로 shutdown과 같음

> 더 자세히는, shutdown() 호출 후 하루를 기다려도 작업이 완료되지 않으면 shutdownNow()를 호출

> 호출한 스레드에 인터럽트가 발생해도 shutdownNow() 호출

---

 

ExecutorService 우아한 종료 - 구현

showdown()으로 이미 들어온 모든 작업을 처리 후 서비스를 우아하게 종료하는 것이 이상적이지만, 요청이 너무 많이 들어와 큐에 대기중인 작업이 너무 많거나, 버그가 발생해 특정 작업이 끝나지 않을 수도 있다.

이럴 경우, shutdownNow()로 작업을 강제 종료한다.

 

ExecutorShutdownMain.java : shutdown()을 통해 우아한 종료를 시도하고, 10초간 종료되지 않으면 shutdownNow()로 강제 종료하는 방식 구현 -> ExecutorService 공식 API 문서에서 제안하는 방식

package thread.executor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static thread.executor.ExecutorUtils.printState;
import static util.MyLogger.log;

public class ExecutorShutdownMain {
    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(2);

        es.execute(new RunnableTask("taskA")); // 1초
        es.execute(new RunnableTask("taskB")); // 1초
        es.execute(new RunnableTask("taskC")); // 1초
        es.execute(new RunnableTask("longTask", 100_000)); // 100초

        printState(es);
        log("== shutdown 시작 ==");
        shutdownAndAwaitTermination(es);
        log("== shutdown 완료 ==");
    }

    private static void shutdownAndAwaitTermination(ExecutorService es) {
        es.shutdown(); // non-blocking, 새로운 작업을 받지 않음

        try {
            // 이미 대기중인 작업들을 모두 완료할 때까지 10초 기다림
            if(!es.awaitTermination(10, TimeUnit.SECONDS)) {
                // 정상 종료가 너무 오래 걸리면 ...
                log("서비스 정상 종료 실패 -> 강제 종료 시도");

                es.shutdownNow(); 

                // showdownNow 호출 후 작업이 취소될 때까지 대기
                if(!es.awaitTermination(10, TimeUnit.SECONDS)) {
                    log("서비스가 종료되지 않았습니다");
                }
            }

        } catch(InterruptedException e) {
            // awaitTermination()으로 대기중인 현재 스레드가 인터럽트 될 수 있음
            es.shutdownNow();
        }

    }
}

 

실행 결과

 

실행 분석

if(!es.awaitTermination(10, TimeUnit.SECONDS))

-> 블로킹 메서드

-> main 스레드는 대기하며 서비스 종료를 10초간 기다리고 작업이 완료된다면 true 반환

-> taskA, taskB, taskC는 수행이 완료되었지만 longTask는 완료되지 않고 false 반환

 

es.shutdownNow();

-> 강제 종료에 들어가고 작업 중인 스레드에 인터럽트가 발생

-> RunnableTask에서 sleep() 호출 시 InterruptedException이 터짐

-> 스레드도 작업을 종료하고, shutdownNow()를 통한 강제 shutdown이 완료됨

 

 

 

=> 서비스 종료 시 생각보다 고려해야 할 점이 많다.

=> 기본적으로 우아한 종료를 선택하고, 우아한 종료가 되지 않으면 그 다음으로 강제 종료를 하는 방식으로 접근하는 것이 좋다.

---

Executor 스레드 풀 관리

Executor 프레임워크가 어떤식으로 스레드를 관리하는지 깊이있게 알아보자.

 

PoolSizeMainV1.java

package thread.executor.poolsize;

import thread.executor.ExecutorUtils;
import thread.executor.RunnableTask;

import java.util.concurrent.*;

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

public class PoolSizeMainV1 {
    public static void main(String[] args) {
        ArrayBlockingQueue workQueue = new ArrayBlockingQueue<>(2);

        ExecutorService es = new ThreadPoolExecutor(2, 4,
                3000, TimeUnit.MILLISECONDS, workQueue);
        printState(es);

        es.execute(new RunnableTask("task1"));
        printState(es, "task1");

        es.execute(new RunnableTask("task2"));
        printState(es, "task2");

        es.execute(new RunnableTask("task3"));
        printState(es, "task3");

        es.execute(new RunnableTask("task4"));
        printState(es, "task4");

        es.execute(new RunnableTask("task5"));
        printState(es, "task5");

        es.execute(new RunnableTask("task6"));
        printState(es, "task6");

        try {
            es.execute(new RunnableTask("task7"));
        } catch (RejectedExecutionException e) {
            log("task7 실행 거절 예외 발생: " + e);
        }

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

        sleep(3000);
        log("== maximumPoolSize 대기 시간 초과 ==");
        printState(es);

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

 

실행 결과

 

 

 

실행 분석

 

1. 초기 상태

10::08::55.866 [     main] [pool=0, active=0, queuedTasks=0, completedTasks=0]

 

2. task1, task2 작업 요청

 

10::08::55.904 [     main] task1 -> [pool=1, active=1, queuedTasks=0, completedTasks=0]
10::08::55.904 [     main] task2 -> [pool=2, active=2, queuedTasks=0, completedTasks=0]

-> Executor는 스레드 풀에 스레드가 core 사이즈 만큼 있는지 확인

-> core 사이즈 만큼 없으므로 스레드를 생성하고(2개) 각 작업을 수행

 

3. task3, task4 작업 요청

10::08::55.905 [     main] task3 -> [pool=2, active=2, queuedTasks=1, completedTasks=0]
10::08::55.905 [     main] task4 -> [pool=2, active=2, queuedTasks=2, completedTasks=0]

-> Executor에 core 사이즈만큼 스레드가 이미 만들어져 있으므로 큐에 작업을 보관한다.

 

5. task5 작업 요청

10::08::55.905 [     main] task5 -> [pool=3, active=3, queuedTasks=2, completedTasks=0]

-> 스레드 풀에 스레드가 core 사이즈 만큼 참 = 2개

-> 큐도 가득 참 = 2개

-> Excutor는 maximumPoolSize까지 초과 스레드를 만들어 작업을 수행 = 4개

-> 초과 스레드인 스레드3을 만든다.

-> 작업을 큐에 넣지 않고 스레드3이 바로 task5를 처리한다.

 

6. task6 작업 요청

5번과 동일

 

7. task7 작업 요청

 

10::08::55.909 [     main] task7 실행 거절 예외 발생: java.util.concurrent.RejectedExecutionException: 
Task thread.executor.RunnableTask@119d7047 rejected from java.util.concurrent.ThreadPoolExecutor@776ec8df
[Running, pool size = 4, active threads = 4, queued tasks = 2, completed tasks = 0]

-> 큐도 가득 찼고, 스레드 풀도 max 사이즈 만큼 가득 참

-> RejectedExecutionException 발생하며 작업을 거절

 

8. 이후 task1~task6의 작업이 완료된다.

10::08::58.914 [     main] == 작업 수행 완료 ==
10::08::58.916 [     main] [pool=4, active=0, queuedTasks=0, completedTasks=6]
10::08::58.916 [     main] [pool=4, active=0, queuedTasks=0, completedTasks=6]
10::09::01.921 [     main] == maximumPoolSize 대기 시간 초과 ==
10::09::01.922 [     main] [pool=2, active=0, queuedTasks=0, completedTasks=6]

-> 스레드3과 스레드4와 같은 초과 스레드들은 지정된 시간(여기서는 3초)까지 작업을 하지 않고 대기하면 제거된다.

-> 초과 스레드가 작업을 처리할 때 마다 이 시간은 계속 초기화 된다.

 

9. shutdown()으로 풀의 스레드도 모두 제거된다.

10::09::01.922 [     main] == shutdown 완료 ==
10::09::01.923 [     main] [pool=0, active=0, queuedTasks=0, completedTasks=6]

 

 

 

스레드 미리 생성

응답시간이 아주 중요한 서버라면, 서버가 고객의 첫 요청을 받기 전 스레드를 미리 스레드 풀에 생성해두고 싶을 수도 있다.

ThreadPoolExecutor.prestartAllCoreThreads()로 기본 스레드를 미리 생성할 수도 있다.

 

ExecutorService es = Executors.newFixedThreadPool(1000);
ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) es;
poolExecutor.prestartAllCoreThreads();

---

자바는 Executors 클래스를 통해 3가지 기본 전략을 제공한다.

1) newSingleThreadPool() : 단일 스레드 풀 전략

2) newFixedThreadPool(nThreads) : 고정 스레드 풀 전략

3) newCachedThreadPool() : 캐시 스레드 풀 전략

 

Executor 전략 - 고정 풀 전략

newFixedThreadPool(nThreads)

 

> 스레드 풀에 nThreads 만큼의 기본 스레드를 생성, 초과 스레드 생성 X

> 큐 사이즈에 제한이 없음 : LinkedBlockingQueue

> 스레드 수가 고정되어 있기에 CPU, 메모리 리소스가 어느정도 예측 가능한 안정적인 방식

 

 

 

PoolSizeMainV2.java

package thread.executor.poolsize;

import thread.executor.RunnableTask;

import java.util.concurrent.*;

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

public class PoolSizeMainV2 {
    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(2);

        log("pool 생성");
        printState(es);

        for(int i = 1; i <= 6; i++) {
            String taskName = "task" + i;
            es.execute(new RunnableTask(taskName));
            printState(es, taskName);
        }

        es.shutdown();
        log("== shutdown 완료 ==");
    }
}

 

실행 결과

-> 2개의 스레드가 안정적으로 작업을 처리하고 있음

 

특징

: 스레드 수가 고정적이라 CPU, 메모리 리소스가 어느정도 예측 가능함

: 큐 사이즈도 제한이 없어 작업을 많이 담아두어도 문제가 없음

 

주의사항

갑작스런 요청 증가?

-> 사용자가 늘어나도 CPU나 메모리 사용량이 확 늘어나지 않음

-> 큐에 작업 내용만 쌓여가는 것

-> 스레드의 작업이 처리되는 시간보다 쌓이는 시간이 더 빠를 경우 문제가 될 수 있음

-> 사용자가 응답 받는 시간이 점점 느려짐

---

Executor 전략 - 캐시 풀 전략

newCachedThreadPool()

 

> 기본 스레드를 사용하지 않고 60초 생존 주기를 가진 초과 스레드만 사용

> 초과 스레드 수의 제한이 X

> 큐에 작업을 저장하지 않음 : SynchronousQueue

> 모든 요청이 대기하지 않고 스레드가 바로바로 처리해 빠른 처리가 가능

 

* SynchronousQueue

: BlockingQueue 인터페이스 구현체 중 하나

: 내부에 저장 공간이 없음 -> 생산자 작업을 소비자 스레드에게 직접 전달

: 이름 그대로 생산자와 소비자를 동기화하는 큐

: 중간에 버퍼를 두지 않는 스레드간 직거래 개념

 

 

 

PoolSizeMainV3.java

package thread.executor.poolsize;

import thread.executor.RunnableTask;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

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

public class PoolSizeMainV3 {
    public static void main(String[] args) {
        ExecutorService es = Executors.newCachedThreadPool();

        log("pool 생성");
        printState(es);

        for(int i = 1; i <= 4; i++) {
            String taskName = "task" + i;
            es.execute(new RunnableTask(taskName));
            printState(es, taskName);
        }

        es.shutdown();
        log("== shutdown 완료 ==");
    }
}

 

실행 결과

-> 모든 작업이 대기하지 않고 작업 수 만큼 스레드가 생기며 바로 실행됨

 

특징

: 매우 빠르고 유연한 전략

: 기본 스레드도 없고 대기 큐에 작업도 쌓이지 않음

: 작업 요청이 오면 초과 스레드로 작업을 바로바로 처리

: 초과 스레드 수의 제한이 없어 CPU와 메모리 자원만 허용한다면 시스템 자원을 최대로 사용할 수 있음

: 초과 스레드는 60초간 생존하기 때문에(keepAliveTime 속성 참고) 작업 수에 맞춰 적절한 수의 스레드가 재사용됨

 

주의사항

서비스의 사용자가 점점 늘어나고, 갑작스런 요청이 증가하면?

-> CPU 사용량이 이미 100%, 메모리 사용량도 지나치게 높아질 수 있음 -> 시스템 전체가 느려지거나 멈출 수 있음

-> 스레드가 무한으로 생성되기에 시스템이 너무 많은 스레드에 잠식 당할 수 있음

 

 

 

=> 고정 스레드 풀 전략은 서버 자원은 여유가 있는데 사용자만 점점 느려지는 문제가 발생할 수 있고,

=> 캐시 스레드 풀 전략은 서버의 자원을 최대로 활용하지만 서버가 감당할 수 있는 임계점을 넘는 순간 시스템이 다운될 수 있음

---

Executor 전략 - 사용자 정의 풀 전략

다음과 같이 세분화된 전략을 사용해 사용자 확대 및 갑작스런 요청 증가의 상황을 어느정도 대응할 수 있다.

 

1. 일반 : 일반적인 상황에는 CPU나 메모리 자원을 예측할 수 있도록 고정 크기 스레드로 서비스 운영

2. 긴급 : 사용자 요청이 갑자기 증가하면 긴급하게 스레드를 추가로 투입

3. 거절 : 사용자 요청이 폭증해 긴급 대응도 어렵다면, 사용자 요청 거절

 

 

 

PoolSizeMainV4.java

ExecutorService es = new ThreadPoolExecutor(100, 200, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));

-> 100개의 기본 스레드 사용

-> 긴급 대응 가능한 긴급 스레드 100개 사용, 긴급 스레드 생명주기는 60초

-> 1000개의 작업이 큐에 대기 가능

 

package thread.executor.poolsize;

import thread.executor.RunnableTask;

import java.util.concurrent.*;

import static thread.executor.ExecutorUtils.printState;
import static util.MyLogger.log;

public class PoolSizeMainV4 {
    static final int TASK_SIZE = 1100; // 1. 일반
    // static final int TASK_SIZE = 1200; // 2. 긴급
    // static final int TASK_SIZE = 1201; // 3. 거절

    public static void main(String[] args) {
        ExecutorService es = new ThreadPoolExecutor(100, 200, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));
        printState(es);

        long startMs = System.currentTimeMillis();

        for(int i = 1; i <= TASK_SIZE; i++) {
            String taskName = "task" + i;
            try {
                es.execute(new RunnableTask(taskName));
                printState(es, taskName);
            } catch(RejectedExecutionException e) {
                log(taskName + " -> " + e);
            }
        }

        long endMs = System.currentTimeMillis();
        log("time: " + (endMs - startMs));

        es.shutdown();
        log("== shutdown 완료 ==");
    }
}

 

실행 결과

1. 일반

-> 100개의 기본 스레드가 모두 처리

 

2. 긴급

-> 100개의 기본 스레드 + 100개의 초과 스레드가 처리

-> 큐도 가득 차야 긴급 상황으로 인지!

 

3. 거절

-> 큐에 담긴 작업 1000개 초과, 초과 스레드도 넘어가 예외 발생

---

Executor 예외 정책

생산자 소비자 문제를 실무에서 사용할 때는, 결국 소비자가 처리할 수 없을 정도로 생산 요청이 가득 차면 어떻게 할지를 정해야 한다.

개발자에게 로그를 남겨줘야 하고, 사용자에게 현재 시스템에 문제가 있다고 알리는 것도 필요하다.

이를 위해 예외 정책이 필요하다.

 

ThreadPoolExecutor는 큐가 가득 차고 초과 스레드도 더이상 할당할 수 없다면 작업을 거절한다. 그리고 작업을 거절하는 다양한 정책을 제공한다.

 

1. AbortPolicy : 새로운 작업을 제출할 때 RejectedExecutionException을 발생시킴 = 기본

2. DiscardPolicy : 새로운 작업을 조용히 버림

3. CallerRunsPolicy : 새로운 작업을 제출한 스레드가 대신 직접 작업 실행

4. 사용자 정의(RejectedExecutionHandler) : 개발자가 직접 정의한 거절 정책 사용

 

 

 

1. AbortPolicy

RejectMainV1.java

package thread.executor.reject;

import thread.executor.RunnableTask;

import java.util.concurrent.*;

import static util.MyLogger.log;

public class RejectMainV1 {

    public static void main(String[] args) {
        // 큐에 작업을 넣지 않도록 SynchronousQueue 사용
        ExecutorService executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>(), new ThreadPoolExecutor.AbortPolicy());

        executor.submit(new RunnableTask("task1"));

        try {
            executor.submit(new RunnableTask("task2"));
        } catch(RejectedExecutionException e) {
            log("요청 초과");
            log(e);
        }
        executor.shutdown();
    }
}

 

실행 결과

-> task2 요청 시 허용 작업을 초과하고 RejectedExecutionException을 던짐

-> 예외를 잡아 작업을 포기하거나 사용자에게 알리거나 다시 시도하면 됨

 

 

 

2. DiscardPolicy

RejectMainV2.java

package thread.executor.reject;

import thread.executor.RunnableTask;

import java.util.concurrent.*;

import static util.MyLogger.log;

public class RejectMainV2 {

    public static void main(String[] args) {
        // 큐에 작업을 넣지 않도록 SynchronousQueue 사용
        ExecutorService executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>(), new ThreadPoolExecutor.DiscardPolicy());

        executor.submit(new RunnableTask("task1"));
        executor.submit(new RunnableTask("task2"));
        executor.submit(new RunnableTask("task3"));
        executor.submit(new RunnableTask("task4"));

        executor.shutdown();
    }
}

 

실행 결과

-> task2, 3, 4는 거절됨

 

DiscardPolicy 구현 확인 : 아무것도 하지 않음

empty

 

 

 

3. CallerRunsPolicy 

RejectMainV3.java

package thread.executor.poolsize;

import thread.executor.RunnableTask;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

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

public class PoolSizeMainV3 {
    public static void main(String[] args) {
        ExecutorService es = Executors.newCachedThreadPool();

        log("pool 생성");
        printState(es);

        for(int i = 1; i <= 4; i++) {
            String taskName = "task" + i;
            es.execute(new RunnableTask(taskName));
            printState(es, taskName);
        }

        es.shutdown();
        log("== shutdown 완료 ==");
    }
}

 

실행 결과

-> task1은 풀에 스레드가 있어서 수행

-> task2는 작업할 스레드가 없어 거절해야 하지만, 작업을 요청한 main 스레드가 대신 수행

 

=> 생산자 스레드가 소비자 대신 일을 수행해주지만, 덕분에 작업 생산 자체가 느려질 수 있음

=> main 스레드는 task2를 다 실행하고 나서야 task3을 생산하게 됨

 

CallerRunsPolicy 구현 확인

-> r.run()으로 별도 스레드에서 수행하는 것이 아닌 main 스레드가 직접 수행

 

 

 

4. 사용자 정의

RejectMainV4.java : RejectedExecutionHandler 인터페이스를 구현해 자신만의 거절 처리 전략 정의

package thread.executor.reject;

import thread.executor.RunnableTask;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static util.MyLogger.log;

public class RejectMainV4 {

    public static void main(String[] args) {
        // 큐에 작업을 넣지 않도록 SynchronousQueue 사용
        ExecutorService executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>(), new MyRejectedExecutionHandler());

        executor.submit(new RunnableTask("task1"));
        executor.submit(new RunnableTask("task2"));
        executor.submit(new RunnableTask("task3"));

        executor.shutdown();
    }

    static class MyRejectedExecutionHandler implements RejectedExecutionHandler {

        static AtomicInteger count = new AtomicInteger();

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            int i = count.incrementAndGet();
            log("경고 : 거절된 누적 작업 수 -> " + i);
        }
    }
}

 

실행 결과

-> 거절된 작업을 버리지만 대신 경고 로그를 남겨 개발자가 문제를 인지할 수 있음

---

정리

가장 좋은 최적화는 최적화하지 않는 것

많은 개발자가 미래에 발생하지 않은 일로 코드를 최적화하는 경우가 많다.

예를 들어, A와 관련된 기능을 매우 많이 최적화했는데 사용자가 없어 결국 버리게 되는 경우도 있다.

중요한 것은 예측 불가능한 너무 먼 미래 보다 현재 상황에 맞는 최적화가 필요하다는 점

 

-> 시스템의 상황을 잘 모니터링 하다가 최적화가 필요한 부분이 발생하면 그때 필요한 부분들을 개선하면 됨

 

일반적인 상황이라면 고정 스레드 풀 전략이나 캐시 스레드 풀 전략으로 충분하다.

한번에 처리할 수 있는 수를 제안하고 안정적 처리를 원한다면 고정 풀을, 사용자 요청을 빠르게 대응하고 싶다면 스레드 풀을 사용하면 된다.

 

백엔드 서버 개발자라면 시스템 자원을 적절히 활용하되 최악의 경우 적절한 거절을 통해 시스템이 다운되지 않도록 하자.