ExecutorService로 작업 실행하기

지금까지는 동시에 처리할 일이 생기면 Thread를 직접 만들어 start()로 시작하고 join()으로 끝나기를 기다렸습니다. 스레드가 두세 개일 때는 이것으로 충분합니다. 하지만 웹 서버처럼 요청이 들어올 때마다 일을 처리하는 프로그램에서 요청마다 스레드를 새로 만들면 챙길 일이 많아집니다. 스레드를 몇 개까지 만들지, 다 쓴 스레드를 언제 정리할지, 작업이 실패하면 어떻게 알아챌지를 요청을 받는 코드마다 직접 처리해야 합니다.

자바는 이 일을 대신 맡아 주는 도구를 java.util.concurrent 패키지에 마련해 두었습니다. 작업을 넘겨받아 알맞은 스레드에서 실행해 주는 객체를 실행기(executor)라고 하며, 이 장의 제목에 쓴 실행자도 같은 말입니다. 실행기를 쓰면 업무 코드는 “이 작업을 실행해 달라”고 말하기만 하고, 스레드를 몇 개 두고 언제 만들지는 실행기를 만드는 한곳에서 정합니다. 이 글에서는 실행기에 작업을 맡기는 방법과, 작업자 수·큐 용량·종료 절차를 정하는 방법을 배웁니다.

작업은 Runnable 람다로 쓰고, 결과를 모을 때는 AtomicInteger와 동시성 컬렉션을 씁니다. 예제마다 코드 제목에 적힌 클래스 이름으로 .java 파일을 만들어 실행하면 되고, Thread.ofPlatform()을 쓰는 예제는 Java 21 이상이 필요합니다.

요청마다 스레드를 만들면 생기는 문제

다음 프로그램은 요청 5,000개가 한꺼번에 들어왔다고 보고, 요청마다 플랫폼 스레드를 하나씩 만들어 0.1초 동안 자는 작업을 시킵니다. 플랫폼 스레드(platform thread)는 프로세스와 스레드에서 본 것처럼 운영체제 스레드와 하나씩 짝을 이루는 보통의 자바 스레드이고, Thread.ofPlatform().start(작업)은 그런 스레드를 만들어 곧바로 시작합니다. 메모리가 작은 환경에서는 스레드를 만들다 실패할 수 있으니 읽고 넘어가도 괜찮습니다.

import java.util.ArrayList;
import java.util.List;

public final class ThreadPerRequest {
    public static void main(String[] args) {
        List<Thread> threads = new ArrayList<>();
        for (int request = 0; request < 5_000; request++) {
            threads.add(Thread.ofPlatform().start(() -> {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }
        System.out.println("created=" + threads.size());
    }
}

created=5000

created=5000은 스레드 5,000개를 시작하는 데 성공했다는 뜻일 뿐, 5,000개가 동시에 살아 있었다는 뜻은 아닙니다. 루프가 도는 사이에 먼저 시작한 스레드가 0.1초를 다 자고 끝나기도 하기 때문입니다. 또 main은 이 줄을 출력하자마자 끝나지만 JVM은 바로 종료되지 않습니다. 이 스레드들은 데몬이 아닌 사용자 스레드여서, JVM은 마지막 스레드가 끝날 때까지 기다립니다.

비용도 무시할 수 없습니다. 작업마다 0.1초씩 자는 것이 전부인데도, 이 글을 쓴 환경(2코어 Linux, JDK 21)에서는 프로그램이 끝나기까지 1초 안팎이 걸렸습니다. 같은 환경에서 아무 일도 하지 않는 프로그램은 0.05초 정도면 끝나므로, 시간의 대부분은 스레드 5,000개를 만들고 정리하는 데 들어간 셈입니다. 플랫폼 스레드를 하나 만들 때마다 운영체제에 스레드를 요청하고 그 스레드가 쓸 스택 메모리를 따로 마련해야 하기 때문입니다. 운영체제가 허용하는 스레드 수나 메모리를 넘으면 start()가 OutOfMemoryError를 던져 루프가 중간에 멈추고, created 줄은 출력되지도 않습니다.

정리하면 요청마다 스레드를 만드는 방식에는 문제가 세 가지 있습니다. 첫째, 같은 일을 할 스레드를 매번 새로 만들고 버리느라 비용이 듭니다. 둘째, 동시에 실행되는 스레드 수에 상한이 없습니다. 요청이 1만 개 몰리면 스레드도 1만 개가 되고, 메모리가 바닥나거나 그 스레드들이 한꺼번에 접속하는 데이터베이스가 먼저 버티지 못합니다.

셋째, 스레드를 기다리고 정리하는 일과 작업에서 난 예외를 알아내는 일을 스레드를 만드는 모든 곳에서 따로 처리해야 합니다. 글 끝의 '더 깊이' 단락에서 볼 가상 스레드는 첫째 문제를 크게 줄여 주지만, 동시 실행 수의 상한과 종료 처리는 여전히 따로 정해야 합니다.

스레드 풀에 작업 맡기기

스레드 풀(thread pool)은 스레드를 몇 개만 만들어 두고 계속 재사용하는 구조입니다. 풀에 속한 스레드를 작업자 스레드(worker thread), 줄여서 작업자라고 부릅니다. 작업이 들어오면 쉬고 있는 작업자가 그 작업을 맡고, 모든 작업자가 바쁘면 작업은 작업 큐(work queue)에 줄을 서서 차례를 기다립니다. 작업을 마친 작업자는 종료하지 않고 큐에서 다음 작업을 꺼내므로, 작업이 6개든 6만 개든 스레드는 정해 둔 수만큼만 만들어집니다.

자바는 실행기를 인터페이스 두 개로 나타냅니다. Executor에는 작업을 넘기는 execute(Runnable) 메서드 하나만 있고, ExecutorService는 Executor를 확장해 실행기를 종료하는 메서드와 작업 결과를 돌려받는 submit 같은 메서드를 더합니다. 스레드 풀은 Executors 클래스의 정적 메서드로 간단히 만들 수 있습니다. 다음 예제는 Executors.newFixedThreadPool(2)로 작업자가 2개인 풀을 만들어 작업 6개를 맡기고, 각 작업은 자기를 실행한 스레드의 이름을 출력합니다.

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

public final class FirstPoolDemo {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        for (int task = 1; task <= 6; task++) {
            int id = task;
            executor.execute(() ->
                    System.out.println("task " + id + " on " + Thread.currentThread().getName()));
        }
        executor.shutdown();
        boolean terminated = executor.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("terminated=" + terminated);
    }
}

task 1 on pool-1-thread-1
task 3 on pool-1-thread-1
task 4 on pool-1-thread-1
task 5 on pool-1-thread-1
task 6 on pool-1-thread-1
task 2 on pool-1-thread-2
terminated=true

작업은 6개였지만 출력에 나온 스레드 이름은 pool-1-thread-1과 pool-1-thread-2 두 개뿐입니다. 두 작업자가 작업을 나눠 가져가며 재사용됐다는 뜻입니다. 이 이름은 Executors가 만드는 스레드에 기본으로 붙는 것으로, 첫 번째로 만든 풀의 첫 번째와 두 번째 스레드라는 뜻입니다. 람다 안에서 task 대신 id를 쓰는 것은, 반복할 때마다 값이 바뀌는 task를 람다 안에서 쓸 수 없기 때문입니다.

어느 작업자가 몇 개를 맡는지와 출력 순서는 실행할 때마다 달라집니다. 이 결과에서는 첫 번째 작업자가 다섯 개를 처리했지만, 여러 번 실행해 보니 두 번째 작업자가 다섯 개를 처리하기도 하고 넷과 둘로 나뉘기도 했습니다. 두 작업자가 동시에 작업을 가져가므로 누가 먼저 가져갈지는 스케줄러에 달려 있기 때문입니다. 반면 스레드 이름이 두 개뿐이라는 점, 작업마다 정확히 한 줄씩 나온다는 점, terminated=true가 맨 끝에 나온다는 점은 바뀌지 않습니다.

마지막 세 줄은 실행기를 정리하는 코드입니다. shutdown()은 실행기에 더는 새 작업을 받지 말라고 알립니다. 이미 받은 작업은 큐에 남은 것까지 모두 실행하지만, shutdown() 자신은 그 작업들이 끝나기를 기다리지 않고 바로 돌아옵니다. 기다리는 일은 awaitTermination(시간, 단위)가 맡습니다.

awaitTermination은 모든 작업이 끝나 실행기가 완전히 종료되면 true를, 정한 시간이 먼저 지나면 false를 돌려줍니다. 그래서 terminated=true는 언제나 작업 여섯 줄이 모두 출력된 뒤에 나옵니다. 1초 안에 작업이 다 끝나지 않을 만큼 컴퓨터가 바쁘다면 false가 나올 수도 있습니다.

정리 코드를 빼면 어떻게 될까요? shutdown()부터 마지막 줄까지 지우고 실행하면 작업 여섯 줄은 출력되지만 프로그램이 끝나지 않습니다. 풀의 작업자는 할 일이 없어도 다음 작업을 기다리며 살아 있고, 사용자 스레드이므로 JVM이 종료할 수 없기 때문입니다. 그래서 프로그램 안에서 만든 풀은 다 쓴 뒤 반드시 종료해야 합니다.

try-with-resources로 실행기 닫기

Java 19부터 ExecutorService는 AutoCloseable을 구현하므로, try-with-resources로 종료를 맡길 수 있습니다. 블록을 벗어날 때 호출되는 close()는 shutdown()을 부른 다음, 제출된 작업이 모두 끝나 실행기가 종료될 때까지 기다립니다. awaitTermination과 달리 기다리는 시간에 제한이 없습니다.

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

public final class CloseWaitsDemo {
    public static void main(String[] args) {
        AtomicInteger done = new AtomicInteger();
        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
            for (int task = 0; task < 4; task++) {
                executor.execute(() -> {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    done.incrementAndGet();
                });
            }
        }
        System.out.println("done=" + done.get());
    }
}

done=4

작업 4개는 각각 0.1초를 자고 나서 카운터를 1 올립니다. close()가 네 작업이 모두 끝날 때까지 기다리므로, 블록 다음 줄에서 읽는 값은 실행할 때마다 4입니다. 작업자 2개가 작업 4개를 두 번에 나눠 처리하므로 블록을 벗어나기까지는 0.2초쯤 걸립니다.

close()가 기다리는 동안 현재 스레드가 인터럽트되면, 실행 중인 작업에는 shutdownNow()처럼 중단을 요청하고 큐에 남은 작업은 실행하지 않습니다. 그래도 실행 중이던 작업이 끝날 때까지는 계속 기다린 뒤, 인터럽트 상태를 다시 설정하고 돌아옵니다. 작업이 언제 끝날지 모르는 서버 프로그램에서는 시간 제한이 있는 awaitTermination을 쓰고, 이 글의 예제처럼 작업이 곧 끝나는 프로그램에서는 close()를 쓰면 간편합니다.

어떤 스레드 풀을 고를까

Executors에는 자주 쓰는 실행기를 만드는 메서드가 준비돼 있습니다. 이름만 보고 고르기 쉽지만, 각각 어디에 제한이 없는지 알고 써야 합니다.

작업의 성격에 따라 출발점이 다릅니다. 계산만 하는 CPU 작업에는 코어 수 정도의 고정 크기 풀이 알맞습니다. 코어보다 많은 스레드가 동시에 계산해 봐야 컨텍스트 스위칭과 스레드 수 비용만 늘기 때문입니다. 외부 서비스를 호출하는 작업이라면 상대 서비스가 감당할 수 있는 동시 요청 수에 맞춰 작업자 수를 정합니다. 대기 시간이 대부분인 짧은 작업이 아주 많다면 가상 스레드를 검토하고, 작업을 반드시 하나씩 차례로 처리해야 한다면 단일 스레드 실행기를 씁니다.

다만 위의 메서드가 만드는 실행기는 모두 큐 용량이나 동시 실행 수 가운데 하나에 제한이 없습니다. 요청이 처리 속도보다 빠르게 들어오는 상황까지 대비하려면 둘 다 제한한 풀을 직접 만들어야 하고, 그때 쓰는 클래스가 ThreadPoolExecutor입니다. JDK 21에서 확인해 보면 newFixedThreadPool과 newCachedThreadPool이 돌려주는 객체도 설정만 다른 ThreadPoolExecutor입니다.

큐 용량을 정한 풀 만들기

다음 BoundedExecutorDemo는 작업자 2개와 용량 4인 큐를 가진 풀을 ThreadPoolExecutor 생성자로 직접 만들고, 작업 6개를 맡깁니다. 각 작업은 자기를 실행한 스레드의 이름을 목록에 추가하고, main은 종료를 기다린 뒤 완료된 작업 수와 작업자 이름을 출력합니다.

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public final class BoundedExecutorDemo {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2, 2, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(4),
                Thread.ofPlatform().name("report-worker-", 0).factory(),
                new ThreadPoolExecutor.AbortPolicy());
        List<String> names = Collections.synchronizedList(new ArrayList<>());
        for (int task = 0; task < 6; task++) {
            executor.execute(() -> names.add(Thread.currentThread().getName()));
        }
        executor.shutdown();
        if (!executor.awaitTermination(2, TimeUnit.SECONDS)) executor.shutdownNow();
        System.out.println("completed=" + executor.getCompletedTaskCount());
        System.out.println("workerNames=" + names.stream().distinct().sorted().toList());
    }
}

생성자에 넘기는 인수가 많으니 순서대로 하나씩 보겠습니다. 앞의 두 개는 작업자 수, 다음 두 개는 시간이고, 나머지는 큐와 스레드 팩터리, 거부 정책입니다.

• 2, 2: 핵심 작업자 수(corePoolSize)와 최대 작업자 수(maximumPoolSize)입니다. 둘이 같으므로 작업자는 많아야 2개입니다.
• 0, TimeUnit.MILLISECONDS: 핵심 수를 넘어 추가로 만든 작업자가 일 없이 얼마나 기다리다 정리될지 정하는 시간입니다. 이 설정에서는 추가 작업자가 생기지 않으므로 쓰이지 않습니다.
• new ArrayBlockingQueue<>(4): 작업자를 기다리는 작업을 담는 큐로, 최대 4개까지 담습니다.
• Thread.ofPlatform().name("report-worker-", 0).factory(): 작업자 스레드를 만들어 주는 스레드 팩터리(ThreadFactory)입니다. 스레드 이름을 report-worker-0, report-worker-1처럼 차례로 붙이며, Java 21에 추가된 Thread.Builder API를 씁니다.
• new ThreadPoolExecutor.AbortPolicy(): 작업을 더 받을 수 없을 때 어떻게 할지 정하는 거부 정책(rejection policy)입니다. AbortPolicy는 작업을 맡긴 쪽에 RejectedExecutionException을 던집니다.

completed=6
workerNames=[report-worker-0, report-worker-1]

10번 실행해 10번 모두 같은 결과가 나왔습니다. completed는 실행기가 세어 둔 완료 작업 수(getCompletedTaskCount())이고, workerNames는 작업을 실행한 스레드 이름에서 중복을 없애고 정렬한 목록입니다. 어느 작업자가 몇 개를 처리했는지는 실행마다 다르지만, 중복을 없앴기 때문에 출력에는 드러나지 않습니다. 스레드에 이름을 붙여 두면 로그나 스레드 덤프에서 어느 풀의 스레드인지 바로 알아볼 수 있습니다.

단계 그림이 보여 주듯, 처음 두 번의 execute는 작업자를 새로 만들고 그 작업을 새 작업자의 첫 작업으로 바로 넘깁니다. 작업자가 핵심 수보다 적으면 다른 작업자가 쉬고 있더라도 새 작업자를 만들고, 이때는 큐를 거치지 않습니다. 세 번째 제출부터는 작업이 큐에 들어가고, 앞 작업을 마친 작업자가 큐에서 꺼내 실행합니다. 그런데 작업자들이 큐를 동시에 비워 가므로, 이 프로그램에서 큐에 작업 네 개가 한꺼번에 쌓였는지는 알 수 없습니다.

작업자도 큐도 가득 차면

작업자 2개와 큐 4칸이라는 설정은, 작업이 하나도 끝나지 않는 동안 받아 둘 수 있는 작업이 최대 6개라는 뜻입니다. 작업마다 오래 걸려서 아무것도 끝나지 않는다고 가정하고 같은 풀에 execute를 일곱 번 호출하면 상태는 다음과 같이 바뀝니다.

작업이 끝나지 않도록 막아 두고 실제로 일곱 번 제출해 보면, 6개를 받고 1개를 거부하며 그 순간 실행 중 2개, 대기 4개로 표와 같습니다. 일곱 번째 호출에서 execute는 자리가 날 때까지 기다리지 않고 곧바로 예외를 던집니다. 이 예외는 작업자가 아니라 execute를 호출한 스레드, 여기서는 main에서 발생하므로 작업을 맡기는 코드가 처리해야 합니다.

반대로 앞의 예제처럼 작업이 금방 끝나면 그사이 자리가 비므로, 같은 풀에 작업을 훨씬 많이 맡겨도 거부되지 않을 수 있습니다. 거부 여부를 정하는 것은 지금까지 맡긴 작업의 총개수가 아니라 작업을 맡기는 그 순간에 남은 자리입니다. 어떤 거부 정책을 고를지는 스레드 풀의 작업 거부 정책에서 다룹니다.

핵심 작업자 수와 최대 작업자 수를 다르게 두면 규칙이 한 단계 늘어납니다. ThreadPoolExecutor는 작업자가 핵심 수보다 적으면 새 작업자를 만들고, 핵심 수에 이르면 큐에 넣으며, 큐가 가득 찼을 때 비로소 최대 수까지 작업자를 더 만듭니다. 그래도 받을 수 없으면 거부합니다. 정확한 규칙은 ThreadPoolExecutor 공식 문서에 있고, 단계마다 직접 따라가 보는 일은 ThreadPoolExecutor의 스레드 수와 작업 큐에서 합니다.

작업에서 예외가 나면

execute로 맡긴 작업이 예외를 던지면 어떻게 될까요? 다음 프로그램은 작업자가 하나뿐인 실행기에 작업 세 개를 맡기고, 두 번째 작업에서 예외를 던집니다.

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

public final class UncaughtTaskFailure {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(() -> System.out.println("first on " + Thread.currentThread().getName()));
        executor.execute(() -> {
            throw new IllegalStateException("boom");
        });
        executor.execute(() -> System.out.println("third on " + Thread.currentThread().getName()));
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
    }
}

first on pool-1-thread-1
Exception in thread "pool-1-thread-1" java.lang.IllegalStateException: boom
	at UncaughtTaskFailure.lambda$main$1(UncaughtTaskFailure.java:10)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
third on pool-1-thread-2

두 번째 작업의 예외는 아무도 잡지 않았으므로, 그 작업을 실행하던 pool-1-thread-1이 예외와 함께 끝났습니다. 스택 트레이스는 프로세스와 스레드에서 본 것처럼, 잡히지 않은 예외를 JVM의 기본 처리기가 표준 오류에 출력한 것입니다. 그런데 세 번째 작업은 그대로 실행됐고, 스레드 이름은 pool-1-thread-2로 바뀌었습니다. 작업자가 하나뿐인 실행기도 예외로 끝난 작업자 대신 새 작업자를 만들어 남은 작업을 이어 간 것입니다.

실행기는 이렇게 계속 동작하지만, main은 두 번째 작업이 실패했다는 사실을 전혀 알지 못합니다. 표준 오류에 찍힌 메시지를 누군가 읽지 않으면 실패는 그대로 묻힙니다. 덧붙여 표준 출력과 표준 오류는 서로 다른 스트림이라서, 실행할 때마다 third 줄이 스택 트레이스 앞이나 중간에 끼어 나오기도 합니다. 스택 트레이스의 줄 번호는 JDK 21 기준이며 버전에 따라 다릅니다.

실패를 알아내는 간단한 방법은 작업을 한 겹 감싸서 예외를 직접 잡는 것입니다. 다음 ObservedExecutor는 작업을 실행기에 넘기기 전에 try-catch로 감싸, RuntimeException이 나면 실패 횟수를 세고 작업 이름을 표준 오류에 출력합니다. Error는 잡지 않으므로 그대로 작업자를 끝냅니다. 예외 처리만 보여 주려는 예제라서 큐 용량에 제한이 없는 newFixedThreadPool(2)를 썼습니다.

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

public final class ObservedExecutor {
    private final ExecutorService delegate = Executors.newFixedThreadPool(2);
    private final AtomicInteger failures = new AtomicInteger();

    void execute(String name, Runnable task) {
        delegate.execute(() -> {
            try {
                task.run();
            } catch (RuntimeException e) {
                failures.incrementAndGet();
                System.err.println("task=" + name + " failed=" + e.getClass().getSimpleName());
            }
        });
    }

    public static void main(String[] args) throws InterruptedException {
        ObservedExecutor service = new ObservedExecutor();
        service.execute("ok", () -> {});
        service.execute("broken", () -> {
            throw new IllegalStateException("boom");
        });
        service.delegate.shutdown();
        service.delegate.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("failures=" + service.failures.get());
    }
}

task=broken failed=IllegalStateException
failures=1

첫 줄은 래퍼의 catch가 표준 오류에 출력한 것이고, 둘째 줄은 main이 종료를 기다린 뒤 표준 출력에 출력한 것입니다. 이번에는 스택 트레이스가 나오지 않습니다. 예외가 래퍼 안에서 처리되어 작업자까지 올라가지 않았으므로 작업자도 끝나지 않습니다. 10번 실행해 10번 모두 failures=1이 나왔습니다.

이 래퍼로도 세지 못하는 실패가 있습니다. 실행기를 shutdown()한 뒤에 작업을 맡기면 execute가 호출한 스레드에서 곧바로 RejectedExecutionException을 던집니다. 이 예외는 작업이 실행되기 전, 작업을 맡기는 단계에서 생긴 것이므로 래퍼의 catch까지 가지 않고 failures도 늘지 않습니다. 작업이 실행되다 실패한 것과 작업을 맡기지 못한 것은 생기는 곳이 다르므로 따로 처리해야 합니다. 다음 글에서 볼 submit과 Future를 쓰면 이런 래퍼 없이도 작업의 반환값과 예외를 작업을 맡긴 쪽에서 받을 수 있습니다.

awaitTermination이 false를 돌려줄 때

지금까지의 예제는 종료를 기다린 뒤 결과를 읽었습니다. 그런데 awaitTermination이 false를 돌려주면, 즉 정한 시간 안에 작업이 다 끝나지 않으면 이야기가 달라집니다. ObservedExecutor는 반환값을 아예 확인하지 않으므로, 1초 안에 종료되지 않았다면 failures를 0으로 읽은 뒤에 catch가 뒤늦게 1로 올릴 수도 있습니다.

BoundedExecutorDemo는 false일 때 shutdownNow()를 호출하지만, 그 직후 바로 결과를 읽는다는 점이 문제입니다. shutdownNow()는 실행 중인 작업에 인터럽트로 중단을 요청하고, 큐에서 아직 시작하지 않은 작업을 꺼내 목록으로 돌려줄 뿐입니다. 실행 중인 작업이 끝나기를 기다리지 않으며, 인터럽트에 반응하지 않는 작업은 계속 실행될 수 있습니다. 그러면 그 시점의 getCompletedTaskCount()는 아직 변하는 중인 근삿값이고, 작업자가 names에 이름을 추가하는 중일 수도 있습니다.

Collections.synchronizedList가 감싼 목록은 add 같은 메서드 하나하나를 안전하게 만들어 주지만, stream()이나 반복자로 목록 전체를 훑는 동안까지 지켜 주지는 않습니다. 공식 문서도 이렇게 순회할 때는 synchronized (names)처럼 목록을 직접 잠그라고 요구합니다. 잠그지 않고 훑는 사이 다른 스레드가 원소를 추가하면, 예외가 나는 것을 포함해 결과를 예측할 수 없습니다. 결국 결과를 최종값으로 읽으려면 awaitTermination이 true를 돌려준 것부터 확인해야 합니다.

executor.shutdown();
if (executor.awaitTermination(2, TimeUnit.SECONDS)) {
    System.out.println("completed=" + executor.getCompletedTaskCount());
} else {
    List<Runnable> neverStarted = executor.shutdownNow();
    System.out.println("not terminated yet, never started=" + neverStarted.size());
}

true인 쪽에서는 모든 작업이 끝났으므로 완료 수와 결과 목록을 그대로 믿을 수 있습니다. false인 쪽에서는 시작하지 못한 작업 목록을 받아 기록하거나 다른 곳에 넘기고, 결과는 중간값으로 다룹니다. 여러 단계를 거쳐 실행기를 종료하는 절차는 ExecutorService를 안전하게 종료하기에서 자세히 다룹니다.

업무 코드와 실행 방식 나누기

업무 코드가 ThreadPoolExecutor를 직접 만들거나 그 구체 타입에 의존하면, 풀 설정을 바꿀 때마다 업무 코드도 함께 고쳐야 합니다. 실행기를 쓰는 쪽은 Executor 인터페이스만 알고, 어떤 실행기를 쓸지는 객체를 조립하는 한곳에서 정하는 편이 좋습니다. 다음 ReportService는 생성자로 받은 Executor에 작업을 넘기기만 합니다.

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

public final class ReportService {
    private final Executor executor;

    ReportService(Executor executor) {
        this.executor = executor;
    }

    void requestReport(String reportId) {
        executor.execute(() ->
                System.out.println(reportId + " on " + Thread.currentThread().getName()));
    }

    public static void main(String[] args) {
        new ReportService(Runnable::run).requestReport("test-report");

        try (ExecutorService pool = Executors.newFixedThreadPool(2)) {
            new ReportService(pool).requestReport("daily-report");
        }
    }
}

test-report on main
daily-report on pool-1-thread-1

첫 줄은 Runnable::run을 실행기로 넘긴 경우입니다. Executor는 추상 메서드가 execute 하나뿐인 함수형 인터페이스이므로, 받은 작업의 run()을 그 자리에서 호출하는 메서드 참조로도 만들 수 있습니다. 그래서 작업이 호출한 스레드인 main에서 바로 실행되고, 테스트에서는 결과를 곧바로 확인할 수 있습니다. 둘째 줄은 실제 풀을 넘긴 경우로, 같은 ReportService 코드가 풀의 작업자에서 실행됐습니다. 업무 코드는 그대로 두고 실행 방식만 바꾼 것입니다.

실행기를 종료하는 일은 그 실행기를 만든 쪽이 맡습니다. ReportService는 받은 실행기를 닫지 않고, 풀을 만든 main이 try-with-resources로 닫습니다. 여러 기능이 함께 쓰는 풀을 한 기능이 마음대로 종료하면, 다른 기능이 맡기는 작업이 갑자기 RejectedExecutionException으로 거부되기 때문입니다. 반대로 만든 쪽이 닫는 것을 잊으면, 앞에서 본 것처럼 작업자가 남아 프로세스가 끝나지 않습니다.

운영하는 서비스라면 작업마다 요청 ID와 작업 종류를 붙이고, 작업을 맡긴 시각과 시작한 시각, 끝난 시각을 같은 ID로 기록해 두는 것이 좋습니다. 그래야 응답이 느릴 때 작업이 큐에서 오래 기다린 것인지, 실행 자체가 오래 걸린 것인지 구분할 수 있습니다. 앞의 경우라면 작업자가 부족하다는 뜻이고, 뒤의 경우라면 작업 자체를 살펴봐야 합니다. 로그에 남는 작업 이름에는 개인정보를 넣지 않습니다.

더 깊이: 가상 스레드 실행기

Java 21에서 정식 기능이 된 가상 스레드(virtual thread)는 JVM이 관리하는 가벼운 스레드입니다. Executors.newVirtualThreadPerTaskExecutor()는 작업마다 가상 스레드를 새로 만들어 실행하는 실행기로, 처음에 본 요청별 스레드 방식과 모양은 같지만 비용이 훨씬 작습니다. 다음은 앞의 ThreadPerRequest와 같은 작업 5,000개를 이 실행기로 실행하는 예입니다.

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

public final class VirtualPerTask {
    public static void main(String[] args) {
        AtomicInteger done = new AtomicInteger();
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int request = 0; request < 5_000; request++) {
                executor.execute(() -> {
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    done.incrementAndGet();
                });
            }
        }
        System.out.println("done=" + done.get());
    }
}

done=5000

같은 환경에서 이 프로그램은 0.3초 안팎에 끝나, 플랫폼 스레드 5,000개를 만들던 프로그램보다 훨씬 빨랐습니다. 가상 스레드는 만드는 비용이 작고, 자거나 입출력을 기다리는 동안에는 대개 운영체제 스레드를 붙잡고 있지 않기 때문입니다. 그래서 이 실행기는 스레드를 재사용하지 않고 작업마다 새로 만듭니다. 싸게 만들 수 있는 스레드를 풀에 모아 둘 이유가 없기 때문입니다.

하지만 가상 스레드가 코어를 늘려 주지는 않으므로 CPU 계산만 하는 작업은 빨라지지 않습니다. 또 이 실행기는 동시에 실행되는 작업 수를 제한하지 않습니다. 작업 5,000개가 모두 연결이 20개뿐인 데이터베이스를 호출한다면, 작업자 수 대신 세마포어 같은 다른 수단으로 동시 호출 수를 제한해야 합니다. 그 방법은 가상 스레드와 ScopedValue에서 다룹니다.

정리

• 실행기에 작업을 맡기면 업무 코드는 무엇을 실행할지만 말하고, 스레드를 몇 개 두고 언제 만들지는 실행기를 만드는 곳에서 정합니다.
• 스레드 풀은 작업자 스레드를 재사용합니다. 모든 작업자가 바쁘면 작업은 큐에서 기다립니다.
• shutdown()은 새 작업을 막을 뿐 기다리지 않습니다. 결과는 awaitTermination이 true를 돌려준 뒤, 또는 try-with-resources의 close()가 끝난 뒤에 읽습니다.
• Executors의 메서드가 만드는 실행기는 큐 용량이나 동시 실행 수 가운데 하나에 제한이 없습니다. 둘 다 제한하려면 ThreadPoolExecutor로 작업자 수, 큐 용량, 거부 정책을 함께 정합니다.
• execute로 맡긴 작업의 예외는 작업을 맡긴 쪽에 전달되지 않습니다. 작업을 감싸서 잡거나, 다음 글의 submit과 Future를 씁니다.
• 실행기를 종료하는 일은 그 실행기를 만든 쪽이 맡습니다.

연습 문제

1. 작업자 3개, 큐 용량 6인 실행기에 작업 아홉 개를 맡기고 종료를 요청하세요. 각 작업은 자기를 실행한 작업자 이름별 처리 횟수를 ConcurrentHashMap의 merge로 셉니다. 최대 1초 동안 종료를 기다린 뒤 전체 처리 횟수와 작업자별 처리 횟수를 출력하고, 종료가 확인되지 않았다면 이 값을 어떻게 읽어야 할지도 생각해 보세요.
2. 핵심 작업자 수와 최대 작업자 수가 모두 3이고 큐 용량이 2이며 AbortPolicy를 쓰는 실행기가 있습니다. 작업이 하나도 끝나지 않는 동안 execute를 일곱 번 호출하면 실행 중인 작업, 대기 중인 작업, 거부된 호출은 각각 몇 개인가요? 거부되는 호출은 자리가 날 때까지 기다리나요?
3. 2번의 상태에서 실행 중인 세 작업은 인터럽트를 받아도 무시하고 계속 실행된다고 합시다. shutdown()을 호출하고 awaitTermination이 false를 돌려준 뒤 shutdownNow()를 호출했습니다. shutdownNow()가 돌려주는 목록에는 작업이 몇 개 들어 있나요? 이 시점에 실행기는 종료됐나요?
4. ObservedExecutor에서 두 작업이 끝나고 실행기가 종료된 것을 확인한 뒤, 같은 실행기에 정상 작업을 하나 더 맡겼습니다. 어떤 예외가 어느 스레드에서 발생하며, failures는 몇이 되나요?

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

public final class WorkerCountSolution {
    public static void main(String[] args) throws InterruptedException {
        var counts = new ConcurrentHashMap<String, Integer>();
        var executor = new ThreadPoolExecutor(3, 3, 0, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(6),
                Thread.ofPlatform().name("count-worker-", 0).factory());
        for (int i = 0; i < 9; i++) {
            executor.execute(() -> counts.merge(Thread.currentThread().getName(), 1, Integer::sum));
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
        System.out.println("total=" + counts.values().stream().mapToInt(Integer::intValue).sum());
        System.out.println(counts);
    }
}

total=9
{count-worker-1=1, count-worker-2=7, count-worker-0=1}

다음 글

Callable과 Future로 작업 결과 받기에서는 execute 대신 submit으로 작업을 맡기고, 작업이 돌려준 값과 던진 예외를 Future로 받아 봅니다.