[JAVA] 블로킹 방식의 작업 완료 통보_2 (스레드풀_3)

블로킹 방식의 작업 완료 통보_2

작업 처리 결과를 외부 객체에 저장

스레드가 작업 처리를 완료하고 외부 Result 객체에 저장하면, 애플리케이션에서 Result 객체를 사용하여 작업을 진행할 수 있다. (대개 Result 객체는 공유객체가 되어, 두 개 이상의 스레드 작업을 취합할 목적으로 이용된다.)

ExecutorService.submit(Runnable task, V result) 에서 V가 Result 타입이 된다.

class Task implements Runnable{
    Result result;
    Task(Result result){this.result = result;}
    @Override
    public void run(){
        // 작업내용, result 저장
    }
}

class Result{ // 결과 값을 저장하는 임시 객체
    int value;
    synchronized void saveValue(int value){
        this.value = value;
    }
}

처리 결과를 Result(외부객체)에 저장

public class ResultByRuunableExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        System.out.println("=작업 처리 요청=");

        class Task implements Runnable{ // 작업을 정의
            Result result;
            Task(Result result) {this.result = result;} // 외부 객체를 필드에 저장

            @Override
            public void run() {
                int sum = 0;
                for(int i=0; i<=10; i++){
                    sum += i;
                }
                result.addValue(sum);
            }
        }

        /*** 두 가지 작업 처리를 요청 ***/
        Result result = new Result();
        Runnable task1 = new Task(result);
        Runnable task2 = new Task(result);
        Future<Result> future1 = executorService.submit(task1, result);
        Future<Result> future2 = executorService.submit(task2, result);

        /*** 두 가지 작업 결과를 취합 ***/
        try {
            result = future1.get();
            result = future2.get();
            System.out.println("결과 >>>>" + result.sumValue);
            System.out.println("=작업 처리 완료=");
        }catch (Exception e){
            e.printStackTrace();
        }

        executorService.shutdown();
    }
}

class Result{
    int sumValue;
    synchronized void addValue(int value){
        sumValue += value;
    }
}

실행 결과

=작업 처리 요청= 

결과 >>>>110 

=작업 처리 완료=

 

작업 완료 순으로 통보

스레드는 작업 요청 순서에 따라 작업처리가 완료되지 않기 때문에 작업완료 순으로 통보를 받기 위해선 CompletionService를 이용해야 한다.

CompletionService는 처리 완료된 작업을 가져오는 poll()과 task()메소드를 제공한다.

CompletionService 구현 클래스는 ExecutorCompletionService이며, 객체를 생성할 때 생성자 매개값으로 ExecutorService를 제공하면 된다.

ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
CompletionService<V> completionService = new ExecutorCompletionService<V>(executorService);

// while 문은 애플리케이션 종료될 때까지 반복되어야 하므로 스레드 풀의 스레드를 이용하는것이 좋음
// Callable 작업이 있을 때까지 블로킹되었다가 완료된 작업의 Future를 얻고 결과를 얻음(get())
executorService.submit(new Runnable(){ 
    @Override
    public void run(){
        while(true){
            try{
                Future<Integer> future = completionService.task(); // 완료된 작업이 있을 때까지 블로킹, 완료되면 Future 리턴
                int value = future.get(); // get()은 블로킹되지 않고 바로 작업결과 리턴
                System.out.println("결과 : " + value);
            }catch(Exception e){
                break;
            }
        }
    }
});

 

task() 메소드가 리턴하는 완료된 작업은 처리 요청 순서가 아닌것을 명심해야함(먼저 요청한 것이 나중에 완료될 수도 있음)

3개의 Callable 작업을 요청하고, 완료된 순서대로 결과값을 콘솔에 출력

public class CompletionServiceExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

        CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(executorService);

        System.out.println("=작업 처리 요청=");
        for(int i=0; i<5; i++){
            completionService.submit(new Callable<Integer>() { // 스레드풀 작업요청
                @Override
                public Integer call() throws Exception {
                    int sum = 0;
                    for(int i=1; i<=10; i++){
                        sum += i;
                    }
                    return sum;
                }
            });
        }

        System.out.println("=처리 완료된 작업 확인=");
        executorService.submit(new Runnable(){ // 스레드 풀의 스레드에서 실행
            @Override
            public void run() {
                while(true){
                    try{
                        Future<Integer> future = completionService.take();
                        int value = future.get();
                        System.out.println("처리결과 >>>> " + value);
                    }catch(Exception e){
                        System.out.println(e.toString());
                        break;
                    }
                }
            }
        });

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdownNow();
        // shutdown()을 하면 스레드가 끝나지 않는다. (작업 완료된 것이 끝나면 스레드풀이 닫히기 떄문이다.)
        // 쉽게 실수 할 수있다., 나는 shutdown()으로 호출해서 한참 삽질했다.
    }
}

실행 결과

=작업 처리 요청= 

=처리 완료된 작업 확인= 

처리결과 >>>> 55 

처리결과 >>>> 55 

처리결과 >>>> 55 

처리결과 >>>> 55 

처리결과 >>>> 55 

java.lang.InterruptedException