Fork/Join 프레임워크
Fork/Join 프레임워크는 Java 7부터 도입된 병렬 처리를 위한 프레임워크다. 큰 작업을 작은 단위로 분할(fork)하고, 각각을 병렬로 처리한 후 결과를 합치는(join) 분할 정복(divide and conquer) 방식을 사용한다.
핵심 개념 - 분할 정복 (Divide and Conquer)
- Fork(분할): 하나의 큰 태스크를 더 이상 쪼갤 수 없을 때까지 작은 단위로 나눈다.
- Join(병합): 나뉜 작은 태스크들이 완료되면, 그 결과들을 상위 단계로 전달하며 최종 결과를 만들어낸다.
핵심 개념 - Work Stealing
Work Stealing 알고리즘을 사용하는데, 각 스레드가 자신의 작업 큐를 가지고 있다가 작업이 없으면 다른 스레드의 큐에서 작업을 가져와 처리한다. 이를 통해 CPU 코어를 효율적으로 활용할 수 있다.
주요 구성 요소
- ForkJoinPool: 작업을 실행하는 스레드 풀이다. 일반 ExecutorService와 달리 work-stealing 방식으로 동작한다.
- ForkJoinTask: 실행할 작업을 나타내는 추상 클래스이다. 주로 두 가지 하위 클래스를 사용한다. 수행할 작업에 따라 이중 하나를 구현해서 사용한다.
-RecursiveTask<V>: 반환값이 있는 작업을 구현할 때 사용
-RecursiveAction: 반환값이 없는 작업을 구현할 때 사용
두 클래스 모두 compute()라는 추상 메서드를 가지고 있고, 이 추상 메서드를 구현하면 된다.
compute 메서드는 태스크를 서브태스크로 분할하는 로직과 더 이상 분할할 수 없을 때 개별 서브태스크의 결과를 생산할 알고리즘을 정의한다. 따라서 대부분의 compute 메서드 구현은 다음과 같은 형식을 유지한다.
if (태스크가 충분히 작거나 더 이상 분할할 수 없으면) { 순차적으로 태스크 계산(알고리즘) } else { 태스크를 두 서브태스크로 분할(재귀적 호출, Fork) 모든 서브태스크의 연산이 완료될 때까지 기다림 각 서브태스크의 결과를 합칩(Join) }
주요 메서드
fork()
- 현재 실행 중인 태스크를 비동기(Asynchronous) 상태로 전환하고, 이를 해당 스레드의 작업 큐에 넣는 역할을 한다.
- fork()를 호출하면 새로운 태스크가 현재 스레드의 전용 큐(Work Queue)인 Deque(Double Ended Queue)의 'Top' 위치에 push된다.
- 호출 즉시 제어권이 반환되므로(비동), 현재 스레드는 멈추지 않고 바로 다음 코드(보통 다른 태스크의 계산)를 실행할 수 있다.
- 큐에 들어간 태스크는 나중에 해당 스레드에 의해 처리되거나, 노는 다른 스레드에 의해 '작업 훔치기(Work-Stealing)'를 당해 처리된다.
join()
- join()은 fork()로 던져진 작업의 결과를 동기(Synchronous)적으로 기다리는 과정이다. 하지만 일반적인 Thread.join()과는 내부 동작이 완전히 다르다.
- 일반적인 스레드는 join() 호출 시 작업이 끝날 때까지 CPU를 점유하며 아무것도 하지 않고 기다립니다. 하지만 Fork/Join의 join()은 작업이 끝날 때까지 기다리는 동안 다른 작업을 수행한다.
예시 코드
class SumTask extends RecursiveTask<Long> {
private static final int THRESHOLD = 1000;
private int[] array;
private int start, end;
public SumTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start;
// 작업이 충분히 작으면 직접 계산
if (length <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
// 1. 작업을 둘로 나눔 (fork)
int mid = start + length / 2;
SumTask leftTask = new SumTask(array, start, mid);
SumTask rightTask = new SumTask(array, mid, end);
leftTask.fork(); // 2. 비동기 실행
long rightResult = rightTask.compute(); // 2-1. 현재 스레드에서 실행
long leftResult = leftTask.join(); // 3. 결과 대기
return leftResult + rightResult; // 결과 합침
}
}
// 사용
ForkJoinPool pool = new ForkJoinPool();
int[] array = new int[10000];
Long result = pool.invoke(new SumTask(array, 0, array.length));위 코드는 배열의 합을 구하는 예시이다.
- 작업이 작은 경우 직접 계산하고 큰 경우엔 배열을 쪼개어 SumTask 인스턴스를 만든다.
- leftTask.fork() 하여 비동기로 실행한다,
- rightTask.compute()하여 작업을 현 스레드에서 실행한다. rightTask를 fork()로 실행하지 않는 이유는 이 마저 fork()하게 되면 현재 스레드가 놀기 때문이다.
- leftTask.join()하여 실행 결과를 가져온다.
주의점
- Threshold가 너무 낮으면 즉, 작업을 너무 잘게 쪼개면 오히려 쪼개는 비용이 계산 비용보다 커져서 성능이 떨어질 수 있다.
- 보통의 애플리케이션에서는
ForkJoinPool은 한번만 인스턴스화해서 정적 필드에 싱글톤으로 관리한다. - 포크/조인 프레임워크를 이용하는 병렬 계산은 디버깅하기 어렵다.
- 포크/조인 프레임워크를 사용하는 것이 순차 처리보다 무조건 빠르지 않다. 각 서브태스크의 실행시간은 새로운 태스크를 포킹하는 데 드는 시간보다 길어야 포크/조인 프레임워크를 사용하는 것이 순차 처리보다 빠르다. 그러므로 상황에 맞게 결과를 측정하며 사용해야 한다.
Fork/Join 팁
- 두 서브태스크가 모두 시작된 다음에 join을 호출해야 한다.
- RecursiveTask 내에서는 ForkJoinPool의 invoke 메서드를 사용하지 말아야 한다.
- 서브태스크에 fork 메서드를 호출해서 ForkJoinPool의 일정을 조절할 수 있다.
활용 사례
- 재귀적으로 분할 가능한 대용량 데이터 처리
- CPU 집약적 작업의 병렬화
- 배열/컬렉션의 병렬 처리
Fork/Join 프레임워크는 Java 7부터 도입된 병렬 처리를 위한 프레임워크다. 큰 작업을 작은 단위로 분할(fork)하고, 각각을 병렬로 처리한 후 결과를 합치는(join) 분할 정복(divide and conquer) 방식을 사용한다. 주요 구성 요소는 ForkJoinPool과 ForkJoinTask가 있다. ForkJoinPool작업을 실행하는 스레드 풀이며 work-stealing 방식으로 동작한다. 각 스레드가 자신의 작업 큐를 가지고 있다가 작업이 없으면 다른 스레드의 큐에서 작업을 가져와 처리한다. 이를 통해 CPU 코어를 효율적으로 활용할 수 있다. ForkJoinTask는 실행할 작업을 나타내는 추상 클래스이다. 하위 클래스로는 RecursiveTask, RecursiveAction이 있다. 작업의 반환 값이 있는 경우에는 RecursiveTask를, 없는 경우에는 RecursiveAction을 상속받아서 구현하면된다. 이 클래스를 상속받으면 compute()를 오버라이딩 해야한다. compute 메서드는 태스크를 서브태스크로 분할하는 로직과 더 이상 분할할 수 없을 때 개별 서브태스크의 결과를 생산할 로직을 정의한다. Fork/Join 프레임워크를 사용하면 재귀적으로 분할 가능한 대용량 데이터 처리나 컬렉션의 병렬처리를 할 수 있다.
