Fork/Join 패턴
일을 작은 단위로 분할한 다음 나중에 합치는 패턴을 말한다. 멀티 스레드 환경에서 여러 스레드에서 실행한 결과를 합치는 걸 생각하면 된다. 이런 Fork/Join 패턴을 잘 사용할 수 있게 해주는 프레임워크인 ForkJoinPool이 있다. 다만, 실무에서 직접 다룰 일은 거의 없다.
ForkJoinPool
ExecutorService처럼 스레드풀을 생성하지만, Fork/Join 환경에 특화된 스레드풀이다. 생성 방법은 new ForkJoinPool()로 생성하면 되는데, 괄호 안에 아무것도 안 넣으면 cpu에 맞춰서 스레드 갯수가 정해지고 아니면은 숫자를 넣어 스레드 갯수를 정해줄 수도 있다.
Pool를 만들고 작업을 수행시키려면 RecursiveTask / RecursiveAction을 구현할 필요가 있다.
RecursiveTask / RecursiveAction
이 둘을 구현하면 compute()라는 메서드르 재정의해야 하며, 임계값을 가지는데 임계값을 넘지 않으면 직접 철, 넘으면 다른 스레드와 분할 처리를 한다.정확히는 할 거 없는 스레드의 작업 큐가 바쁜 스레드의 작업을 훔치는 작업 훔치기 알고리즘이 작동한다.
이 둘의 차이는 Task는 결과를 반환하고, Action은 결과를 반환하지 않는다.
주요 메서드로는 fork()와 join()이 있다.
import java.util.concurrent.RecursiveTask;
// Long 타입의 결과를 반환하는 RecursiveTask
class SumTask extends RecursiveTask<Long> {
private final long start;
private final long end;
private static final long THRESHOLD = 10_000; // 작업을 나눌 기준점
public SumTask(long start, long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
long length = end - start;
// 기준점보다 작으면 직접 계산 (더 이상 쪼개지 않음)
if (length <= THRESHOLD) {
long sum = 0;
for (long i = start; i <= end; i++) {
sum += i;
}
return sum;
}
// 기준점보다 크면 작업을 반으로 나눔
long mid = start + length / 2;
SumTask leftTask = new SumTask(start, mid);
SumTask rightTask = new SumTask(mid + 1, end);
// 왼쪽 작업은 비동기적으로 실행 (fork)
leftTask.fork();
// 오른쪽 작업은 현재 스레드에서 실행하고,
// 왼쪽 작업이 끝날 때까지 기다린 후 결과를 합산 (join)
long rightResult = rightTask.compute();
long leftResult = leftTask.join();
return leftResult + rightResult;
}
}fork()/join()
fork()는 다른 스레드에게 분할하는 건데, 정확히는 스레드가 자신의 작업 큐에 작업을 저장한다. 그러면 작업 훔치기로 다른 스레드가 훔치기 떄문에 다른 스레드에게 작업을 분할하는 것처럼 보이게 된다.
join()는 스레드에서처럼 결과를 기다리고 반환하는 메서드다. fork()로 분활된 적업을 나중에 join으로 합치면 된다.
이렇게 따로 RecursiveTask / RecursiveAction로 작업을 수행 로직을 짠 다음 메인에서 ForkJoinPool을 생성하고 invoke(task)로 스레드 푸에 작업을 넣어주고 결과를 반환받기만 하면 된다.
참고로 어느 단위만큼 작업을 쪼갤 것인지는 상황에 따라 다르다. 너무 많아도 안 좋고, 너무 적어도 좋지 않다. 작업 수는 cpu 코어 수의 4~10배이면 좋긴 하나, 그것도 환경마다 달라진다.
import java.util.concurrent.ForkJoinPool;
public class ForkJoinExample {
public static void main(String[] args) {
// ForkJoinPool 생성 (일반적으로 CPU 코어 수에 맞게 생성됨)
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
// 전체 작업을 생성
SumTask wholeTask = new SumTask(1, 1_000_000);
// 풀에 작업을 넣고 실행 시작 (결과가 나올 때까지 대기)
long result = forkJoinPool.invoke(wholeTask);
System.out.println("결과: " + result);
}
}공유풀
ForkJoinPool을 따로 커스텀으로 설정할 수 있지만, 공용으로 사용하는 공용풀을 쓸 수도 있다. ForkJoinPool.commonPool()로 접근할 수도 있지만, 그냥 작업.invoke()로 공유풀에 작업을 던질 수 있다. 즉, 따로 공유풀을 만드는 로직을 작성하지 않아도 RecursiveTask / RecursiveAction을 구현한 작업에다 invoke을 호출해도 공유풀을 사용할 수 있다.
공유풀을 사용하면 애플리케이션 내에서 알아서 관리를 생명주기, 자원 관리 등을 알아서 해주기 때문에 편해진다. 다만, 말 그대로 공유풀이기 때문에 여러 스레드가 동시에 작업을 요청할 수도 있다. 그러면 공유풀은 한정되어 있어 한 작업을 처리할 동안 다른 작업은 일리게 되는 문제점이 발생한다.
참고로 공유풀의 스레드 수는 cpu 코어 수보다 1개 적게 만들어진다.
