스레드 풀의 작업 거부 정책

앞 글에서 ThreadPoolExecutor는 코어 스레드, 큐, 추가 스레드 순서로 새 작업의 자리를 찾고, 그래도 자리가 없으면 작업을 거부한다는 것을 보았습니다. 거부는 풀이 고장 나서 생기는 일이 아닙니다. 스레드 수와 큐 용량에 한도를 정했기 때문에 생기는, 예정된 마지막 단계입니다. 한도가 없다면 거부도 없겠지만, 그 대신 메모리나 데이터베이스 연결 같은 다른 자원이 먼저 바닥납니다. 이미 shutdown()을 호출한 풀에 작업을 제출할 때도 거부가 일어납니다.

거부가 일어나면 풀은 거부 처리기(RejectedExecutionHandler)를 호출하고, 그 작업을 어떻게 할지는 처리기가 정합니다. 예외를 던질 수도 있고, 제출한 스레드가 직접 실행하게 할 수도 있고, 조용히 버릴 수도 있습니다. 어느 쪽이 맞는지는 코드보다 업무가 정합니다. 이 작업을 잃어도 되는지, 제출하는 쪽을 느리게 만들어도 되는지, 다시 시도해도 같은 일이 두 번 일어나지 않는지부터 따져야 합니다. 결제나 예약처럼 잃으면 안 되는 요청과, 화면 갱신처럼 가장 최근 것 하나만 있으면 되는 작업에 같은 정책을 쓸 수는 없습니다.

거부 예외를 처리하지 않으면 생기는 일

거부 처리기를 따로 정하지 않으면 AbortPolicy가 쓰입니다. 이 정책은 거부된 작업을 제출한 쪽에 RejectedExecutionException을 던집니다. 아래 예제는 스레드가 1개뿐이고 작업을 저장하지 않는 SynchronousQueue를 쓰는 풀에서, 첫 작업이 래치에서 기다리는 동안 두 번째 작업을 제출합니다.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class UnhandledRejection {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch release = new CountDownLatch(1);
        var executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>());
        executor.execute(() -> await(release));
        executor.execute(() -> System.out.println("이 줄은 제출되지 않는다"));
        release.countDown();
        executor.shutdown();
    }

    private static void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

표준 출력에는 아무것도 나오지 않고, 표준 오류에 다음 내용이 출력된 뒤 프로그램이 멈춘 채 끝나지 않습니다. 스택 트레이스의 맨 아래 줄은 두 번째 execute를 호출한 12번째 줄이고, 그 위의 세 줄은 거부를 처리한 JDK 내부 코드입니다.

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task UnhandledRejection$$Lambda/0x00007f7d68000c30@5e481248 rejected from java.util.concurrent.ThreadPoolExecutor@4554617c[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
	at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2081)
	at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:841)
	at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1376)
	at UnhandledRejection.main(UnhandledRejection.java:12)

두 번째 execute를 호출하는 순간 하나뿐인 스레드는 래치에서 기다리는 중이고, SynchronousQueue에는 작업을 맡겨 둘 칸이 없습니다. 최대 스레드 수도 1이라 더 만들 수 없으므로 작업은 거부되고, AbortPolicy가 던진 예외가 main을 빠져나갑니다. 그래서 뒤에 있는 release.countDown()과 shutdown()은 실행되지 않습니다.

프로그램이 끝나지 않는 이유도 여기에 있습니다. 기본 스레드 팩터리가 만든 작업자 스레드는 데몬 스레드가 아니어서, main이 예외로 끝난 뒤에도 래치에서 계속 기다리며 JVM이 종료되지 못하게 붙잡습니다. 겉보기에는 예외로 끝난 것 같아도 프로세스는 살아 있으므로 Ctrl+C로 멈춰야 합니다. 예외 메시지의 대괄호 부분에는 거부 순간의 풀 상태가 들어 있는데, 스레드 1개가 일하는 중이고 큐에 기다리는 작업이 없으니 스레드가 모자라서 거부되었다는 것을 알 수 있습니다. Lambda/0x...와 @ 뒤의 16진수는 실행할 때마다 달라지고, ThreadPoolExecutor.java의 줄 번호는 JDK 버전에 따라 다릅니다.

이 예제는 실패를 일부러 드러낸 것입니다. 실제 서비스에서 이 예외를 제출하는 자리에서 잡지 않으면, 웹 요청을 처리하던 코드나 배치 작업을 조율하는 스레드가 예상하지 못한 곳에서 중단될 수 있습니다. 반대로 예외를 잡아서 조용히 무시하면 호출한 쪽은 작업이 접수되었다고 믿게 됩니다. 그래서 거부는 제출하는 자리에서 잡아 '지금은 처리할 수 없음'이라는 업무 오류나, 나중에 다시 시도하라는 응답(예를 들어 HTTP 503)으로 바꿔 주어야 합니다.

네 가지 기본 거부 정책

ThreadPoolExecutor에는 미리 만들어 둔 거부 정책 네 가지가 중첩 클래스로 들어 있습니다. 같은 정책이라도 포화되어 거부할 때와 종료된 뒤에 거부할 때 하는 일이 다를 수 있다는 점에 주의해야 합니다.

AbortPolicy는 두 경우 모두 예외를 던지므로, 제출한 쪽이 거부를 모르고 지나갈 수 없습니다. CallerRunsPolicy는 풀이 포화되었을 때 execute를 호출한 스레드가 그 작업을 직접 실행합니다. 하지만 풀이 종료된 뒤에는 작업을 실행하지 않고 그냥 버립니다. DiscardPolicy는 두 경우 모두 작업을 조용히 버리며, 공식 문서는 이 정책이 작업 완료에 전혀 기대지 않는 드문 경우를 위한 것이라고 설명합니다.

DiscardOldestPolicy는 풀이 종료되지 않았다면 큐 맨 앞의 작업을 버리고 새 작업으로 execute를 다시 시도합니다. 다시 시도해도 실패하면 이 과정을 되풀이하고, 풀이 종료된 뒤에는 새 작업을 버립니다. 큐 맨 앞이 가장 오래 기다린 작업이라는 가정은 먼저 들어온 것이 먼저 나가는(FIFO) 큐에서만 맞습니다. 우선순위로 정렬하는 큐라면 맨 앞은 가장 급한 작업이므로, 이 정책은 가장 중요한 작업을 버리게 됩니다. 공식 문서도 이 정책을 받아들일 만한 경우는 드물다고 적고 있습니다.

네 정책 가운데 AbortPolicy를 뺀 셋은 작업을 실행하지 않았더라도 execute가 정상적으로 반환할 수 있습니다. 예외가 나지 않았다는 사실만으로는 작업이 접수되었는지 알 수 없다는 뜻입니다. 특히 CallerRunsPolicy는 포화 때 작업을 실행해 주므로 '버리지 않는 정책'처럼 보이지만, 종료 뒤에는 아무 표시 없이 버립니다.

버린 작업의 Future는 끝나지 않는다

작업을 execute가 아니라 submit으로 제출했다면 버리는 정책은 더 곤란한 문제를 만듭니다. submit은 작업을 FutureTask로 감싸 execute에 넘기고, 그 Future를 돌려줍니다. 그런데 DiscardPolicy는 넘겨받은 FutureTask를 실행하지도 취소하지도 않고 버리므로, 이 Future는 영원히 완료되지 않습니다. 다음 예제로 확인해 보겠습니다.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class DiscardedFuture {
    public static void main(String[] args) throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        var executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>(), new ThreadPoolExecutor.DiscardPolicy());
        executor.execute(() -> {
            try { release.await(); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        });
        Future<String> future = executor.submit(() -> "result");
        try {
            System.out.println(future.get(200, TimeUnit.MILLISECONDS));
        } catch (TimeoutException e) {
            System.out.println("timeout, done=" + future.isDone());
        }
        release.countDown();
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("after shutdown, done=" + future.isDone());
    }
}

실행 결과는 다음과 같습니다. 첫 줄은 200밀리초 안에 결과가 오지 않아 catch에서 출력한 것이고, 둘째 줄은 풀을 종료하고 기다린 뒤에 다시 확인한 Future의 상태입니다.

timeout, done=false
after shutdown, done=false

get에 200밀리초 제한을 주었기 때문에 TimeoutException으로 빠져나왔지만, 제한 없이 get()을 호출했다면 main은 영원히 기다렸을 것입니다. 풀을 종료하고 종료가 끝나기를 기다린 뒤에도 isDone()은 여전히 false입니다. 버려진 작업은 큐에도 들어간 적이 없으므로 shutdownNow()가 돌려주는 미실행 작업 목록에도 나오지 않습니다.

공식 문서는 DiscardOldestPolicy를 설명하면서, 작업을 버릴 때는 대부분의 경우 그 작업을 취소해 완료를 기다리는 쪽이 예외를 받게 하거나 실패를 기록하라고 권합니다. 같은 문제는 DiscardPolicy에도 그대로 있습니다. 거부된 작업이 Future라면 취소하는 처리기를 만들어 넘기면 됩니다.

RejectedExecutionHandler cancelOnReject = (task, executor) -> {
    if (task instanceof Future<?> future) {
        future.cancel(false);
    }
};

이 처리기를 넘긴 풀에서는 버려진 작업의 get()이 기다리지 않고 바로 CancellationException을 던지고, isDone()도 true가 됩니다. 실제로는 여기에 거부 횟수를 세거나 로그를 남기는 코드를 더해서, 작업을 버렸다는 사실이 어딘가에 드러나게 해야 합니다.

거부를 결과 값으로 돌려주기

거부 예외를 제출하는 자리에서 처리하는 한 가지 방법은 예외를 결과 값으로 바꿔 돌려주는 것입니다. 아래 ExplicitAdmission의 submit은 작업을 받아들이면 ACCEPTED, 풀이 꽉 차 있으면 SATURATED, 풀이 종료되었으면 CLOSED를 돌려줍니다. 호출하는 쪽은 RejectedExecutionException이라는 구현 세부를 몰라도 되고, 결과에 따라 다시 시도할지, 바로 실패를 알릴지, 다른 곳에 저장할지를 고르면 됩니다.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class ExplicitAdmission implements AutoCloseable {
    enum Admission { ACCEPTED, SATURATED, CLOSED }

    private final ThreadPoolExecutor pool = new ThreadPoolExecutor(
            2, 4, 10, TimeUnit.SECONDS, new ArrayBlockingQueue<>(8));

    Admission submit(Runnable task) {
        if (pool.isShutdown()) return Admission.CLOSED;
        try {
            pool.execute(task);
            return Admission.ACCEPTED;
        } catch (RejectedExecutionException rejected) {
            return pool.isShutdown() ? Admission.CLOSED : Admission.SATURATED;
        }
    }

    @Override
    public void close() throws InterruptedException {
        pool.shutdown();
        if (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }
    }

    public static void main(String[] args) throws Exception {
        try (var admission = new ExplicitAdmission()) {
            System.out.println(admission.submit(() -> System.out.println("accepted")));
        }
    }
}

한 번 실행한 결과입니다. 대소문자만 다른 두 단어가 출력되는데, 하나는 작업이 출력한 것이고 다른 하나는 submit의 반환값입니다.

accepted
ACCEPTED

ACCEPTED는 풀이 작업을 받아들였다는 뜻일 뿐, 작업이 끝났다는 뜻이 아닙니다. 작업이 출력하는 accepted와 main이 출력하는 ACCEPTED는 서로 다른 스레드에서 나오므로 순서가 바뀔 수 있습니다. 이 환경에서 20번 실행했을 때는 모두 accepted가 먼저 나왔지만, 그렇다고 순서가 정해진 것은 아닙니다.

SATURATED와 CLOSED를 나누는 이유는 대응이 다르기 때문입니다. 포화는 지금 자리가 없다는 뜻이므로 잠시 뒤에 다시 시도해 볼 수 있습니다. 이때는 재시도 횟수의 상한과 시도할 때마다 길어지는 대기 시간이 필요하고, 같은 요청이 여러 번 들어와도 한 번만 처리되게 하는 멱등성(idempotency) 키도 있어야 합니다. 종료는 이 실행기의 수명이 끝났다는 뜻이므로 다시 시도해도 소용이 없고, 다른 실행기로 보내거나 요청을 취소해야 합니다.

이 분류가 완벽하지는 않습니다. 예외를 잡은 뒤에 isShutdown()을 확인하므로, 포화 때문에 거부된 직후 다른 스레드가 풀을 종료하면 CLOSED로 분류됩니다. 결과가 거부의 원인을 정확히 말해 주지는 않는다는 점을 알고 써야 합니다. close()도 단순하게 만들었습니다. 1초 안에 종료되지 않으면 shutdownNow()를 요청하지만 그 뒤에 종료되었는지는 다시 확인하지 않고, 기다리는 동안 인터럽트되면 InterruptedException을 그대로 던집니다. 종료 절차를 제대로 갖추는 방법은 실행기 종료를 다룬 글에서 볼 수 있습니다.

CallerRunsPolicy로 제출 속도 늦추기

CallerRunsPolicy는 풀이 포화되면 작업을 제출한 스레드에게 그 작업을 직접 실행시킵니다. 제출자가 작업을 실행하는 동안에는 다음 작업을 제출할 수 없으므로, 작업이 들어오는 속도가 저절로 처리 속도 쪽으로 내려갑니다. 이처럼 처리하는 쪽이 밀릴 때 그 압력을 보내는 쪽으로 되돌려 속도를 늦추게 하는 것을 역압(backpressure)이라고 합니다. 버퍼를 끝없이 늘리지 않고도 과부하를 조절할 수 있다는 것이 이 정책의 장점입니다.

다음 예제는 스레드 1개, 큐 1칸인 풀에 30밀리초짜리 작업 5개를 제출하면서, execute 호출 하나가 얼마나 걸렸는지를 마이크로초 단위로 출력합니다.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class CallerRunsBackpressure {
    public static void main(String[] args) throws InterruptedException {
        var executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int task = 1; task <= 5; task++) {
            int id = task;
            long before = System.nanoTime();
            executor.execute(() -> {
                sleep(30);
                System.out.printf("task=%d worker=%s%n", id,
                        Thread.currentThread().getName());
            });
            long submitMicros = (System.nanoTime() - before) / 1_000;
            System.out.printf("submit=%d latencyMicros=%d%n", id, submitMicros);
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

한 번 실행한 결과입니다. submit= 줄은 execute 한 번에 걸린 시간을, task= 줄은 작업을 실행한 스레드 이름을 보여 줍니다. worker=main인 줄과 그 바로 뒤에 나오는 latencyMicros 값을 함께 보세요.

submit=1 latencyMicros=6378
submit=2 latencyMicros=23
task=1 worker=pool-1-thread-1
task=3 worker=main
submit=3 latencyMicros=30614
submit=4 latencyMicros=15
task=2 worker=pool-1-thread-1
task=5 worker=main
submit=5 latencyMicros=30679
task=4 worker=pool-1-thread-1

첫 번째 제출은 풀에 스레드가 없으므로 스레드를 새로 만들어 작업 1을 맡깁니다. 스레드를 만들어 시작하는 비용과 처음 실행되는 코드의 준비 비용이 더해져 수 밀리초가 걸렸습니다. 두 번째 제출은 큐의 빈칸에 들어가므로 수십 마이크로초 만에 끝납니다. 세 번째 제출 때는 스레드가 작업 1을 실행하는 중이고 큐도 찼으며 최대 스레드 수도 1이라 거부됩니다. 그래서 main이 작업 3을 직접 실행했고, submit=3의 지연이 작업 시간인 30밀리초 남짓이 되었습니다.

main이 작업 3을 실행하는 30밀리초 동안 풀의 스레드는 작업 1을 끝내고 큐에서 작업 2를 꺼내 갔습니다. 큐가 비었으니 네 번째 제출은 다시 빠르게 큐에 들어가고, 다섯 번째 제출은 큐가 또 차 있어서 main이 직접 실행합니다. 결국 main은 작업 두 개를 직접 처리하는 동안 새 작업을 제출하지 못했고, 그만큼 제출 속도가 풀의 처리 속도에 맞춰졌습니다.

이 출력은 실행할 때마다 달라집니다. 12번 실행했을 때 main이 작업 3과 5를 실행한 경우가 9번, 작업 3만 실행한 경우가 1번, 작업 4만 실행한 경우가 2번이었습니다. 지연 수치와 줄의 순서도 매번 조금씩 다릅니다. 어떤 작업이 호출자에게 넘어올지는 제출하는 순간 스레드와 큐가 비어 있는지에 따라 정해지기 때문입니다.

CallerRunsPolicy는 제출하는 스레드가 잠시 멈춰도 괜찮을 때만 써야 합니다. HTTP 연결을 받아들이는 스레드라면 그동안 새 연결을 받지 못하고, 스레드 하나가 준비된 연결을 차례로 처리하는 이벤트 루프라면 그 루프가 맡은 모든 요청이 멈춥니다. 제출자가 락을 쥔 채로 execute를 호출했다면, 작업이 끝날 때까지 그 락을 계속 쥐고 있게 되어 같은 락을 기다리는 다른 스레드가 모두 늦어집니다.

제출하는 스레드가 여럿이라는 점도 기억해야 합니다. 제출자마다 작업을 직접 실행할 수 있으므로, 동시에 실행되는 작업 수는 풀의 최대 스레드 수에 제출자 수를 더한 만큼까지 늘어날 수 있습니다. 풀의 최대 스레드 수로 데이터베이스 같은 외부 자원의 동시 사용을 제한하고 있었다면, 이 정책이 그 제한을 넘게 만듭니다.

예외가 전달되는 곳도 달라집니다. 풀의 스레드가 실행한 Runnable이 던진 예외는 그 작업자 스레드의 잡히지 않은 예외의 처리기로 가지만, 호출자가 직접 실행한 작업이 던진 예외는 execute 호출을 빠져나와 제출자에게 전달됩니다. 이 환경에서 포화된 풀에 IllegalStateException을 던지는 작업을 제출해 보니, execute를 감싼 try에서 그 예외가 잡혔습니다. 그리고 앞의 표에서 본 것처럼, 풀이 종료된 뒤에는 이 정책도 작업을 조용히 버립니다.

직접 만든 거부 정책: 횟수를 세고 따로 보관하기

업무상 버릴 수 없는 요청이라면 거부 처리기를 직접 만들어, 거부 횟수를 기록하고 작업을 다른 저장소로 넘길 수 있습니다. RejectedExecutionHandler는 rejectedExecution 메서드 하나만 가진 인터페이스이고, 풀은 거부할 때마다 이 메서드에 거부된 작업과 풀 자신을 넘깁니다. 이 메서드는 execute를 호출한 스레드에서 실행된다는 점이 중요합니다. 처리기가 오래 걸리면 그동안 제출자도 함께 멈춥니다.

아래 예제는 학습을 위해 거부된 작업을 메모리 큐에 보관합니다. 운영에서는 크기가 정해진 메시지 브로커, 데이터베이스의 outbox 테이블(보낼 작업을 업무 데이터와 함께 저장해 두는 테이블), 명시적인 실패 응답처럼 장애가 났을 때 어떻게 되는지 분명한 수단을 골라야 합니다.

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

public final class MeasuredRejection {
    static final class OverflowHandler implements RejectedExecutionHandler {
        private final LongAdder rejected = new LongAdder();
        private final Queue<Runnable> overflow;

        OverflowHandler(Queue<Runnable> overflow) {
            this.overflow = overflow;
        }

        @Override
        public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
            rejected.increment();
            if (executor.isShutdown() || !overflow.offer(task)) {
                throw new RejectedExecutionException("대체 경로도 작업을 받지 못했습니다");
            }
        }

        long rejectedCount() {
            return rejected.sum();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Queue<Runnable> overflow = new ConcurrentLinkedQueue<>();
        var handler = new OverflowHandler(overflow);
        var executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new SynchronousQueue<>(), handler);
        executor.execute(() -> sleep(40));
        executor.execute(() -> System.out.println("overflow 실행 후보"));
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
        System.out.printf("rejected=%d saved=%d%n",
                handler.rejectedCount(), overflow.size());
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

실행 결과는 다음과 같습니다. rejected는 처리기가 거부된 작업을 받은 횟수이고, saved는 그 작업을 보관한 overflow 큐의 크기입니다. 두 값이 같다면 거부된 작업이 하나도 사라지지 않은 것입니다.

rejected=1 saved=1

첫 작업은 40밀리초 동안 잠들어 있고 두 번째 execute는 곧바로 이어서 호출되므로, 그 순간에는 작업을 넘겨받을 스레드가 없습니다. 그래서 두 번째 작업은 거부되어 처리기로 가고, 처리기는 횟수를 1 늘린 뒤 작업을 overflow에 넣습니다. 10번 실행해서 모두 이 결과가 나왔습니다.

다만 sleep(40)이 두 제출의 순서를 정해 주는 것은 아닙니다. 두 execute 사이에서 main이 40밀리초 넘게 멈추면, 첫 작업을 끝낸 스레드가 다음 작업을 기다리고 있다가 두 번째 작업을 넘겨받아 실행합니다. 그러면 overflow 실행 후보가 출력되고 rejected=0 saved=0이 나옵니다. 도표의 두 경로가 바로 이 두 경우입니다.

어느 경로든 overflow에 들어간 작업은 실행되지 않습니다. 이 코드에는 overflow에서 작업을 꺼내 다시 풀에 넣는 부분이 없기 때문입니다. 즉 saved=1은 작업을 보관했다는 뜻이지 처리했다는 뜻이 아닙니다. awaitTermination의 반환값도 확인하지 않으므로, 이 출력만으로 풀이 종료되었다고 말할 수도 없습니다.

처리기는 풀이 이미 종료되었거나 overflow가 작업을 받지 못하면 RejectedExecutionException을 던집니다. RejectedExecutionHandler 문서는 다른 방법이 없을 때 처리기가 이 비검사 예외를 던질 수 있고, 그 예외는 execute를 호출한 쪽으로 전달된다고 설명합니다. 같은 예외 타입을 쓰면 호출하는 쪽은 기본 정책을 쓸 때와 같은 방식으로 실패를 처리할 수 있습니다. 예제의 ConcurrentLinkedQueue는 크기 제한이 없어서 offer가 실패하지 않지만, 처리기는 Queue 타입으로 받으므로 용량이 정해진 큐를 넘기면 그 분기가 쓰입니다.

메모리 큐인 overflow는 크기 제한도 없고, 프로세스가 죽으면 내용이 사라집니다. 이름은 대체 저장소이지만 운영에서 믿을 수 있는 저장소는 아닌 셈입니다. 운영에서는 거부 횟수만 보지 말고 큐 사용률, 활성 스레드 수, 제출 지연, 완료 처리량을 같은 시간 축에 놓고 봅니다. 거부가 늘 때 처리량도 함께 늘었다면 잠깐 몰린 요청일 가능성이 크고, 처리량은 떨어지는데 거부가 늘면 외부 I/O가 막혔거나 작업 시간이 길어진 것을 의심해야 합니다. 알림은 거부 한 건이 아니라 거부가 얼마나 오래, 어떤 비율로 이어지는지에 걸어야 불필요한 알림이 줄어듭니다.

더 깊이: 가득 찼을 때 제출자를 기다리게 하고 싶다면

CallerRunsPolicy처럼 제출자에게 일을 시키는 대신, 자리가 날 때까지 제출자를 그냥 기다리게 하고 싶을 때도 있습니다. 기본 정책에는 그런 것이 없어서 처리기 안에서 executor.getQueue().put(task)를 호출하는 방법이 종종 쓰이지만, 권하지 않습니다. 공식 문서는 getQueue()가 주로 디버깅과 모니터링을 위한 것이라고 밝히고 있고, 큐에 직접 넣으면 execute가 하는 종료 확인과 스레드 생성 판단을 모두 건너뜁니다. 그래서 풀이 종료를 시작한 뒤에 들어간 작업이 실행된다는 보장이 없습니다.

더 안전한 방법은 제출하기 전에 세마포어로 자리를 확보하는 것입니다. 허가 수를 '최대 스레드 수 + 큐 용량'으로 정한 Semaphore에서 제출 전에 acquire()하고, 작업이 끝날 때 finally에서 release()합니다. 제출이 거부되었다면 그 자리에서 바로 허가를 돌려주어야 합니다. 『Java Concurrency in Practice』도 execute를 기다리게 만드는 기본 정책은 없으며, 세마포어로 작업이 들어오는 속도를 제한하면 같은 효과를 얻을 수 있다고 설명합니다. Semaphore의 사용법은 가상 스레드와 ScopedValue에서 볼 수 있습니다.

작업 성격에 따라 정책 고르기

지금까지 본 정책을 작업의 성격에 따라 정리하면 다음과 같습니다.

정책을 정했다면 풀을 실제로 포화시켜 시험해야 합니다. 거부 예외가 어느 계층까지 올라가는지, 호출자 실행이 어느 스레드에서 일어나는지, 종료한 뒤의 제출이 호출하는 쪽에 어떻게 보이는지를 확인합니다. 거부가 한 번도 일어나지 않은 부하 시험은 용량 설계의 절반만 확인한 것입니다.

정리

• 거부는 스레드와 큐가 모두 찼을 때, 그리고 종료된 풀에 제출했을 때 일어납니다. 풀은 이때 RejectedExecutionHandler를 호출하고, 처리기는 execute를 호출한 스레드에서 실행됩니다.
• 기본값인 AbortPolicy는 RejectedExecutionException을 던집니다. 제출하는 자리에서 잡아 업무 오류나 재시도 가능한 응답으로 바꿉니다.
• CallerRunsPolicy, DiscardPolicy, DiscardOldestPolicy는 작업을 실행하지 않고도 execute가 정상 반환할 수 있습니다. submit으로 낸 작업을 버리면 그 Future는 끝나지 않으므로 취소해 주어야 합니다.
• CallerRunsPolicy는 제출자에게 작업을 실행시켜 역압을 만들지만, 제출자가 멈춰도 될 때만 쓸 수 있고 종료 뒤에는 작업을 버립니다.
• 직접 만든 처리기로 거부를 세고 다른 곳에 보관할 수 있지만, 보관은 처리가 아닙니다. 대체 저장소의 용량과 내구성, 다시 실행하는 경로를 따로 설계해야 합니다.

연습 문제

1. 등급별로 다르게 거부하는 제출 API

CRITICAL, INTERACTIVE, BEST_EFFORT 세 등급을 받는 제출 API를 설계하세요. 풀이 포화되면 중요 작업은 호출자에게 실패를 알리고, 상호작용 작업은 한 번만 다시 시도하며, 보조 작업은 버리되 버린 개수를 셉니다. 실행기 종료와 단순 포화는 서로 다른 결과로 표현해야 합니다.

import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.LongAdder;

public final class AdmissionPolicySolution {
    enum Grade { CRITICAL, INTERACTIVE, BEST_EFFORT }
    enum Result { ACCEPTED, RETRY_LATER, DROPPED, CLOSED }

    private final ThreadPoolExecutor pool;
    private final LongAdder dropped = new LongAdder();

    AdmissionPolicySolution(ThreadPoolExecutor pool) {
        this.pool = pool;
    }

    Result submit(Grade grade, Runnable task) {
        if (pool.isShutdown()) return Result.CLOSED;
        try {
            pool.execute(task);
            return Result.ACCEPTED;
        } catch (RejectedExecutionException rejected) {
            if (pool.isShutdown()) return Result.CLOSED;
            return switch (grade) {
                case CRITICAL, INTERACTIVE -> Result.RETRY_LATER;
                case BEST_EFFORT -> {
                    dropped.increment();
                    yield Result.DROPPED;
                }
            };
        }
    }

    long droppedCount() {
        return dropped.sum();
    }
}

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class AdmissionPolicyCallerSolution {
    enum Decision { ACCEPTED, FAILED, DROPPED, CLOSED }

    static Decision call(AdmissionPolicySolution api,
                         AdmissionPolicySolution.Grade grade,
                         Runnable task) throws InterruptedException {
        var first = api.submit(grade, task);
        if (first == AdmissionPolicySolution.Result.ACCEPTED) return Decision.ACCEPTED;
        if (first == AdmissionPolicySolution.Result.CLOSED) return Decision.CLOSED;
        if (first == AdmissionPolicySolution.Result.DROPPED) return Decision.DROPPED;
        if (grade != AdmissionPolicySolution.Grade.INTERACTIVE) return Decision.FAILED;

        Thread.sleep(10);
        var retry = api.submit(grade, task); // 최대 한 번의 재제출
        return switch (retry) {
            case ACCEPTED -> Decision.ACCEPTED;
            case CLOSED -> Decision.CLOSED;
            case DROPPED -> Decision.DROPPED;
            case RETRY_LATER -> Decision.FAILED;
        };
    }

    public static void main(String[] args) throws InterruptedException {
        var entered = new CountDownLatch(1);
        var release = new CountDownLatch(1);
        var pool = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1)); // 기본 AbortPolicy
        var api = new AdmissionPolicySolution(pool);
        try {
            pool.execute(() -> {
                entered.countDown();
                try { release.await(); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
            if (!entered.await(1, TimeUnit.SECONDS)) {
                throw new IllegalStateException("worker did not enter");
            }
            pool.execute(() -> {}); // 실행 작업자는 막혀 있으므로 큐 한 칸을 점유
            Runnable target = () -> System.out.println("target-ran");
            System.out.println(call(api, AdmissionPolicySolution.Grade.CRITICAL, target));
            System.out.println(call(api, AdmissionPolicySolution.Grade.INTERACTIVE, target));
            System.out.println(call(api, AdmissionPolicySolution.Grade.BEST_EFFORT, target));
            System.out.println("dropped=" + api.droppedCount());
            pool.shutdown();
            System.out.println(call(api, AdmissionPolicySolution.Grade.INTERACTIVE, target));
        } finally {
            release.countDown();
            pool.shutdownNow();
            if (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
                System.err.println("pool still running");
            }
        }
    }
}

FAILED
FAILED
DROPPED
dropped=1
CLOSED

2. 종료된 풀에 submit하면

CallerRunsPolicy를 쓰는 풀을 shutdown()한 뒤 submit(() -> "result")를 호출했습니다. submit은 예외를 던질까요? 돌려받은 Future의 get()을 호출하면 어떻게 될까요? 거부 정책 표와 버린 작업의 Future 이야기를 함께 떠올려 보세요.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class CallerRunsAfterShutdownSolution {
    public static void main(String[] args) throws Exception {
        var executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.CallerRunsPolicy());
        executor.shutdown();
        Future<String> future = executor.submit(() -> "result");
        System.out.println("submit returned, done=" + future.isDone());
        try {
            System.out.println(future.get(300, TimeUnit.MILLISECONDS));
        } catch (TimeoutException e) {
            System.out.println("timeout, done=" + future.isDone());
        }
    }
}

submit returned, done=false
timeout, done=false

다음 글

이것으로 실행자와 스레드 풀을 다룬 22장을 마칩니다. 다음 장에서는 파일과 네트워크 입출력의 바탕이 되는 문자와 바이트 표현, 그리고 문자 인코딩을 다룹니다.