[Java] Executor 프레임워크 심화

ExecutorService 종료

`ExecutorService`를 종료하는 방법은 크게 두 가지 방법이 있다.

1. es.shutdown(): 논블로킹 우아한 종료. (graceful shutdown)
   새 작업이 추가되는 것만 막고 기존에 실행/대기 중인 작업이 모두 끝날 때까지 기다린 다음 스레드풀을 제거함.
2. es.shutdownNow(): 논블로킹 강제 종료.
   대기 중인 작업을 모두 작업 큐에서 제거하고, 실행 중인 작업 모두에 인터럽트를 걸어 종료시킨다.
   반환값으로 대기하고 있던 작업들을 반환. `List<Runnable>`

`ExecutorService`가 종료된 것을 확인한 다음에 어떤 작업을 해야 한다면, 즉 블로킹 메서드가 필요하다면, `es.awaitTermination()`을 사용해 대기하면 된다.

 

• 우아한 종료 구현 (`ExecutorService`의 공식 안내와 비슷)

private static void shutdownAndAwaitTermination(ExecutorService es) {
    es.shutdown(); // 논블로킹

    try {
        // 이미 대기 중인 작업들을 모두 완료할 때까지 1분 기다린다. (블로킹)
        if (!es.awaitTermination(60, TimeUnit.SECONDS)) {
            // 정상 종료가 너무 오래걸리면...
            log("서비스 정상 종료 실패 -> 강제 종료 시도");
            // 실행 중인 작업에 인터럽트를 걸어서 강제 종료를 시도함
            es.shutdownNow();
            // 작업이 취소될 땔 때까지 대기한다.
            if (!es.awaitTermination(60, TimeUnit.SECONDS)) {
                log("서비스가 종료되지 않았습니다.");
            }
        }
    } catch (InterruptedException e) {
        // awaitTermination()으로 대기 중인 스레드가 인터럽트 될 수 있다.
        es.shutdownNow();
        // 이 스레드의 인터럽트 상태 유지를 위해 필요
        Thread.currentThread().interrupt(); // 인터럽트 상태를 false로 하고 싶다면 넣지 않으면 됨
    }
}

 

강제 종료하는 방식이 인터럽트를 거는 방식이기 때문에, 작업 강제종료가 불가능할 수 있다. 인터럽트를 받지 않는 작업을 하는 상황이 무한 반복하는 경우다. 그런 경우에는 Java 프로그램을 종료해야만 저 작업을 종료시킬 수가 있다. (`while(true) { a++; }` 같은 경우들)

 

인터럽트로 작업을 종료시킬 수 있는 경우는 다음 두 가지 상황이 있다.

1. 인터럽트 예외를 던지는 메서드를 실행 중
2. 인터럽트 상태를 확인하는 `isInterruped()` 메서드로 확인 (주로 반복문에서 사용)

 

인터럽트가 불가능한 상황이 발생하면 개발자가 인지를 하고 고쳐야 하는 상황이기 때문에 로그를 남길 필요가 있다.

`shutdownNow()`를 하고도 좀 기다렸는데 종료되지 않는 상황을 로그로 남긴 것이 위 코드에서 "서비스가 종료되지 않았습니다." 부분이다.

 

ExecutorService의 Thread Pool 관리

corePoolSize와 maximumPoolSize를 정확히 이해해보자.

 

`corePoolSize`만큼은 한 번 만들어지고 나면 더 이상 스레드 수를 줄이지 않는 숫자다.

`maximumPoolSize`는 이 스레드 풀이 가질 수 있는 최대 스레드 수다. 방금 내용은 당연하고 쉽지만, 의외로 당연하다고 생각한 것과 다른 부분, 놓칠 수 있는 부분도 있다.

 

Q. 아래 코드에서 어느 시점에 es의 스레드 풀 크기가 3개, 4개로 올라갈까?

public static void main(String[] args) {
    ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2); // 크기가 2로 고정
    // CorePoolSize = 2, MaximumPoolSize = 4
    ExecutorService es = new ThreadPoolExecutor(2, 4, 3000, TimeUnit.MILLISECONDS, workQueue);

    es.execute(new RunnableTask("task1"));
    es.execute(new RunnableTask("task2"));
    es.execute(new RunnableTask("task3"));
    es.execute(new RunnableTask("task4"));
    es.execute(new RunnableTask("task5"));
    es.execute(new RunnableTask("task6"));
    try {
        es.execute(new RunnableTask("task7"));
    } catch (RejectedExecutionException e) {
        log("task7 실행 거절 예외 발생: " + e);
    }
    es.close();
}

 

task3을 실행했을 때 스레드 풀 크기가 3이 되고,

task4를 실행했을 때 스레드 풀 크기가 4가 될 줄 알았다.

 

그런데 그게 아니라

대기 큐가 꽉 차고 나서 추가 작업이 들어올 때 스레드 풀이 증가했다.

(task5 실행할 때 풀 크기 증가)

 

그리고 스레드 풀 크기가 늘어나서 여유 스레드가 추가로 생성됐을 때, 먼저 들어온 작업이 실행되는게 아니고, 스레드를 추가로 생성하도록 만든 작업, 위 상황에서는 task5가 먼저 실행됐다.

 

위 상황을 토대로 더 생각해볼 수 있는 것은, 작업 대기 큐의 크기가 무한이라면 `maximumPoolSize`를 설정하는 의미가 없어진다. 큐가 꽉 찰 일이 없으니 `corePoolSize` 이상으로 늘어나지 않을 것이다.

`maximumPoolSize`를 활용할 때는 작업을 담는 `BlockingQueue`의 크기 제한이 필요하다는 점을 꼭 명심해야 한다.

 

그리고 작업이 거절되는 경우도 있다.

최대 풀 사이즈까지 커지고 (위 상황에서는 4개)

작업 대기 큐가 가득차면 (위 상황에서는 2개)

새로 들어온 작업을 작업 대기 큐가 받아줄 수도 없고, 스레드가 받아줄 수도 없어서 거절된다.

위 예시에서는 7번째 작업이 들어올 때는 받아줄 스레드도 없고, 작업 큐도 다 차서 `RejectedExecutionException`이 터진다. (RuntimeException)

 

스레드 미리 생성해두기

지금까지 배운 내용은 처음 작업이 들어왔을 때, 스레드를 생성하는 방식이었다. 하지만 웹서버가 미리 응답을 대기하고 있어야 한다던지, 어떤 작업이 실제로 진행되기 전에 미리 스레드를 만드는 것이 좋은 상황도 있다.

구체적인 예시로, 선착순 이벤트 같은걸 하면 특정 시간에 작업이 확 몰리기 때문에 그 전에 미리 만들어두는 것이 좋을 것 같다.

 

`ThreadPoolExecutor`의 `prestartAllCoreThreads()`를 사용하면 정해진 코어 스레드 (기본 스레드 수) 만큼 스레드를 생성한다. (공식 문서: Starts all core threads, causing them to idly wait for work.)

`ExecutorService`의 기본 메서드가 아닌, `ThreadPoolExecutor`만의 메서드이기 때문에 `ThreadPoolExecutor`로 캐스팅해줘야 한다.

public static void main(String[] args) {
    ExecutorService es = Executors.newFixedThreadPool(1000);
    ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) es;
    poolExecutor.prestartAllCoreThreads(); // 1000개 스레드 모두 미리 생성
}

 

스레드 풀 전략

고정풀 전략

지금까지 편하게 자주 사용했던 `Executors.newFixedThreadPool()`을 사용하는 방식이 고정풀 전략이다.
아래 두 라인은 완전히 동일한 역할을 한다. (메서드 안을 들어가보면 생성자를 호출하는 방식으로 작동)

Executors.newFixedThreadPool(2);
new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

 

장점: 안정적인 서버 부하, 처리량 - 예측 가능성 ↑

단점: 예측 불가능한 비즈니스 상황에서는 오히려 예측 가능성 높은 방식이 나쁠 수 있다.

엄청 빠르게 성장하고 있는 서비스인 경우, 내일 아니면 당장 한시간 뒤라도 인바운드 트래픽이 얼마나 증가할지 예측할 수 없다. 갑자기 트래픽이 폭증했을 때는 서버가 더 많은 부하를 받아내도록 유연한 스레드 풀을 사용하는 것이 나을 수 있다.

 

고정풀 전략은 크기 제한이 없는 `LinkedBlockingQueue`를 사용하기 때문에 사용자 요청이 무한히 쌓일 수 있는 위험도 있다. 작업을 처리할 수 있는 스레드가 2개로 고정되어, 처리 속도보다 더 높은 속도로 추가 작업이 들어오면 서버 응답 시간이 계속해서 증가할 것이다.

 

문제가 있는지 확인하는 방법: 대기큐에 쌓인 작업 수를 모니터링. 작업이 계속 쌓이면 서버가 더 많은 요청을 수용할 수 있어야 함.

 

캐시풀 전략

작업을 대기하는 공간도 없고, 기본 코어 스레드 풀 크기도 제한이 없는 전략이다.

아래 두 라인은 완전히 동일한 역할을 한다.

Executors.newCachedThreadPool();
new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());

 

`SynchrononousQueue`는 소비자와 생산자가 직접 거래하는 방식이다. 대기 큐가 없는 방식. [Java] 컬렉션 프레임워크와 동시성 참고.

 

작업 대기할 공간이 없어? → 작업이 들어오자마자 스레드 새로 추가 (`Integer.MAX_VALUE`는 사실상 무제한 스레드 생성 가능)

 

작업 들어올 때마다 새 스레드가 처리해주면 갑자기 증가한 트래픽에도 문제없이 대응해줄 수 있을까?

아니다. 서버가 너무 많은 스레드에 잠식 당해서 거의 다운된다. 메모리도 거의 다 사용하는 상황이 발생할 수 있다.

 

고정풀 전략은 아주 느리게라도 어쨋든 작업을 처리해줄 수 있지만

캐시풀 전략은 선착순 이벤트 등으로 순간적으로 스레드가 많이 생성되면 서버가 다운되어 버릴 수도 있다.

 

문제가 있는지 확인하는 방법: 스레드 풀의 스레드 수를 모니터링. 서버가 감당할 수 있는 스레드 수를 넘어서면 스레드 관리하는 비용에 의해 서버가 뻗어버릴 수 있음.

 

사용자 정의 풀 전략

백엔드 엔지니어가 전략을 직접 설계할 수도 있다.

전략을 세우는데에는 크게 세 가지 상황을 고려해서 만들 수 있다.

• 평소: corePoolSize. 평소 이 정도 트래픽은 들어온다에 대응
• 긴급: maximumPoolSize. 트래픽이 갑자기 몰렸을 때 급한 불을 끌 스레드 풀 크기
• 거절: BlockingQueue<> 크기 제한. 서버가 다운되지 않고 수용 가능한 요청까지만 받도록 요청 제한

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

 

위와 같이 스레드 풀을 생성한다면, 

평소: 100개 정도 스레드면 충분히 작업을 처리할 수 있다.

긴급: 빨리 작업 대기 큐를 비워줘야 할 때는 100개까지 추가 스레드를 사용할 수 있다.

거절: 작업 1200개 초과시 거절 (스레드에서 실행 중 200개 + 대기 큐 1000개)

거절 → 사용자에게 "나중에 다시 시도해주세요." 메시지 넘기기같이

 

이런 숫자들은 직접 부하테스트를 통해서 어느 정도가 안정적일지 확인할 필요가 있다.

작업 실행에 드는 비용이나 서버 스펙 등 상황이 매번 다르다.

 

Executor 예외 정책

위에서 스레드 풀이 작업을 받는 것을 "거절"하는 경우도 있다는 것을 배웠는데, 반드시 `RejectedExecutionException`을 터뜨리는 것은 아니다. 우리가 거절시 어떤 행위를 할 것인지, 예외 정책을 조절할 수 있다.

• AbortPolicy: `RejectedExecutionException`을 터뜨리도록 하는 기본 정책
• DiscardPolicy: 새 작업을 조용히 버린다
• CallerRunsPolicy: 새 작업을 제출한 스레드가 대신해서 직접 작업을 실행한다.
• 사용자 정의 (`RejectedExecutionHandler`): `RejectedExecutionHandler`를 직접 구현하면 된다.

기본 예외 정책 말고, 다른 예외 정책을 사용하고 싶다면, `ThreadPoolExecutor` 생성자의 6번째 매개변수에 `RejectedExecutionHandler`를 넘겨주면 된다.

ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
        new SynchronousQueue<>(), new ThreadPoolExecutor.CallerRunsPolicy());

 

이렇게 정의한 예외 정책은, ExecutorService `shutdown()`시에 새 작업을 차단할 때도 똑같이 적용된다.

 

단, `CallarRunsPolicy`는 shutdown시에 작업을 받지 않도록 `!e.isShutdown()`을 확인한다.

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    if (!e.isShutdown())
        r.run();
}

 

• 사용자 정의 정책

static class MyRejectedExecutionHandler implements RejectedExecutionHandler {
    static AtomicInteger count = new AtomicInteger(0);
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        int i = count.incrementAndGet();
        log("[경고] 거절된 누적 작업 수: " + i);
    }
}

 

`RejectedExecutionHandler`를 구현하고 그 객체를 사용할 수도 있는데, 위와 같이 거절된 작업 수를 어플리케이션 로그로 모니터링하는 시스템을 여기에 넣을 수도 있다.

 

거절은 나쁜거 아니야? 할 수도 있지만 서버가 다운되지 않는게 제일 중요하기 때문에 적절한 거절이 있어야 한다. 대신 그런 상황이 발생하면 인지할 수 있도록 모니터링을 하자.

 

 

실무에서 전략 선정하기

일반적으로는 다음과 같이 정하면 된다고 한다.

• 고정 스레드 풀 전략: 트래픽이 일정하고, 시스템 안전성이 가장 중요할 때
• 캐시 스레드 풀 전략: 일반적인 성장하는 서비스
• 사용자 정의 풀 전략: 다양한 상황에 대응

시스템을 잘 최적화하는 것도 좋지만, 사실 우리 인건비가 제일 비싸다.

예측할 수 없는 미래까지 다 고려해서 최적화하는 것은 오버 엔지니어링이 될 수 있다.

현재 상황에 최적화를 하자.

 

일반적인 상황에서는 고정/캐시 스레드 풀 전략이면 충분하고 일반적인 상황을 벗어날 정도로 서비스가 너무 잘 되면 그 때 더 나은 최적화를 하는 것이 좋다. 그래서 제일 중요한 것은 현재 시스템을 잘 모니터링할 수 있는 시스템을 구축하는 것이다.