서버 요청과 풀 격리

로컬 main에서 병렬 Stream 하나가 빨라도 서버에서는 여러 요청이 동시에 같은 공용 풀을 사용할 수 있습니다.

요청마다 원본을 여러 조각으로 나누면 풀 큐에 작업이 겹치고, 긴 작업 하나가 다른 끝점의 짧은 계산을 밀어낼 수 있습니다.

핵심은 “병렬인가”가 아니라 어떤 자원을 누가 공유하고, 대기와 실패가 어느 요청까지 전파되는가입니다.

종료된 전용 풀에 작업 제출

애플리케이션이 공유하는 실행기를 다음 요청보다 먼저 닫으면 새 작업 제출이 거부됩니다. 반면 전용 ForkJoinPool.close()는 이미 제출된 작업을 마칠 때까지 기다리므로, Future를 반환하는 메서드 안에서 호출하면 비동기 반환도 그만큼 늦어질 수 있습니다.

아래 코드는 이미 종료된 전용 풀에 submit을 호출하므로 RejectedExecutionException으로 거부되는 경로입니다.

lab/ClosedDedicatedPoolFailure.java

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();
    }
}

원칙은 실행기의 소유자와 수명을 개별 작업보다 길게 두는 것입니다.

애플리케이션 구성 요소가 풀을 한 번 만들고 여러 요청이 공유하게 하며, 정상 종료 과정에서는 새 작업의 제출을 막고 진행 중인 작업을 마무리합니다.

요청 하나가 풀을 임의로 닫으면 다른 요청의 자원까지 파괴합니다.

여러 요청의 공용 풀 공유

아래 확인 프로그램은 래치로 네 요청의 시작 대기를 함께 해제하고, 각 요청이 병렬 CPU 작업을 수행하게 합니다. 실제 시작 시각과 작업자 배치는 스케줄러가 정합니다.

원문은 경과시간이나 스레드 이름을 출력하지 않고 완료 요청 수와 결과를 XOR한 값만 출력합니다.

src/ConcurrentCommonPoolRequests.java

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.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;
    }

    public static void main(String[] args) throws Exception {
        int requestCount = 4;
        CountDownLatch start = new CountDownLatch(1);
        try (ExecutorService requests = Executors.newFixedThreadPool(requestCount)) {
            List<Future<Long>> results = new ArrayList<>();
            for (int request = 0; request < requestCount; request++) {
                results.add(
                        requests.submit(
                                () -> {
                                    start.await();
                                    return IntStream.range(0, 5_000)
                                            .parallel()
                                            .mapToLong(ConcurrentCommonPoolRequests::cpuWork)
                                            .sum();
                                }));
            }
            start.countDown();

            long checksum = 0;
            for (Future<Long> result : results) {
                checksum ^= result.get();
            }
            System.out.println("completed=" + results.size());
            System.out.println("combined-checksum=" + checksum);
        }
    }
}

모든 get()이 정상 반환하면 completed=4이며 동일한 계산 네 개를 XOR한 값은 0입니다. XOR은 같은 오답도 상쇄할 수 있어 개별 합계의 정확성을 증명하는 검사는 아닙니다.

요청 수 1, 2, 4, 8에 따른 지연을 비교하려면 원문에 없는 시간 계측과 부하 실험을 별도로 설계해야 합니다.

이 확인 프로그램의 출력만으로 서비스 용량을 정하지 않습니다.

실제 서버 스레드 풀, 컨테이너 CPU 제한, 다른 공용 풀 사용자와 함께 부하를 측정합니다.

ForkJoinPool CPU 예산

특정 분석 기능의 작업이 충분히 크고 CPU 중심이라면 애플리케이션 범위의 전용 풀을 둘 수 있습니다.

요청 처리기는 그 풀에 계산을 제출하고 기한 안에 결과를 기다립니다.

병렬성 설정은 애플리케이션의 CPU 예산을 고려해 정할 작업자 병렬성 목표입니다. CPU 코어를 예약하거나 사용량을 강제로 제한하는 설정은 아닙니다. 사용자 풀 안의 병렬 Stream이 그 풀을 사용하는 경로는 JDK 구현에 따른 것이며 Stream API가 직접 풀을 지정하는 계약과는 구분합니다.

src/DedicatedAnalyticsPool.java

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 대기와 사용자 정의 close가 수행하는 일을 분리하면 다음과 같습니다.

결과 대기와 풀 종료의 서로 다른 경계
경계	원문의 호출	보장과 한계
결과 대기	get(timeout)	기한 만료 시 예외 · 작업 취소는 호출하지 않음
종료 요청	shutdown()	새 작업 거부 · 이미 제출한 작업은 수행
종료 확인	awaitTermination(5초)	false이면 shutdownNow() · 그 뒤 재대기 없음
대기 중 인터럽트	catch 블록	shutdownNow() 후 현재 스레드 인터럽트 상태 복원

결과 대기: 원문의 호출: get(timeout)
보장과 한계: 기한 만료 시 예외 · 작업 취소는 호출하지 않음
종료 요청: 원문의 호출: shutdown()
보장과 한계: 새 작업 거부 · 이미 제출한 작업은 수행
종료 확인: 원문의 호출: awaitTermination(5초)
보장과 한계: false이면 shutdownNow() · 그 뒤 재대기 없음
대기 중 인터럽트: 원문의 호출: catch 블록
보장과 한계: shutdownNow() 후 현재 스레드 인터럽트 상태 복원

get의 제한은 결과를 기다리는 구간에 적용됩니다. 제출과 close까지 포함한 전체 호출의 상한이나, 계산 함수가 interrupt에 반응한다는 보장은 아닙니다.

HTTP 요청이 취소될 때 백그라운드 CPU 작업도 취소할지, 이미 얻은 부분 결과를 버릴지 결정해야 합니다.

명시적 작업 그래프 선택

병렬 Stream은 같은 형태의 원소로 이루어진 컬렉션을 처리할 때 간결합니다.

서로 다른 두 서비스 호출이나 CPU 계산을 동시에 실행하고 결과를 조합하려면 실행기를 명시한 CompletableFuture가 의존성 그래프와 예외 구분을 더 잘 드러냅니다.

src/ExplicitExecutorComposition.java

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

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(
                            () ->
                                    java.util.stream.IntStream.range(0, 10_000)
                                            .map(value -> value % 3 == 0 ? 1 : 0)
                                            .sum(),
                            cpuPool);
            CompletableFuture<Integer> posts =
                    CompletableFuture.supplyAsync(
                            () ->
                                    java.util.stream.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);
        }
    }
}

두 비동기 작업 가운데 하나가 실패하면 조합 단계도 실패합니다.

재시도가 필요하다면 연산별 멱등성과 최대 시도 횟수를 설계합니다.

두 supplyAsync 작업은 명시한 실행기에 제출됩니다. thenCombine은 비동기 실행기를 지정하지 않은 후속 동작이므로 공급 작업의 완료 스레드나 다른 완료 호출자에서 실행될 수 있습니다.

블로킹 I/O와 Fork/Join 풀 분리

데이터베이스, 파일, 원격 API처럼 대부분의 시간을 기다리는 작업은 CPU 작업의 작업 훔치기와는 다른 자원 문제입니다.

Java 25의 가상 스레드 실행기는 블로킹 호출마다 플랫폼 스레드를 점유하는 비용을 줄이지만 데이터베이스 연결 수나 원격 서비스의 처리율 제한까지 늘리지는 않습니다.

세마포어, 연결 풀, 비율 제한기로 하위 시스템의 동시성을 제한합니다.

src/VirtualThreadBlockingBoundary.java

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);
        }
    }
}

이 예제는 I/O 모의 작업의 경계를 보여 줄 뿐 가상 스레드와 병렬 Stream의 속도를 비교하지 않습니다.

CPU 중심 작업을 가상 스레드 수천 개로 쪼개도 CPU 코어 수는 늘지 않습니다.

운영 선택 기준

컬렉션 변환이 작거나 값싸면 순차 Stream을 기본값으로 둡니다.
충분히 크고 균일한 CPU 중심 작업은 결과 법칙을 확인한 뒤 병렬 Stream을 측정합니다.
공용 풀의 경쟁이 문제이고 CPU 예산을 분리해야 한다면 애플리케이션이 소유하는 실행기를 둡니다.
서로 다른 작업 사이의 의존성과 개별 시간 제한은 명시적인 CompletableFuture 그래프로 표현합니다.
블로킹 I/O에는 가상 스레드나 I/O 전용 실행기를 사용하고 하위 시스템의 동시성을 제한합니다.
큐 길이, 활성 작업 수, 작업 훔치기 횟수, 시간 제한, 요청의 꼬리 지연 시간을 함께 관찰합니다.

풀 격리는 무한한 자원을 만드는 방법이 아닙니다.

엔드포인트마다 전용 풀을 만들면 전체 스레드 수와 문맥 전환이 늘어납니다.

소수의 자원 유형별로 실행기를 정하고 소유권을 중앙에서 관리합니다.

요청마다 새 ForkJoinPool을 만들면 서로 완전히 격리되지 않나요?

논리적으로 큐는 나뉘지만 같은 CPU와 메모리를 두고 경쟁하며 풀 생성·종료 비용도 반복됩니다.

요청 수만큼 작업자가 늘어나는 구조는 과부하를 확대합니다.

애플리케이션이 공유하는 상한이 있는 풀에 요청별 시간 제한, 취소, 요청 수용 제어를 더하는 편이 일반적으로 안전합니다.

연습 문제

사용자 점수 합계와 주문 금액 합계를 작업자 수가 2인 같은 CPU 실행기에 각각 제출한 뒤 record로 조합하세요. 두 작업이 실제로 시간상 겹쳐 실행됐는지를 결과값만으로 단정하지 않습니다.

공용 풀을 사용하지 않고 실행기를 확실히 닫아야 합니다.

해설 보기

exercise/BoundedReportCompositionSolution.java

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);
        }
    }
}

정상 계산값은 userScore=100, orderAmount=6000입니다. 원문의 main에는 실패 주입이 없으므로 예외 전파를 실행으로 시험한 것은 아닙니다. 실패 전파까지 확인하려면 각 공급 작업의 예외 경로도 별도로 시험해야 합니다.

newFixedThreadPool(2)는 작업자 수를 제한하지만 작업 큐는 무제한입니다. 이 설정만으로 요청 수용량이나 대기시간의 상한이 생기지는 않습니다.

운영 환경에서는 기한과 취소 정책을 추가합니다.