서버에서 공용 풀과 전용 풀 나눠 쓰기

앞 글까지는 병렬 스트림을 main 하나에서 실행하고 결과와 시간을 확인했습니다. 그때 프로그램은 CPU를 거의 혼자 쓰고 있었습니다. 서버는 사정이 다릅니다. 요청마다 처리 스레드가 따로 돌고, 여러 요청이 같은 순간에 병렬 스트림을 실행할 수 있습니다. 그런데 병렬 스트림은 요청마다 새 스레드 묶음을 만들지 않고, JVM에 하나뿐인 공용 풀(common pool)을 함께 씁니다.

그래서 서버에서 병렬 처리를 쓸 때는 '병렬로 돌리면 빨라지는가'보다 '이 풀을 누가 함께 쓰는가'를 먼저 물어야 합니다. 이 글에서는 여러 요청이 동시에 병렬 스트림을 실행할 때 처리 시간이 어떻게 변하는지 직접 재 보고, 계산이 큰 기능을 전용 풀로 떼어 내는 방법과 그 풀을 만들고 닫는 시점을 살펴봅니다. 이어서 서로 다른 작업을 조합할 때와 I/O를 기다리는 작업을 다룰 때 어떤 실행기를 고르는지 정리합니다.

공용 풀은 JVM에 하나뿐입니다

ForkJoinPool.commonPool()이 돌려주는 공용 풀은 JVM 안에 정적으로 하나만 만들어집니다. ForkJoinPool 문서는 특정 풀에 명시적으로 제출하지 않은 Fork/Join 작업이 모두 이 풀에서 실행된다고 설명합니다. 병렬 스트림이 바로 그런 작업이고, 실행기를 넘기지 않은 CompletableFuture.supplyAsync 같은 비동기 메서드도 이 풀을 씁니다. 즉 어느 요청의 처리 코드에서 병렬 스트림을 실행하든 실제 계산은 같은 작업자 스레드들이 나눠 맡습니다.

공용 풀의 작업자는 많지 않습니다. 동시에 계산할 작업자 수의 목표값을 병렬성(parallelism)이라고 하는데, OpenJDK는 공용 풀의 병렬성을 기본적으로 사용 가능한 프로세서 수보다 하나 적게 정합니다. 병렬 스트림을 호출한 스레드도 계산에 참여하므로, 호출한 스레드 하나와 작업자를 합치면 프로세서 수와 같아집니다. 이 글의 예제를 실행한 환경은 프로세서가 2개여서 공용 풀의 병렬성이 1이었습니다.

공용 풀은 닫을 수도 없습니다. shutdown(), shutdownNow(), close()를 호출해도 공용 풀의 실행 상태는 바뀌지 않습니다. 프로세스가 끝날 때까지 모든 코드가 함께 쓰는 자원이라는 뜻이고, 앞 글에서 본 것처럼 시스템 속성으로 병렬성을 바꾸면 그 영향도 프로세스 전체에 미칩니다.

여러 요청이 동시에 병렬 스트림을 실행하면

공용 풀을 함께 쓴다는 것이 실제로 어떤 차이를 만드는지 재 보겠습니다. 아래 프로그램은 요청 처리 스레드를 흉내 내는 고정 크기 풀을 만들고, 요청 1개만 처리할 때와 프로세서 수의 두 배만큼 요청을 한꺼번에 처리할 때를 비교합니다. 요청 하나가 하는 일은 정수 100만 개에 같은 계산을 적용해 합을 구하는 것이고, 이 일을 순차 스트림과 병렬 스트림으로 각각 실행합니다. 출력하는 값은 요청마다 걸린 시간입니다.

CountDownLatch는 요청들이 되도록 같은 순간에 계산을 시작하도록 출발 신호를 한 번에 주는 데 씁니다. 각 요청은 start.await()에서 기다리다가 main이 countDown()을 호출하면 함께 출발합니다. 또 요청마다 자기가 구한 합을 미리 구해 둔 기대값과 비교하므로, 시간은 짧게 나왔는데 답이 틀린 경우를 놓치지 않습니다. 결과를 실제로 사용하기 때문에 JIT 컴파일러가 계산을 생략할 수도 없습니다.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.stream.IntStream;

public final class ConcurrentCommonPoolRequests {
    private static long cpuWork(int value) {
        long result = value;
        for (int round = 0; round < 200; round++) {
            result = (result * 1_103_515_245L + 12_345L) & 0x7fff_ffffL;
        }
        return result;
    }

    private static long sum(boolean parallel) {
        IntStream values = IntStream.range(0, 1_000_000);
        if (parallel) {
            values = values.parallel();
        }
        return values.mapToLong(ConcurrentCommonPoolRequests::cpuWork).sum();
    }

    // 요청 하나: 합계를 구해 기대값과 비교하고, 걸린 시간을 밀리초로 돌려준다
    private static long handle(boolean parallel, long expected) {
        long start = System.nanoTime();
        if (sum(parallel) != expected) {
            throw new IllegalStateException("wrong sum");
        }
        return (System.nanoTime() - start) / 1_000_000;
    }

    // 요청 count개를 동시에 출발시키고 요청마다 걸린 시간을 모은다
    private static List<Long> run(int count, boolean parallel, long expected)
            throws Exception {
        CountDownLatch start = new CountDownLatch(1);
        try (ExecutorService requests = Executors.newFixedThreadPool(count)) {
            List<Future<Long>> results = new ArrayList<>();
            for (int i = 0; i < count; i++) {
                results.add(requests.submit(() -> {
                    start.await();
                    return handle(parallel, expected);
                }));
            }
            start.countDown();
            List<Long> millis = new ArrayList<>();
            for (Future<Long> result : results) {
                millis.add(result.get());
            }
            return millis;
        }
    }

    public static void main(String[] args) throws Exception {
        int cores = Runtime.getRuntime().availableProcessors();
        System.out.println("cores=" + cores + ", common-parallelism="
                + ForkJoinPool.commonPool().getParallelism());

        long expected = sum(false);
        run(1, true, expected); // 예열

        int many = cores * 2;
        System.out.println("요청 1개, 순차: " + run(1, false, expected) + " ms");
        System.out.println("요청 1개, 병렬: " + run(1, true, expected) + " ms");
        System.out.println("요청 " + many + "개, 순차: " + run(many, false, expected) + " ms");
        System.out.println("요청 " + many + "개, 병렬: " + run(many, true, expected) + " ms");
    }
}

프로세서가 2개인 환경에서 실행한 결과의 한 예입니다. 시간은 실행할 때마다 달라지고, 프로세서 수가 다르면 요청 수와 공용 풀의 병렬성도 달라집니다.

cores=2, common-parallelism=1
요청 1개, 순차: [314] ms
요청 1개, 병렬: [163] ms
요청 4개, 순차: [563, 630, 619, 603] ms
요청 4개, 병렬: [522, 746, 517, 351] ms

요청이 하나뿐일 때는 병렬 스트림이 거의 두 배 빨랐습니다. 요청 스레드와 공용 풀의 작업자 하나가 일을 나눠 맡았고, 두 스레드가 쓸 수 있는 프로세서도 두 개였기 때문입니다. 컴퓨터가 한가할 때 여러 번 실행해 보니 순차는 약 300ms, 병렬은 약 160ms로 꾸준했습니다.

요청 4개가 동시에 들어오자 그림이 바뀌었습니다. 순차로 처리한 요청은 각각 혼자일 때의 두 배쯤 걸렸습니다. 요청 하나에 약 300ms의 계산이 필요한데 요청 4개의 계산 1,200ms어치를 프로세서 두 개가 나눠 처리하니, 요청마다 600ms 안팎이 걸리는 것이 자연스럽습니다.

눈여겨볼 줄은 마지막 줄입니다. 병렬로 바꿔도 요청들은 혼자일 때처럼 빨라지지 않았습니다. 평균은 순차와 비슷했고, 요청 사이의 차이만 커졌습니다. 컴퓨터가 한가할 때 세 번 실행한 결과를 모아 보면 순차는 500에서 660ms 사이에 모였지만, 병렬은 350에서 810ms 사이로 흩어졌습니다.

이유는 남는 프로세서가 없다는 데 있습니다. 요청 스레드 4개만으로도 이미 프로세서 2개보다 많고, 공용 풀의 작업자 1개는 요청 4개가 나눠 써야 합니다. 그 작업자가 마침 도와준 요청은 351ms처럼 빨리 끝나지만, 나머지는 자기 스레드만으로 계산하면서 프로세서를 차지하려고 경쟁합니다. 일을 잘게 나누고 합치는 비용은 그대로 드는데 일을 나눠 맡을 빈 프로세서는 없으니, 병렬 스트림이 줄 수 있는 이득이 사라진 것입니다. 실제로 다른 프로그램이 CPU를 함께 쓰던 때에 실행하자, 요청이 하나뿐이어도 병렬과 순차가 모두 300ms대로 거의 같게 나왔습니다.

이 실험에서 서버에 중요한 사실 두 가지를 읽을 수 있습니다. 첫째, 병렬 처리는 남는 프로세서가 있을 때만 응답 시간을 줄입니다. 동시 요청이 프로세서를 모두 채우면 요청 전체를 처리하는 시간은 전체 계산량을 프로세서 수로 나눈 값보다 짧아질 수 없고, 병렬 스트림은 요청들 가운데 누가 먼저 끝나는지만 바꿉니다. 둘째, 공용 풀을 쓰는 요청의 응답 시간은 다른 요청이 무엇을 하는지에 따라 달라집니다. 어떤 기능이 큰 계산으로 공용 풀의 작업자를 오래 차지하면 다른 기능의 짧은 계산은 도움을 받지 못합니다. 누군가 병렬 스트림 안에서 파일을 읽거나 원격 서비스를 호출하는 식으로 공용 풀에 기다리는 작업을 넣으면, 작업자는 아무 계산도 하지 않으면서 자리만 차지하게 됩니다.

그렇다고 서버에서 병렬 스트림을 쓰지 말라는 뜻은 아닙니다. 요청이 드물고 요청 하나의 계산이 큰 기능이라면 병렬 처리가 응답 시간을 분명히 줄일 수 있습니다. 다만 그 이득은 남는 프로세서가 있을 때만 생기고, 공용 풀을 쓰는 순간 다른 기능과 같은 작업자를 나눠 쓰게 된다는 점을 알고 써야 합니다. 이 두 가지를 직접 통제하고 싶을 때 쓰는 방법이 다음 절의 전용 풀입니다.

계산이 큰 기능에 전용 ForkJoinPool 두기

특정 기능의 계산이 크고 CPU만 쓰는 작업이라면, 그 기능만 쓰는 ForkJoinPool을 애플리케이션에 하나 만들어 둘 수 있습니다. 예를 들어 통계 분석 기능에 병렬성이 2인 풀을 주면 그 기능의 계산은 이 풀의 작업자들이 맡고, 공용 풀의 작업자는 건드리지 않습니다. 분석 요청이 몰려도 다른 기능이 쓰는 공용 풀의 작업자는 그대로 남습니다.

그런데 병렬 스트림에는 실행할 풀을 지정하는 매개변수가 없습니다. 대신 전용 풀에 제출한 작업 안에서 병렬 스트림을 실행하면 그 스트림은 전용 풀에서 실행됩니다. 아래 프로그램으로 확인해 보겠습니다. 병렬 스트림이 원소를 처리할 때마다 현재 스레드의 이름을 모으는데, 이름 끝의 번호는 실행마다 달라질 수 있어서 #으로 바꿔 모읍니다. 원소마다 cpuWork로 계산 시간을 조금씩 들이는 것은 여러 스레드가 일을 나눠 가질 시간을 주기 위해서이고, 합계 자체는 쓰지 않습니다.

import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;

public final class WhichPoolRunsTheStream {
    private static long cpuWork(int value) {
        long result = value;
        for (int round = 0; round < 100_000; round++) {
            result = (result * 1_103_515_245L + 12_345L) & 0x7fff_ffffL;
        }
        return result;
    }

    // 병렬 스트림의 원소를 처리한 스레드 이름을 모은다. 끝의 번호는 #으로 바꾼다.
    private static Set<String> threadNames() {
        Set<String> names = ConcurrentHashMap.newKeySet();
        IntStream.range(0, 1_000)
                .parallel()
                .mapToLong(value -> {
                    names.add(Thread.currentThread().getName().replaceAll("\\d+$", "#"));
                    return cpuWork(value);
                })
                .sum();
        return new TreeSet<>(names);
    }

    public static void main(String[] args) throws Exception {
        System.out.println("main에서 호출: " + threadNames());
        try (ForkJoinPool analyticsPool = new ForkJoinPool(2)) {
            Set<String> names = analyticsPool.submit(WhichPoolRunsTheStream::threadNames).get();
            System.out.println("전용 풀에서 호출: " + names);
        }
    }
}

main에서 호출: [ForkJoinPool.commonPool-worker-#, main]
전용 풀에서 호출: [ForkJoinPool-1-worker-#]

main에서 호출한 병렬 스트림은 호출한 main 스레드와 공용 풀의 작업자가 함께 처리했습니다. 반면 전용 풀에 제출한 작업 안에서 호출한 스트림은 ForkJoinPool-1-worker-#, 즉 이 프로그램이 처음 만든 사용자 풀의 작업자만 처리했습니다. 이때는 스트림을 호출한 스레드부터가 전용 풀의 작업자이므로 공용 풀은 한 번도 쓰이지 않습니다. 열두 번 실행했을 때 모두 같은 결과가 나왔습니다.

다만 이 동작은 Stream API가 약속한 것이 아니라 JDK 구현에서 나오는 결과입니다. 이유는 아래 '더 깊이'에서 설명합니다. 실행할 풀을 코드에 분명히 드러내고 싶다면 RecursiveTask 같은 Fork/Join 작업을 직접 만들어 풀에 제출하거나, 뒤에서 볼 CompletableFuture에 실행기를 넘기는 방법이 더 명확합니다.

이제 이런 풀을 서버 코드에서 쓰는 모양을 보겠습니다. 아래 DedicatedAnalyticsPool은 분석 기능의 전용 풀을 감싼 클래스입니다. 생성자에서 풀을 한 번 만들고, 요청 처리 코드는 score를 호출해 계산을 맡긴 뒤 정해진 시간까지만 결과를 기다립니다. 애플리케이션이 끝날 때는 close가 풀을 종료합니다. AutoCloseable을 구현했으므로 예제의 main처럼 try-with-resources로 닫을 수도 있습니다.

import java.time.Duration;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

public final class DedicatedAnalyticsPool implements AutoCloseable {
    private final ForkJoinPool pool;

    public DedicatedAnalyticsPool(int parallelism) {
        pool = new ForkJoinPool(parallelism);
    }

    public long score(List<Integer> values, Duration timeout) throws Exception {
        return pool.submit(
                        () ->
                                values.parallelStream()
                                        .mapToLong(value -> (long) value * value)
                                        .sum())
                .get(timeout.toMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public void close() {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
                pool.shutdownNow();
            }
        } catch (InterruptedException exception) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws Exception {
        try (DedicatedAnalyticsPool analytics = new DedicatedAnalyticsPool(2)) {
            long result = analytics.score(List.of(1, 2, 3, 4, 5), Duration.ofSeconds(1));
            System.out.println("score=" + result);
        }
    }
}

score=55

입력 1, 2, 3, 4, 5의 제곱을 더하면 1 + 4 + 9 + 16 + 25 = 55입니다. 계산은 간단하지만, score 안의 병렬 스트림은 앞에서 확인한 대로 공용 풀이 아니라 이 클래스가 만든 풀에서 실행됩니다.

생성자에 넘긴 병렬성 2는 이 풀이 동시에 계산에 쓰려는 작업자 수의 목표값입니다. 프로세서 두 개를 이 풀에 예약하거나, 이 풀의 CPU 사용량을 운영체제 수준에서 제한하는 설정은 아닙니다. 다른 풀과 요청 스레드도 같은 프로세서를 두고 경쟁하므로, 전용 풀이 정해 주는 것은 '이 기능이 한꺼번에 몇 개의 작업자로 계산하는가'까지입니다. 그래서 이 값은 애플리케이션 전체의 CPU 예산, 즉 프로세서를 기능별로 어떻게 나눠 쓸지를 생각해서 정합니다.

더 깊이: 병렬 스트림이 실행할 풀을 고르는 방식

병렬 스트림의 최종 연산은 내부에서 Fork/Join 작업을 만들어 실행하고, 일을 나눌 때 ForkJoinTask.fork()를 호출합니다. ForkJoinTask 문서에 따르면 fork()는 현재 작업이 실행 중인 풀이 있으면 그 풀에서, Fork/Join 풀 안에서 실행 중이 아니면 공용 풀에서 작업을 비동기로 실행하도록 맡깁니다. 그래서 전용 풀의 작업자가 병렬 스트림을 호출하면 나뉜 조각이 그 풀의 큐로 들어가고, main처럼 풀 밖의 스레드가 호출하면 공용 풀로 들어갑니다.

fork()의 이 규칙은 문서에 적혀 있지만, 병렬 스트림이 내부에서 Fork/Join 작업을 쓴다는 사실은 Stream 문서가 약속하지 않는 구현 세부입니다. 널리 쓰이는 방법이기는 하지만, 이 동작에 기대는 코드라면 그 사실을 주석으로 남겨 두는 편이 좋습니다. 나중에 누군가 score의 계산을 풀 밖으로 옮기면 아무 오류 없이 공용 풀로 돌아가 버리기 때문입니다.

시간 제한은 결과를 기다리는 시간에만 걸립니다

score의 get(timeout.toMillis(), TimeUnit.MILLISECONDS)는 결과를 최대 그 시간까지만 기다립니다. 시간 안에 계산이 끝나면 결과를 돌려주고, 계산이 예외로 끝났다면 그 예외를 ExecutionException으로 감싸 던집니다. 시간이 지나도록 끝나지 않으면 TimeoutException을 던지는데, 이것은 호출자가 기다리기를 그만뒀다는 뜻일 뿐입니다. score는 cancel을 호출하지 않으므로 계산은 풀 안에서 계속 실행됩니다. score의 기다림과 close의 기다림이 각각 어떻게 끝나는지 단계별로 보면 다음과 같습니다.

첫 단계가 서버에서 문제가 되는 이유는 계산이 계속 풀의 작업자를 차지하기 때문입니다. 요청은 시간 초과로 실패 응답을 받았는데 그 요청의 계산이 작업자 두 개를 붙잡고 있으면, 뒤따르는 분석 요청은 큐에서 기다려야 합니다. 그러면 뒤의 요청도 시간 초과가 나기 쉽고, 결과를 받아 줄 사람이 없는 계산이 쌓이면서 풀 전체가 밀립니다.

그렇다면 시간이 초과됐을 때 cancel(true)를 호출하면 될까요? 이 글의 환경(JDK 21)에서 오래 걸리는 병렬 스트림 계산을 score와 같은 방식으로 제출하고, 100ms만 기다린 뒤 cancel(true)를 호출해 봤습니다. isCancelled()는 true가 됐지만 0.2초 뒤에도 getActiveThreadCount()는 2였고, 이어서 shutdown()을 호출한 풀은 30초를 기다려도 종료되지 않았습니다.

작업에 취소 표시는 붙었지만, 이미 여러 조각으로 나뉘어 실행 중인 계산은 취소 여부를 확인하지 않기 때문입니다. 병렬 스트림의 계산 코드는 인터럽트 상태도 확인하지 않으므로 true를 넘겨 인터럽트를 요청해도 멈출 계기가 없습니다. 실험한 JDK 21에서는 submit이 Callable을 ForkJoinTask의 기본 cancel을 쓰는 작업으로 감싸는데, 기본 cancel은 인터럽트를 아예 쓰지 않습니다. JDK 22부터는, 이 트랙의 기준인 JDK 25를 포함해, main처럼 Fork/Join 작업자가 아닌 스레드가 submit으로 맡긴 Callable이면 cancel(true)가 그 작업을 실행 중인 작업자에게 인터럽트를 겁니다. 하지만 그 작업자도 인터럽트를 확인하지 않고 나뉜 조각을 실행하는 다른 작업자에게는 인터럽트가 가지 않으므로, 이 경우에도 계산이 멈출 계기는 없습니다. 계산을 중간에 멈추려면 계산 코드가 스스로 중단 조건을 확인해야 하며, 그 방법은 Future 취소와 시간 초과에서 다뤘습니다.

이제 close를 보겠습니다. shutdown()은 새 작업을 받지 않게 하되 이미 제출된 작업은 계속 실행하게 합니다. 이어서 awaitTermination으로 최대 5초 동안 풀이 끝나기를 기다리고, 그 안에 끝나 true가 나오면 그대로 반환합니다. 5초 안에 끝나지 않아 false가 나오면 shutdownNow()로 작업을 취소하고 멈추라고 요청한 뒤 반환합니다. 기다리는 도중 close를 호출한 스레드가 인터럽트되면 InterruptedException을 잡아 shutdownNow()를 호출하고, 예외가 던져지면서 지워진 인터럽트 상태를 Thread.currentThread().interrupt()로 되살립니다.

여기서 주의할 점은 마지막 두 경우 모두 shutdownNow() 뒤에 다시 기다리지 않는다는 것입니다. shutdownNow()는 종료를 요청할 뿐 끝날 때까지 기다리지 않고, 앞에서 본 것처럼 취소를 확인하지 않는 계산은 요청을 받은 뒤에도 계속 돕니다. 실제로 같은 환경에서 병렬 스트림이 계산 중인 풀에 shutdownNow()를 호출하고 1분을 더 기다렸지만 풀은 종료되지 않았습니다. 따라서 close가 반환됐다고 해서 모든 작업이 끝났다고 볼 수는 없습니다. ExecutorService를 안전하게 종료하기에서 본 두 단계 종료처럼 shutdownNow() 뒤에 한 번 더 기다리고 그 결과를 기록하면, 적어도 종료되지 않은 채 반환했다는 사실은 알 수 있습니다.

ForkJoinPool의 shutdownNow()는 ThreadPoolExecutor와 달리 항상 빈 목록을 돌려줍니다. 작업끼리 서로의 결과를 기다리는 관계가 있어도 풀이 종료될 수 있도록, 아직 시작하지 않은 작업까지 실행 중인 작업과 함께 모두 취소하기 때문입니다. 그래서 시작하지 못한 작업을 회수해 나중에 다시 실행하는 방식은 Fork/Join 풀에서는 쓸 수 없습니다.

서버라면 HTTP 요청이 취소되거나 시간이 초과됐을 때 뒤에서 돌던 계산도 멈출지, 이미 얻은 부분 결과를 버릴지도 미리 정해야 합니다. 시간 제한은 '얼마나 기다릴지'만 정해 줍니다. '계산을 언제 멈출지'는 따로 설계해야 하는 문제입니다.

풀은 애플리케이션이 만들고 애플리케이션이 닫습니다

전용 풀을 쓸 때 가장 흔한 실수는 풀의 수명을 요청 하나에 맞추는 것입니다. 먼저 이미 닫힌 풀에 작업을 제출하면 어떻게 되는지 보겠습니다.

import java.util.concurrent.ForkJoinPool;

public final class ClosedDedicatedPoolFailure {
    public static void main(String[] args) {
        ForkJoinPool analyticsPool = new ForkJoinPool(2);
        analyticsPool.close();
        analyticsPool.submit(() -> 42).join();
    }
}

Exception in thread "main" java.util.concurrent.RejectedExecutionException

닫힌 풀은 새 작업을 받지 않으므로 submit이 RejectedExecutionException을 던집니다. JDK 21에서 실행하면 메시지 없이 예외 이름만 나오고, 그 아래로 스택 트레이스가 이어집니다. 이 코드에서는 실수가 한눈에 보이지만 서버에서는 같은 일이 덜 눈에 띄게 일어납니다. 여러 요청이 함께 쓰는 풀을 어느 요청의 처리 코드가 try-with-resources로 감싸거나 close()를 호출해 버리면, 그 뒤에 들어오는 모든 요청이 이 예외로 실패합니다. 요청 하나의 코드가 다른 요청들이 쓰는 자원을 망가뜨린 것입니다.

반대 방향의 실수도 있습니다. 공유를 피하려고 요청마다 새 풀을 만들고 닫는 것입니다. 아래 scoreAsync는 CompletableFuture를 돌려주니 비동기 메서드처럼 보이지만, 메서드 안에서 만든 풀을 try-with-resources로 닫습니다.

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;

public final class PoolPerRequestMistake {
    private static long cpuWork(int value) {
        long result = value;
        for (int round = 0; round < 2_000; round++) {
            result = (result * 1_103_515_245L + 12_345L) & 0x7fff_ffffL;
        }
        return result;
    }

    // 비동기처럼 보이지만, 풀을 닫느라 계산이 끝날 때까지 반환하지 않는다
    static CompletableFuture<Long> scoreAsync() {
        try (ForkJoinPool pool = new ForkJoinPool(2)) {
            return CompletableFuture.supplyAsync(
                    () -> IntStream.range(0, 300_000)
                            .parallel()
                            .mapToLong(PoolPerRequestMistake::cpuWork)
                            .sum(),
                    pool);
        }
    }

    public static void main(String[] args) {
        long start = System.nanoTime();
        CompletableFuture<Long> score = scoreAsync();
        long millis = (System.nanoTime() - start) / 1_000_000;
        System.out.println("반환까지 " + millis + " ms, 반환 시점에 이미 끝났는가: " + score.isDone());
    }
}

반환까지 1211 ms, 반환 시점에 이미 끝났는가: true

시간은 실행할 때마다 달라지지만 true는 매번 같았습니다. return 문이 CompletableFuture를 만들어도 메서드가 실제로 반환되기 전에 try-with-resources가 close()를 호출하고, 전용 풀의 close()는 제출된 작업이 모두 끝날 때까지 기다리기 때문입니다. 그래서 호출자는 계산이 끝날 때까지 붙잡혀 있고, 돌려받은 CompletableFuture는 이미 완료된 상태입니다. 비동기로 처리하려던 의도가 통째로 사라진 것입니다.

요청마다 풀을 만들면 스레드 수도 요청 수에 비례해 늘어납니다. 병렬성 2인 풀을 요청 100개가 동시에 하나씩 만들면 작업자가 최대 200개까지 생기고, 이들이 같은 프로세서를 두고 경쟁합니다. 풀을 만들고 종료하는 비용도 요청마다 반복됩니다. 무엇보다 전용 풀로 얻으려던 '이 기능은 작업자 2개까지'라는 한도가 없어집니다. 한도가 요청 수만큼 곱해지기 때문입니다.

그래서 전용 풀은 애플리케이션 구성 요소가 시작할 때 한 번 만들고, 여러 요청이 함께 쓰며, 애플리케이션이 끝날 때 한 번 닫습니다. 종료할 때는 ExecutorService를 안전하게 종료하기의 서버 종료 순서처럼 새 요청을 받지 않도록 수신부터 멈춘 다음 풀을 닫아야, 종료 도중 들어온 요청이 RejectedExecutionException으로 실패하지 않습니다. 한마디로 풀의 수명은 그 풀을 쓰는 어떤 요청보다도 길어야 합니다.

서로 다른 작업을 조합할 때는 실행기를 지정한 CompletableFuture

병렬 스트림은 같은 연산을 많은 원소에 적용할 때 간결합니다. 하지만 한 요청 안에서 성격이 다른 계산 두 가지를 동시에 실행하고 결과를 합쳐야 한다면 '원소의 목록'이라는 모양이 맞지 않습니다. 이럴 때는 invokeAll과 CompletableFuture로 결과 모으기에서 본 CompletableFuture가 어울립니다. 계산마다 따로 시작하고, 둘 다 끝나면 결과를 합친다는 의존 관계를 코드에 그대로 적을 수 있기 때문입니다.

아래 예제는 활성 사용자 수와 게시글 수를 세는 두 계산을 작업자 두 개짜리 고정 크기 풀에 맡기고, 두 결과를 Summary 레코드로 합칩니다. 핵심은 supplyAsync의 두 번째 인자로 실행기를 넘긴 것입니다. 실행기를 생략하면 두 계산은 공용 풀로 가고, 앞에서 본 공유 문제가 그대로 돌아옵니다.

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

public final class ExplicitExecutorComposition {
    private record Summary(int activeUsers, int publishedPosts) {}

    public static void main(String[] args) {
        try (ExecutorService cpuPool = Executors.newFixedThreadPool(2)) {
            CompletableFuture<Integer> users =
                    CompletableFuture.supplyAsync(
                            () -> IntStream.range(0, 10_000)
                                    .map(value -> value % 3 == 0 ? 1 : 0)
                                    .sum(),
                            cpuPool);
            CompletableFuture<Integer> posts =
                    CompletableFuture.supplyAsync(
                            () -> IntStream.range(0, 8_000)
                                    .map(value -> value % 5 == 0 ? 1 : 0)
                                    .sum(),
                            cpuPool);

            Summary summary = users.thenCombine(posts, Summary::new).join();
            System.out.println(summary);
        }
    }
}

Summary[activeUsers=3334, publishedPosts=1600]

users는 0부터 9,999까지의 정수 가운데 3의 배수를 셉니다. 0, 3, 6, …, 9,999이므로 9,999 ÷ 3 + 1 = 3,334개입니다. posts는 0부터 7,999까지에서 5의 배수를 세므로 0, 5, …, 7,995의 7,995 ÷ 5 + 1 = 1,600개입니다. 두 값은 서로 다른 계산의 결과일 뿐이고, 출력만 보고 두 계산이 실제로 같은 시간에 실행됐는지는 알 수 없습니다.

thenCombine은 두 CompletableFuture가 모두 정상적으로 끝난 뒤에 Summary::new를 호출합니다. 둘 중 하나라도 예외로 끝나면 Summary는 만들어지지 않고 합친 단계도 예외로 끝나며, join()은 그 예외를 CompletionException으로 감싸 던집니다. 또 thenCombine처럼 이름이 Async로 끝나지 않는 메서드에 넘긴 함수는 어느 스레드에서 실행될지 정해져 있지 않습니다. 앞 단계를 완료시킨 cpuPool의 스레드가 실행할 수도 있고, 두 계산이 이미 끝난 뒤라면 thenCombine을 호출한 main이 실행할 수도 있습니다. Summary를 만드는 정도의 가벼운 일이라면 상관없지만, 무거운 단계라면 thenCombineAsync에 실행기를 함께 넘깁니다.

I/O를 기다리는 작업은 Fork/Join 풀에 넣지 않습니다

데이터베이스 조회, 파일 읽기, 원격 API 호출처럼 대부분의 시간을 기다리며 보내는 작업은 CPU 계산과 성질이 다릅니다. Fork/Join 풀의 작업자 수는 보통 프로세서 수에 맞춰 작게 잡기 때문에, 작업자 하나가 응답을 기다리는 동안에는 프로세서가 놀고 있는데도 그 작업자는 다른 계산을 하지 못합니다. 공용 풀이라면 작업자 몇 개만 묶여도 애플리케이션의 모든 병렬 스트림이 쓸 작업자가 줄어듭니다.

그래서 기다리는 작업에는 그 성질에 맞는 실행기를 따로 둡니다. Java 21부터 정식 기능인 가상 스레드(virtual thread)가 좋은 선택지입니다. 가상 스레드는 만드는 비용이 작고, 기다리는 동안에는 대개 운영체제 스레드를 붙잡지 않으므로 기다리는 호출마다 스레드를 하나씩 줘도 부담이 적습니다. 아래 예제는 5ms 동안 잠드는 것으로 I/O를 흉내 낸 load를 키 세 개에 대해 가상 스레드로 실행합니다.

import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

public final class VirtualThreadBlockingBoundary {
    private static String load(String key) throws InterruptedException {
        Thread.sleep(Duration.ofMillis(5));
        return key.toUpperCase();
    }

    public static void main(String[] args) throws Exception {
        List<String> keys = List.of("java", "stream", "pool");
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Callable<String>> calls =
                    keys.stream().<Callable<String>>map(key -> () -> load(key)).toList();
            List<String> values =
                    executor.invokeAll(calls).stream()
                            .map(
                                    future -> {
                                        try {
                                            return future.get();
                                        } catch (Exception exception) {
                                            throw new IllegalStateException(exception);
                                        }
                                    })
                            .toList();
            System.out.println(values);
        }
    }
}

[JAVA, STREAM, POOL]

newVirtualThreadPerTaskExecutor()는 작업마다 가상 스레드를 새로 만들어 실행합니다. invokeAll은 세 작업을 한꺼번에 맡기고 모두 끝날 때까지 기다린 뒤, 맡긴 순서대로 Future 목록을 돌려줍니다. 그래서 각 작업이 끝난 순서와 상관없이 출력은 입력 키의 순서를 따릅니다. 이 예제는 기다리는 작업을 어디에 맡기는지를 보여 줄 뿐이고, 가상 스레드와 병렬 스트림의 속도를 비교하지는 않습니다.

가상 스레드가 모든 문제를 풀어 주지는 않습니다. 가상 스레드를 아무리 많이 만들어도 프로세서는 늘어나지 않으므로, CPU 계산만 하는 작업을 가상 스레드 수천 개로 쪼개도 빨라지지 않습니다. 데이터베이스 연결 수나 원격 서비스가 허용하는 요청 수도 그대로입니다. 이 실행기는 동시에 실행되는 작업 수를 제한하지 않으므로, 연결이 20개뿐인 데이터베이스를 호출한다면 세마포어나 연결 풀로 동시 호출 수를 따로 제한해야 합니다. 그 방법은 가상 스레드와 ScopedValue에서 다룹니다.

작업 종류에 맞는 실행기 고르기

지금까지 본 선택지를 작업 종류별로 모으면 다음과 같습니다.

풀을 나눈다고 자원이 무한히 생기지는 않습니다. 기능마다, 엔드포인트마다 전용 풀을 만들면 전체 스레드 수와 컨텍스트 스위칭이 늘어나고, 풀마다 정한 병렬성을 합치면 프로세서 수를 훌쩍 넘기 쉽습니다. CPU 계산용과 I/O용처럼 자원 종류에 따라 소수의 실행기를 정하고, 어떤 실행기가 있는지와 누가 소유하는지를 한곳에서 관리하는 편이 낫습니다.

운영 중에는 풀의 상태도 지켜봐야 합니다. ForkJoinPool은 작업자의 큐에 쌓인 작업 수(getQueuedTaskCount()), 제출됐지만 아직 시작하지 못한 작업 수(getQueuedSubmissionCount()), 작업을 실행하거나 훔칠 작업을 찾는 중인 작업자 수(getActiveThreadCount()), 제출한 스레드가 아닌 다른 스레드가 실행해 끝낸 작업 수(getStealCount())를 알려 줍니다. 모두 근삿값이므로 한 번 읽은 값보다 시간에 따른 추세를 봅니다. 여기에 요청의 시간 초과 횟수와 느린 쪽 응답 시간(꼬리 지연 시간)을 함께 보면, 풀이 밀리기 시작하는 시점을 일찍 알아챌 수 있습니다.

정리

• 병렬 스트림과, 실행기를 넘기지 않은 CompletableFuture의 비동기 메서드는 JVM에 하나뿐인 공용 풀을 함께 씁니다. 공용 풀의 병렬성은 OpenJDK에서 기본적으로 프로세서 수보다 하나 적고, 공용 풀은 닫을 수 없습니다.
• 병렬 처리는 남는 프로세서가 있을 때만 응답 시간을 줄입니다. 동시 요청이 프로세서를 모두 채우면 병렬 스트림의 이득은 사라지고, 요청마다 걸리는 시간이 다른 요청에 따라 들쭉날쭉해집니다.
• 큰 CPU 계산은 애플리케이션이 소유하는 전용 ForkJoinPool로 나눌 수 있습니다. 전용 풀에 제출한 작업 안에서 실행한 병렬 스트림은 그 풀에서 돌지만, 이것은 JDK 구현에 따른 동작입니다.
• get의 시간 제한은 기다림만 끝내고 계산을 멈추지 않습니다. cancel과 shutdownNow도 취소 여부를 확인하지 않는 계산은 멈추지 못합니다.
• 풀은 애플리케이션이 시작할 때 만들어 여러 요청이 함께 쓰고, 끝날 때 한 번 닫습니다. 요청마다 풀을 만들고 닫으면 비동기 처리가 사라지고 스레드 수가 요청 수에 비례해 늘어납니다.
• 서로 다른 계산의 조합은 실행기를 지정한 CompletableFuture로, I/O를 기다리는 작업은 가상 스레드나 I/O 전용 실행기로 처리하고, 하위 시스템의 동시 호출 수는 따로 제한합니다.

연습 문제

1. 프로세서 4개짜리 서버의 공용 풀

프로세서가 4개인 서버에서 요청 8개가 동시에 병렬 스트림을 실행합니다. 공용 풀의 병렬성은 기본값이고, 다른 프로그램은 CPU를 쓰지 않습니다. 공용 풀의 작업자는 몇 개이고, 계산에 참여하는 스레드는 모두 몇 개인가요? 이 상황에서 병렬 스트림이 요청의 응답 시간을 줄일 수 있는지 설명하세요.

2. 두 합계를 조합한 보고서

사용자 점수 합계와 주문 금액 합계를 작업자 수가 2인 같은 실행기에 각각 맡긴 뒤 record로 조합하세요. 공용 풀을 쓰지 말고, 실행기는 확실히 닫아야 합니다.

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

public final class BoundedReportCompositionSolution {
    private record Report(long userScore, long orderAmount) {}

    public static void main(String[] args) {
        List<Integer> scores = List.of(10, 20, 30, 40);
        List<Integer> orders = List.of(1_000, 2_000, 3_000);

        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
            CompletableFuture<Long> scoreTotal =
                    CompletableFuture.supplyAsync(
                            () -> scores.stream().mapToLong(Integer::longValue).sum(), executor);
            CompletableFuture<Long> orderTotal =
                    CompletableFuture.supplyAsync(
                            () -> orders.stream().mapToLong(Integer::longValue).sum(), executor);

            Report report = scoreTotal.thenCombine(orderTotal, Report::new).join();
            System.out.println(report);
        }
    }
}

Report[userScore=100, orderAmount=6000]

3. 주문 계산이 실패하면

2번 문제에서 사용자 점수 합계는 정상으로 끝났지만 주문 합계 계산이 예외를 던졌다고 합시다. 예외를 다른 값으로 바꾸는 복구 단계는 없습니다. Report[userScore=100, orderAmount=6000] 같은 보고서가 만들어질까요? 주문 계산이 IllegalStateException을 던지도록 바꿔서 join()이 무엇을 던지는지 확인하세요.

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public final class FailedOrderCompositionSolution {
    private record Report(long userScore, long orderAmount) {}

    public static void main(String[] args) {
        List<Integer> scores = List.of(10, 20, 30, 40);

        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
            CompletableFuture<Long> scoreTotal =
                    CompletableFuture.supplyAsync(
                            () -> scores.stream().mapToLong(Integer::longValue).sum(), executor);
            CompletableFuture<Long> orderTotal =
                    CompletableFuture.supplyAsync(
                            () -> {
                                throw new IllegalStateException("주문 저장소 응답 없음");
                            },
                            executor);

            try {
                Report report = scoreTotal.thenCombine(orderTotal, Report::new).join();
                System.out.println(report);
            } catch (CompletionException exception) {
                System.out.println("조합 실패: " + exception.getCause());
            }
            System.out.println("점수 합계만 따로: " + scoreTotal.join());
        }
    }
}

조합 실패: java.lang.IllegalStateException: 주문 저장소 응답 없음
점수 합계만 따로: 100

4. close가 반환된 뒤의 풀

DedicatedAnalyticsPool.close에서 awaitTermination이 false를 반환했다고 합시다. close가 반환된 뒤 풀에 대해 확실히 말할 수 있는 것과 말할 수 없는 것은 무엇인가요? 기다리는 도중 InterruptedException이 났다면 호출자에게는 무엇이 남나요? 종료되지 않은 채 반환했다는 사실을 알 수 있도록 close를 고쳐 보세요.

@Override
public void close() {
    pool.shutdown();
    try {
        if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
            pool.shutdownNow();
            if (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
                System.err.println("analytics pool did not terminate");
            }
        }
    } catch (InterruptedException exception) {
        pool.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

다음 글

이 글로 병렬 스트림과 Fork/Join을 다룬 34장을 마칩니다. 35장에서는 지금까지 써 온 람다와 스트림을 프로그래밍 패러다임이라는 더 큰 그림에서 다시 봅니다. 같은 게시판 통계를 명령형·객체지향·함수형 방식으로 각각 구현해 보며 함수형 프로그래밍을 본격적으로 다룹니다.