[Java] ForkJoinPool과 병렬 처리 도구(언제 무엇을 쓰는가)

개요

대량 컬렉션을 집계하거나, 서로 다른 외부 응용 프로그램 인터페이스(API, Application Programming Interface)를 동시에 호출한 뒤 결과를 합치거나, 배치 작업에서 동시 실행 개수를 제한해야 하는 순간들이 많다. 자바에서는 이 순간에 사용할 수 있는 병렬 처리 인터페이스를 제공한다.

키워드	의미	주의가 필요한 지점
분할 정복(Divide and Conquer)	큰 작업을 작은 작업으로 쪼개 병렬 실행 후 결과를 합침	쪼개는 임계값(threshold)을 너무 작게 잡으면 분할 비용이 이득을 넘어선다
work-stealing	유휴 워커가 다른 워커의 큐에서 작업을 가져와 실행	가져가는 지점과 소유 워커가 쓰는 지점이 다르다는 걸 모르면 락 경합을 오해한다
common pool	ParallelStream/CompletableFuture 기본 비동기 메서드가 공유하는 단일 ForkJoinPool	스레드 수가 코어 수 - 1로 고정된다는 걸 모르고 블로킹 호출을 넣는다
비동기 결과 조합	CompletableFuture의 thenCompose/thenCombine/allOf/anyOf	조합 메서드마다 의존/독립 관계가 달라 섞어 쓰면 의도한 순서가 안 나온다

ForkJoinPool의 분할 정복과 work-stealing

ForkJoinPool은 분할 정복 알고리즘으로 동작한다. 큰 작업을 여러 작업으로 잘게 나누어 여러 워커 스레드가 병렬로 처리하고, 그 결과를 다시 합쳐 큰 작업을 완료한다. 이 작업 단위를 자바는 ForkJoinTask라는 추상 클래스로 표현하고, 결과값이 필요하면 RecursiveTask, 필요 없으면 RecursiveAction을 상속해 구현한다.

invoke(task)를 호출하면 내부적으로 제출과 대기가 순서대로 일어난다. 먼저 poolSubmit이 작업을 큐에 밀어 넣는 제출 경로를 거치고, 그다음 join이 완료 결과를 기다리는 대기 경로로 이어진다. 제출 경로는 다음 순서로 진행된다.

poolSubmit(task)이 실행된다. invoke 내부에서 가장 먼저 실행되는 단계로, 작업을 큐에 넣을 준비를 한다(t1).
호출한 스레드가 워커 스레드인지 외부 스레드인지에 따라 경로가 갈린다.
- 워커 스레드라면 자기 자신의 로컬 큐에 바로 push한다.
- 워커가 아닌 외부 스레드(예: main 스레드)라면 SubmissionQueue를 뒤져 빈 외부 큐를 찾아 등록하거나, 기존 공유 큐를 반환받아 그곳에 push한다.
WorkQueue.push가 실행된다. 어느 경로로 왔든 이 지점에서 작업이 실제로 큐에 등록된다.
signalWork가 실행된다. 필요하면 새 워커를 깨우거나 만들어 방금 등록된 작업을 처리하게 한다.

제출이 끝나면 곧이어 대기 경로가 시작된다. join() 쪽은 단순히 결과를 기다리기만 하지 않는다는 점이 핵심이고, 순서는 다음과 같다.

task.join()이 실행된다. poolSubmit이 작업을 큐에 넣고 나면 invoke는 이어서 join을 호출해 완료 결과를 기다리기 시작한다.
awaitDone()이 완료 여부를 확인한다. join은 곧바로 블로킹하지 않고 작업이 이미 끝났는지부터 확인한다.
아직 끝나지 않았다면 두 가지 중 하나가 일어난다.
- 가져올 작업이 있다면 helpJoin()이 WorkQueue.getSlot으로 작업을 조회하고, 원자적 비교와 교체(CAS, Compare-And-Swap)로 slot을 비운 뒤 즉시 실행(doExec)한다(t6). 이때 가져오는 대상은 항상 다른 워커의 Tail이다. 자기 큐가 비어 있을 때만 다른 워커의 큐를 살펴보기 때문이다. 대기 중인 스레드가 노는 대신 다른 작업을 처리하며 기다리는 지점이 여기다.
- 가져올 작업이 없다면 그제서야 상태값이 완료를 뜻하는 음수가 될 때까지 루프를 돌며 대기한다.

메서드 이름만으로는 큐에 실제로 무엇이 쌓이고, 스레드풀 안에서 워커들이 어떻게 배치되는지 안 보인다. main에서 시작해 SubmissionQueue, Worker A의 deque를 거쳐 steal이 일어나는 지점까지, 한 장면 위에 t1부터 t6까지 화살표로 이어보면 다음과 같다.

절반씩 쪼갠다고 워커마다 끝나는 시간이 같아지지는 않는다. 누군가는 먼저 끝내고 할 일이 없어진다. 이렇게 유휴 상태가 된 워커가 다른 워커의 작업을 가져와 처리하는 것이 work-stealing이다. 가져올 작업이 쌓여 있어야 하니 워커는 각자 자기 큐를 하나씩 갖는다.

이 큐는 양쪽 끝을 쓸 수 있고, 큐를 소유한 워커와 유휴 워커가 서로 다른 끝을 쓴다는 게 핵심이다.

누가	어느 끝	무엇을 가져가나	이름
큐를 소유한 워커	Head	방금 넣은 작은 조각	pop
유휴 워커	Tail	가장 먼저 넣은 큰 조각	steal

노는 시간이 사라지므로 작업이 고르게 쪼개지지 않아도 전체 시간이 가장 느린 워커에 끌려가지 않는다. 기다리는 스레드는 다른 워커의 작업을 처리한다는 점에서 일반 스레드 풀의 작업과 큰 차이를 보인다.

class SumTask(
    private val numbers: LongArray,
    private val start: Int,
    private val end: Int,
) : RecursiveTask<Long>() {
 
    override fun compute(): Long {
        if (end - start <= THRESHOLD) {
            return (start until end).sumOf { numbers[it] }
        }
        val mid = (start + end) / 2
        val left = SumTask(numbers, start, mid)
        val right = SumTask(numbers, mid, end)
 
        left.fork()                        // 왼쪽 절반을 다른 워커에게 맡긴다
        val rightResult = right.compute()  // 오른쪽 절반은 내가 직접 처리
        val leftResult = left.join()       // 왼쪽 결과 대기, 필요하면 helpJoin
 
        return leftResult + rightResult
    }
 
    companion object {
        private const val THRESHOLD = 10_000
    }
}
 
val pool = ForkJoinPool()               // 커스텀 풀: common pool 과 완전히 분리된다
val total = pool.invoke(SumTask(numbers, 0, numbers.size))

threshold를 너무 작게 잡으면 태스크를 쪼개고 큐에 등록하는 비용이 실제 계산량을 넘어서 오히려 느려진다. 반대로 너무 크게 잡으면 코어 수만큼 병렬성을 못 뽑는다. 코어 수와 데이터 크기를 고려해 실측으로 조정하는 값이다.

ParallelStream과 common pool 공유

컬렉션을 parallelStream()으로 바꾸면 Stream 인터페이스의 구현체인 ReferencePipeline이 반환된다. 이 파이프라인은 데이터를 나누는 역할을 하는 Spliterator를 함께 들고 있다. forEach 같은 최종 연산이 호출되는 시점에야 실제로 ForkJoinTask가 생성되고 실행된다. 여기까지는 ForkJoinPool의 일반적인 분할 정복과 같지만, 어떤 풀을 쓰는지가 문제다. parallelStream()은 별도로 지정하지 않는 한 항상 ForkJoinPool.commonPool()을 사용한다.

common pool은 JVM 안에 딱 하나만 존재하는 정적 인스턴스다. 스레드 수는 Runtime.availableProcessors() - 1로 고정된다. 그런데 CompletableFuture.supplyAsync나 runAsync도 executor를 명시하지 않으면 같은 common pool을 쓴다. 즉 서로 다른 서비스 코드가 각자 parallelStream과 CompletableFuture를 별개로 썼다고 생각해도, 실행 시점에는 같은 한정된 스레드 풀을 나눠 쓰는 셈이다.

이 공유가 실제로 문제가 되는 경우는 parallelStream 파이프라인 안에서 블로킹 호출을 하는 순간이다. common pool 스레드 수는 고정돼 있으므로, 그 스레드들이 외부 API 응답을 기다리며 멈춰 있으면 같은 순간 다른 요청이 실행하려던 parallelStream이나 CompletableFuture 작업도 실행할 스레드를 못 찾아 지연된다.

// 잘못된 패턴: parallelStream 안에서 외부 API 를 블로킹 호출
// common pool 스레드가 응답을 기다리는 동안, 같은 순간 다른 요청의
// parallelStream/CompletableFuture 작업도 실행할 스레드를 못 찾는다
val results = orderIds.parallelStream()
    .map { id -> externalApiClient.fetchBlocking(id) }
    .toList()
 
// 올바른 패턴: 블로킹 호출은 전용 ExecutorService(가상 스레드)로 분리한다
Executors.newVirtualThreadPerTaskExecutor().use { executor ->
    val futures = orderIds.map { id ->
        executor.submit(Callable { externalApiClient.fetchBlocking(id) })
    }
    val results = futures.map { it.get() }
}

parallelStream은 순수 연산(CPU 바운드)이고 컬렉션이 충분히 크며 블로킹이 전혀 섞이지 않을 때만 쓴다는 원칙을 지키면 이 문제를 피할 수 있다.

CompletableFuture로 비동기 결과 조합하기

CompletableFuture는 비동기 작업의 결과를 나중에 조합하기 위한 도구다. runAsync는 결과값이 없는 작업, supplyAsync는 결과값이 있는 작업을 실행한다. 두 메서드 모두 executor를 생략하면 내부적으로 ForkJoinPool.commonPool()의 병렬성이 1보다 큰지 확인해, 크면 common pool을 쓰고 그렇지 않으면 작업마다 새 스레드를 하나씩 만든다. 즉 parallelStream과 마찬가지로 기본값은 common pool이라는 뜻이다.

비동기 작업 수행 후 여러 작업의 결과를 합쳐야 할 때가 많은데, CompletableFuture는 상황별로 다른 조합 메서드를 제공한다. thenCompose는 앞 작업의 결과에 의존하는 다음 작업을 이어 붙일 때(flatMap과 같은 성격), thenCombine은 서로 독립적으로 실행된 두 작업의 결과를 합칠 때 쓴다. allOf는 여러 작업이 전부 끝나기를 기다리고, anyOf는 가장 먼저 끝난 하나의 결과만 취한다.

val userFuture = CompletableFuture.supplyAsync { userApi.fetch(userId) }              // common pool 기본값
val orderFuture = CompletableFuture.supplyAsync({ orderApi.fetch(userId) }, ioExecutor) // 전용 풀 명시
 
// 서로 독립적으로 실행된 두 결과를 합친다
val combined = userFuture.thenCombine(orderFuture) { user, order -> UserOrderView(user, order) }
 
// 세 작업이 모두 끝날 때까지 기다린 뒤 각각의 결과를 꺼낸다
CompletableFuture.allOf(userFuture, orderFuture, paymentFuture).join()

이것도 마찬가지로 별도 ExecutorService를 정의하지 않고, supplyAsync에 executor를 넘기지 않으면 입출력(I/O, Input/Output) 바운드 작업도 common pool에서 실행되고, 이는 앞서 본 parallelStream과 같은 지연으로 이어진다.

// 잘못된 패턴: I/O 바운드 작업을 기본 executor(common pool)에 맡긴다
CompletableFuture.supplyAsync { jdbcTemplate.queryForObject(sql, Order::class.java) }
 
// 올바른 패턴: I/O 바운드 작업은 전용 executor 를 명시로 전달한다
CompletableFuture.supplyAsync({ jdbcTemplate.queryForObject(sql, Order::class.java) }, ioExecutor)

ExecutorService로 전용 스레드 풀 쓰기

common pool과 완전히 분리된 실행 환경이 필요하면 ExecutorService를 직접 만든다. 자바 동시성 라이브러리의 실행기 계층은 최상위 인터페이스 Executor부터 시작해 아래로 내려갈수록 기능이 구체화된다. Executor는 Runnable을 받아 실행하는 것만 선언하고, ExecutorService는 여기에 작업 제출과 생명주기 관리(submit, shutdown 등)를 더한다. AbstractExecutorService는 ExecutorService 메서드 일부를 공통 구현해 실제 구현체가 재사용하게 하고, ThreadPoolExecutor가 스레드를 미리 만들어 두고 작업을 스케줄링하는 실제 구현체다.

ForkJoinPool은 AbstractExecutorService를 상속하기 때문에 ExecutorService 타입으로도 다룰 수 있다. 다만 내부 큐 구조와 스케줄링 방식(work-stealing)이 ThreadPoolExecutor와는 완전히 다르다.

common pool을 건드리지 않고 격리된 스레드 풀이 필요할 때, 스레드 수나 큐 정책을 직접 통제해야 할 때, I/O 바운드 작업을 대량으로 처리해야 할 때 ExecutorService를 쓴다. I/O 바운드라면 가상 스레드(Virtual Thread) 전용 실행기를 쓰는 편이 스레드 수 제약에서 자유롭다.

Executors.newVirtualThreadPerTaskExecutor().use { executor ->   // Java 19+, AutoCloseable
    val futures = (1..8).map { taskId ->
        executor.submit(Callable { process(taskId) })
    }
    futures.forEach { println(it.get()) }
}   // use 블록 종료 시 shutdown 과 awaitTermination 을 대신 처리한다

CountDownLatch로 완료만 기다리기

CountDownLatch는 여러 스레드의 작업이 끝났다는 신호만 필요하고 결과값 자체는 필요 없을 때 쓴다. 예를 들어 여러 개의 초기화 스레드가 전부 준비를 마쳐야 다음 단계로 넘어갈 수 있는 상황이다.

내부적으로 CountDownLatch는 카운트 값을 상태(state)로 들고, 이 상태를 AbstractQueueSynchronizer(AQS)라는 동기화 프레임워크로 관리한다. countDown()을 호출한 스레드는 CAS로 state를 원자적으로 1씩 감소시키고, await()을 호출한 스레드는 state가 0이 될 때까지 블로킹된다. state가 0에 도달하는 순간 대기 중이던 모든 스레드가 한 번에 풀려난다.

val latch = CountDownLatch(workerCount)
 
repeat(workerCount) { i ->
    Thread.ofPlatform().start {
        process(i)
        latch.countDown()          // state 를 1 감소
    }
}
 
latch.await()                      // state 가 0 이 될 때까지 블로킹
println("모든 워커 완료")

CountDownLatch는 1회용이다. state가 한 번 0에 도달하면 다시 카운트를 채울 수 없다. 여러 라운드에 걸쳐 스레드를 반복해서 모아야 하면 CyclicBarrier를, 동시 접근 허용 개수를 제한하려면 Semaphore를 쓴다. 두 도구 모두 CountDownLatch와 같은 AQS 기반이지만 재사용 가능하다는 점이 다르다.

언제 무엇을 쓰는가

다섯 도구는 겹치는 부분이 있어 보이지만 판단 기준은 명확하다. CPU 바운드인지 I/O 바운드인지, 결과값이 필요한지, common pool과 격리해야 하는지를 순서대로 물어보면 대부분 답이 나온다.

상황	적합한 도구	이유
순수 CPU 바운드 분할 정복(배열 합산, 정렬 등)	ForkJoinPool(RecursiveTask/RecursiveAction)	work-stealing으로 코어를 고르게 활용한다
컬렉션을 병렬로 가공하되 블로킹이 전혀 없음	ParallelStream	코드가 간결하고 common pool을 그대로 써도 무리가 없다
컬렉션 가공 중 블로킹 I/O 나 외부 호출이 섞임	ParallelStream 대신 전용 ExecutorService	common pool 고갈이 서비스 전체로 번진다
서로 다른 API 를 병렬 호출한 뒤 결과를 조합	CompletableFuture(thenCombine/allOf)	비동기 콜백 조합에 특화된 API 다
I/O 바운드 다수 작업, 결과값이 필요함	ExecutorService(가상 스레드) + Future	스레드 수 제약 없이 작업마다 새 스레드를 쓸 수 있다
N 개 작업이 전부 끝날 때까지 단순 대기, 결과값 불필요	CountDownLatch	완료 신호만 필요할 때 오버헤드가 가장 적다

common pool을 쓰는 도구(ParallelStream, CompletableFuture 기본값)와 격리된 풀을 쓰는 도구(ExecutorService, 커스텀 ForkJoinPool)를 구분하는 것이 중요하다. 블로킹 호출이 조금이라도 섞이면 common pool에서 빠져나와야 한다는 원칙 하나만 지켜도 사이드 이펙트 대부분을 피할 수 있다. 가상 스레드의 기본 스케줄러도 ForkJoinPool이기 때문에 work-stealing 구조가 자바 동시성에서 중요한 역할을 한다.

참고

CompletableFuture: mangkyu.tistory.com/263
ExecutorService: mangkyu.tistory.com/259
ForkJoinPool: upcurvewave.tistory.com/653

dpdpdndn 블로그

탐색기

[Java] ForkJoinPool과 병렬 처리 도구(언제 무엇을 쓰는가)

[Java] ForkJoinPool과 병렬 처리 도구(언제 무엇을 쓰는가)

개요

ForkJoinPool의 분할 정복과 work-stealing

ParallelStream과 common pool 공유

CompletableFuture로 비동기 결과 조합하기

ExecutorService로 전용 스레드 풀 쓰기

CountDownLatch로 완료만 기다리기

언제 무엇을 쓰는가

참고

그래프 뷰

목차