서버 요청과 풀 격리
동시 요청의 공용 풀 경쟁을 분석하고 전용 ForkJoinPool, `CompletableFuture`, 블로킹 I/O 실행기의 자원 구분을 설계합니다.
로컬 main에서 병렬 Stream 하나가 빨라도 서버에서는 여러 요청이 동시에 같은 공용 풀을 사용할 수 있습니다.
요청마다 원본을 여러 조각으로 나누면 풀 큐에 작업이 겹치고, 긴 작업 하나가 다른 끝점의 짧은 계산을 밀어낼 수 있습니다.
핵심은 “병렬인가”가 아니라 어떤 자원을 누가 공유하고, 대기와 실패가 어느 요청까지 전파되는가입니다.
종료된 전용 풀에 작업 제출
실행기를 메서드 안에서 만들고 비동기 작업을 반환한 뒤 바로 닫거나, 애플리케이션 생명 주기보다 먼저 닫으면 다음 요청이 거부됩니다.
아래 코드는 실제 RejectedExecutionException을 발생시킵니다.
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 작업을 수행합니다. 출력 시간과 스레드 이름은 환경마다 달라지므로 단언으로 고정하지 않습니다. 완료 요청 수와 체크섬만 기능 결과로 확인합니다.
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);
}
}
}같은 계산을 요청 수 1, 2, 4, 8로 측정하면 포화 이후 지연 시간이 어떻게 증가하는지 볼 수 있습니다. 하지만 이 확인 프로그램의 한 번 출력으로 용량을 정하지 않습니다. 실제 서버 스레드 풀, 컨테이너 CPU 제한, 다른 공용 풀 사용자와 함께 부하를 측정합니다.
ForkJoinPool CPU 예산
특정 분석 기능의 작업이 충분히 크고 CPU 중심이라면 애플리케이션 범위의 전용 풀을 둘 수 있습니다. 요청 처리기는 그 풀에 계산을 제출하고 기한 안에 결과를 기다립니다. 병렬성은 끝점 하나의 희망 스레드 수가 아니라 애플리케이션이 이 기능에 허용한 CPU 예산입니다.
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);
}
}
}시간 제한이 발생했다고 작업이 자동으로 즉시 멈춘다고 가정해서는 안 됩니다.
취소와 interrupt에 반응하는 방법을 계산 함수의 특성에 맞춰 설계합니다.
HTTP 요청이 취소될 때 백그라운드 CPU 작업도 취소할지, 이미 얻은 부분 결과를 버릴지 결정해야 합니다.
명시적 작업 그래프 선택
병렬 Stream은 같은 형태의 원소로 이루어진 컬렉션을 처리할 때 간결합니다.
서로 다른 두 서비스 호출이나 CPU 계산을 동시에 실행하고 결과를 조합하려면 실행기를 명시한 CompletableFuture가 의존성 그래프와 예외 구분을 더 잘 드러냅니다.
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);
}
}
}두 비동기 작업 가운데 하나가 실패하면 조합 단계도 실패합니다. 재시도가 필요하다면 연산별 멱등성과 최대 시도 횟수를 설계합니다. 실행기를 인수로 명시했으므로 공용 풀과 우연히 섞이지 않습니다.
블로킹 I/O와 Fork/Join 풀 분리
데이터베이스, 파일, 원격 API처럼 대부분의 시간을 기다리는 작업은 CPU 작업의 작업 훔치기와는 다른 자원 문제입니다. Java 25의 가상 스레드 실행기는 블로킹 호출마다 플랫폼 스레드를 점유하는 비용을 줄이지만 데이터베이스 연결 수나 원격 서비스의 처리율 제한까지 늘리지는 않습니다. 세마포어, 연결 풀, 비율 제한기로 하위 시스템의 동시성을 제한합니다.
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와 메모리를 두고 경쟁하며 풀 생성·종료 비용도 반복됩니다. 요청 수만큼 작업자가 늘어나는 구조는 과부하를 확대합니다. 애플리케이션이 공유하는 상한이 있는 풀에 요청별 시간 제한, 취소, 요청 수용 제어를 더하는 편이 일반적으로 안전합니다.
연습 문제
사용자 점수 합계와 주문 금액 합계를 같은 상한이 있는 CPU 실행기에서 동시에 계산한 뒤 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);
}
}
}종료 기준은 userScore=100, orderAmount=6000이며 두 작업의 실패가 조합한 CompletableFuture에 전파되는지 확인하는 것입니다.
운영 환경에서는 기한과 취소 정책을 추가합니다.