fork/join pool

1. Java 7에 도입
2. Fork/Join:
   • 어떤 계산 작업을 할 때 여러 개로 나누어 계산한 후 결과를 모으는 방식으로 한다.
   • fork: 여러 개로 나누는 것
   • join: 나누어서 작업한 결과를 모으는 것
3. 목적:
   • CPU를 더 쉽게, 효율적으로 사용하기 위해 만들어졌다.
4. work stealing:
   • fork/join에서 작업을 쪼개고 여러 개의 deque에 나눠서 배치한다. 작업이 없는 deque가 바쁜 deque의 작업을 가져가서 해주는 방식이다.
5. fork/join 작업의 기본 수행 개념:
   • 아래와 같이 fork/join이 실행되기 때문에 보통 이 연산은 recursive하게 수행될 때 많이 사용된다.

   if (작업의 단위가 충분히 작을 경우){
   	해당 작업을 수행
   }else{
   	작업을 반으로 쪼개어 두 개의 작업으로 나눔
       두 작업을 동시에 실행시키고, 두 작업이 끝날 때까지 결과를 기다린다.
   }

6. fork/join 기능을 이용하기 위한 방법:
   • RecursiveTask 클래스나 RecursiveAction 클래스를 상속받아 사용한다.
   • 2개 클래스 모두 compute() 메소드를 overriding하여 구현한다.
   • RecursiveAction: 반환값이 없는 작업을 구현할 때 사용
   • RecursiveTask: 반환값이 있는 작업을 구현할 때 사용
7. 예시 코드

import java.util.concurrent.RecursiveTask;

public class GetSum2 extends RecursiveTask<Long> {
    long from, to;

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

    @Override
    protected Long compute() {
        long gap=to-from;

        try{
            Thread.sleep(1000);
        }catch (Exception e){
            e.printStackTrace();
        }

        log("From="+from+" To="+to);
        if(gap<=3){
            long tempSum = 0;
            for(long loop=from;loop<=to;loop++){
                tempSum+=loop;
            }
            log("Return !! "+from+" ~ "+to+" = "+tempSum);
            return tempSum;
        }

        long middle = (from+to)/2;
        GetSum2 sumPre = new GetSum2(from, middle);
        log("Pre    From="+from+" To="+middle);
        sumPre.fork(); // 작업을 deque에 넣는다. 비동기 메서드이다.
        GetSum2 sumPost = new GetSum2(middle+1,to);
        log("Post   From="+(middle+1)+" To="+to);
        return sumPost.compute()+sumPre.join(); // join 동기 메서드, 호출 결과를 기다린다.
        // 왼쪽은 작업 큐에 넣고, 오른쪽은 쪼갠다.
        // 왼쪽: fork(). 작업큐에서대기.compute()
        // 오른쪽: compute() 메서드를 호출해 재귀
    }

    public void log(String message){
        String threadName = Thread.currentThread().getName();
        System.out.println("["+threadName+"] "+message);
    }
}

import java.util.concurrent.ForkJoinPool;

public class ForkJoinSample {
    static final ForkJoinPool mainPool = new ForkJoinPool(); // 스레드 풀을 생성

    public static void main(String[] args) {
        ForkJoinSample sample = new ForkJoinSample();
        //sample.calculate();
        sample.calculate2();
    }

    public void calculate(){
        long from = 0;
        long to = 10;

        GetSum sum = new GetSum(from, to); // 수행할 작업
        Long result = mainPool.invoke(sum); // invoke()를 호출해서 작업을 수행한다.
        System.out.println("Fork Join: Total sum of "+from+" ~ "+to+" = "+result);
    }

    public void calculate2(){
        long from = 0;
        long to = 10;

        GetSum2 sum = new GetSum2(from, to);
        Long result = mainPool.invoke(sum);
        System.out.println("Fork Join: Total sum of "+from+" ~ "+to+" = "+result);
    }
}

8. 참조:
   • 책 자바의 신