[Java] Thread (9) - ForkJoin 프레임워크

 

 

 

 

 

 

Fork-Join 프레임워크의 등장 배경

 

현재 CPU의 속도의 발전은 한계점에 도달하여, CPU 속도를 향상시키는 것 보다는 코어의 개수를 늘리는 방식을 통해 CPU의 성능을 발전해가고 있습니다.

 

이러한 하드웨어의 변화에 맞춰 프로그래밍도 멀티 코어를 잘 활용할 수 있는 멀티쓰레드 프로그래밍이 점점 더 중요해지고 있습니다.

 

JDK 1.7 버전(자바 7버전)부터 fork-join(포크 - 조인) 프레임워크가 추가되었고, 이는 하나의 작업을 작은 단위로 나눠 처리하는 분할 정복(divide and conquer) 방식을 통하여 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어줍니다.

 

 

포크조인 프레임워크는 어떠한 작업에 대하여 포크(fork)하여 해당 작업이 비동기식으로 실행하기에 충분히 단순해질 때까지 작업을 더 작은 독립 하위 작업으로 재귀적으로 분할합니다.

 

이후 조인(join) 부분이 시작됩니다.

모든 하위 작업의 결과는 단일 결과로 재귀적으로 결합됩니다.

void를 반환하는 작업의 경우 프로그램은 모든 하위 작업이 실행될 때까지 단순히 기다립니다.

 

 

 

 

 

 

 

사용

수행하는 작업에 따라 RecursiveAction과 RecursiveTask둥 하나를 상속받아 구현하야 합니다.

RecursiveAction : 반환값이 없는 작업을 구현할 때 사용합니다.
RecursiveTask : 반환값이 있는 작업을 구현할 때 사용합니다.

 

 

두 클래스 모두 compute()라는 추상 메서드를 가지고 있는데, 우리는 상속을 통해서 이를 구현해주어야 합니다.

 

예시를 통해 살펴보겠습니다.

 

 

 

 

 

예시

1부터 n까지의 합을 계산한 결과를 돌려주는 작업을 구현해보겠습니다.

public class SumTask extends RecursiveTask<Long> { //반환값이 있으므로 RecursiveTask
   private long from;
   private long to;

   public SumTask(long from, long to) {
      this.from = from;
      this.to = to;
   }

   @Override
   protected Long compute() {
      System.out.println("[" + Thread.currentThread().getName()+"] compute()" );
      long size = to - from + 1;

      if (size <= 5 ){//더할 숫자가 5개 이하인 경우
         return sum();//곧바로 더해서 반환
      }

      //범위를 반으로 나눠서 두개의 작업을 생성
      long half = (from + to) / 2;

      SumTask leftSum = new SumTask(from, half);
      SumTask rightSum = new SumTask(half + 1, to);

      leftSum.fork();//작업을 작업 큐에 넣는다

      Long compute = rightSum.compute();

      Long join = leftSum.join();
      return  compute+ join;
   }


   private Long sum() {
      long temp = 0L;

      for ( long i = from; i <= to; i++ ) {
         temp += i;
      }
      return temp;
   }

   public static void main(String[] args) {
      ForkJoinPool forkJoinPool = new ForkJoinPool();

      SumTask sumTask = new SumTask(1L, 30L);
      Long invoke = forkJoinPool.invoke(sumTask);
      System.out.println(invoke);
   }
}

 

 

결과는 다음과 같습니다.

 

 

위의 코드를 이해를 돕기 위해 그림으로 표현해보도록 하겠습니다.

 

코드에서는 작업의 size가 5가 될 때까지 나누지만 그림에서는 2가 될 때까지 나눕니다.

 

더할 숫자의 범위를 반으로 나눠서 한 쪽에서는 fork()를 호출하여 작업 큐에 저장합니다.

하나의 쓰레드는 compute()를 재귀호출하면서 작업을 계속해서 반으로 나눈고, 다른 쓰레드는 fork()에 의해 작업 큐에 추가된 작업을 수행합니다.

 

 

 

 

 

 

다른 쓰레드의 작업 훔쳐오기 (Work-Stealing Algorithm)

fork()가 호출되어 작업 큐에 추가된 작업은 쓰레드풀에 속한 자신의 작업 큐가 비어있는 다른 쓰레드에 의해 실행됩니다.

정확히는 자유 쓰레드(작업 큐가 비어있는 쓰레드)는 다른 쓰레드의 작업 큐에서 작업을 훔치려고 합니다.

이러한 과정을 작업 훔쳐오기(work stealing) 라고 합니다.

 

 

 

 

 

 

fork(), join()

fork()는 작업을 쓰레드의 작업 큐에 넣는 메서드입니다.

fork()의 구현

 

작업 큐에 들어간 작업은 더 이상 나눌 수 없을 때까지 나뉩니다.

즉 compute()로 나누고 fork()로 작업 큐에 넣는 작업이 계속해서 반복됩니다.

 

그리고 나눠진 작업은 각 쓰레드가 골고루 나눠서 처리하고, 작업의 결과는 join()을 호출해서 얻을 수 있습니다.

 

 

 

fork()와 join()은 중요한 차이점이 하나 존재합니다.

 

fork()는 비동기(asynchronous) 메서드이며, join()은 동기(synchronous) 메서드라는 점입니다.

fork() : 해당 작업을 쓰레드 풀의 작업 큐에 넣습니다. 비동기 메서드입니다.
join() : 해당 작업의 수행이 끝날 때까지 기다렸다가 수행이 끝나면 그 결과를 반환합니다. 동기 메서드입니다.

 

 

 

 

 

 

 

 

 

 

Reference

https://www.baeldung.com/java-fork-join

Guide to the Fork/Join Framework in Java | Baeldung