실행기에서 작업의 시작과 결과를 분리하는 처리

CompletionService 클래스에는 작업을 수행자에게 보내는 방법이 있고, 실행이 끝난 다음 작업에Future 대상을 가져오는 방법도 있습니다.내부 구현 메커니즘을 보면 CompletionService 클래스는 Executor 객체를 사용하여 작업을 수행합니다.이 동작의 장점은 Completion 서비스 대상을 공유하고 실행기에 작업을 보내고 다른 대상이 작업의 결과를 처리할 수 있다는 것이다.두 번째 방법은 이미 실행된 임무를 위해future 대상을 얻을 수 밖에 없기 때문에 이Future 대상은 임무의 결과를 얻는 데만 사용할 수 있다.
Code

package com.packtpub.java7.concurrency.chapter4.recipe11.task;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * This class simulates the generation of a report. Is a Callable
 * object that will be executed by the executor inside a 
 * CompletionService
 *
 */
public class ReportGenerator implements Callable<String> {

    /**
     * The sender of the report
     */
    private String sender;
    /**
     * The title of the report
     */
    private String title;
    
    /**
     * Constructor of the class. Initializes the two attributes
     * @param sender The sender of the report
     * @param title The title of the report
     */
    public ReportGenerator(String sender, String title){
        this.sender=sender;
        this.title=title;
    }

    /**
     * Main method of the ReportGenerator. Waits a random period of time
     * and then generates the report as a String.
     */
    @Override
    public String call() throws Exception {
        try {
            Long duration=(long)(Math.random()*10);
            System.out.printf("%s_%s: ReportGenerator: Generating a report during %d seconds
",this.sender,this.title,duration);
            TimeUnit.SECONDS.sleep(duration);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String ret=sender+": "+title;
        return ret;
    }
    
}

package com.packtpub.java7.concurrency.chapter4.recipe11.task;

import java.util.concurrent.CompletionService;

/**
 * This class represents every actor that can request a report. For this example,
 * it simply create three ReportGenerator objects and execute them through a 
 * CompletionService
 *
 */
public class ReportRequest implements Runnable {

    /**
     * Name of this ReportRequest
     */
    private String name;
    
    /**
     * CompletionService used for the execution of the ReportGenerator tasks
     */
    private CompletionService<String> service;
    
    /**
     * Constructor of the class. Initializes the parameters
     * @param name Name of the ReportRequest
     * @param service Service used for the execution of tasks
     */
    public ReportRequest(String name, CompletionService<String> service){
        this.name=name;
        this.service=service;
    }

    /**
     * Main method of the class. Create three ReportGenerator tasks and executes them
     * through a CompletionService
     */
    @Override
    public void run() {
            ReportGenerator reportGenerator=new ReportGenerator(name, "Report");
            service.submit(reportGenerator);
    }

}

package com.packtpub.java7.concurrency.chapter4.recipe11.task;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * This class will take the results of the ReportGenerator tasks executed through
 * a CompletinoService
 *
 */
public class ReportProcessor implements Runnable {

    /**
     * CompletionService that executes the ReportGenerator tasks
     */
    private CompletionService<String> service;
    /**
     * Variable to store the status of the Object. It will executes until the variable
     * takes the true value
     */
    private boolean end;
    
    /**
     * Constructor of the class. It initializes the attributes of the class
     * @param service The CompletionService used to execute the ReportGenerator tasks
     */
    public ReportProcessor (CompletionService<String> service){
        this.service=service;
        end=false;
    }

    /**
     * Main method of the class. While the variable end is false, it
     * calls the poll method of the CompletionService and waits 20 seconds
     * for the end of a ReportGenerator task
     */
    @Override
    public void run() {
        while (!end){
            try {
// CompletionService poll() ， Future 。
                Future<String> result=service.poll(20, TimeUnit.SECONDS);
                if (result!=null) {
                    String report=result.get();
                    System.out.printf("ReportReceiver: Report Recived: %s
",report);
                }            
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        
        System.out.printf("ReportSender: End
");
    }

    /**
     * Method that establish the value of the end attribute
     * @param end New value of the end attribute.
     */
    public void setEnd(boolean end) {
        this.end = end;
    }
    
    
}

package com.packtpub.java7.concurrency.chapter4.recipe11.core;

import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import com.packtpub.java7.concurrency.chapter4.recipe11.task.ReportProcessor;
import com.packtpub.java7.concurrency.chapter4.recipe11.task.ReportRequest;

/**
 * Main class of the example creates all the necessary objects and throws the tasks
 *
 */
public class Main {

    /**
     * @param args
     */
    public static void main(String[] args) {
        // Create the executor and thee CompletionService using that executor
        ExecutorService executor=(ExecutorService)Executors.newCachedThreadPool();
        CompletionService<String> service=new ExecutorCompletionService<>(executor);

        // Crete two ReportRequest objects and two Threads to execute them
        ReportRequest faceRequest=new ReportRequest("Face", service);
        ReportRequest onlineRequest=new ReportRequest("Online", service);
        Thread faceThread=new Thread(faceRequest);
        Thread onlineThread=new Thread(onlineRequest);
        
        // Create a ReportSender object and a Thread to execute  it
        ReportProcessor processor=new ReportProcessor(service);
        Thread senderThread=new Thread(processor);
        
        // Start the Threads
        System.out.printf("Main: Starting the Threads
");
        faceThread.start();
        onlineThread.start();
        senderThread.start();
        
        // Wait for the end of the ReportGenerator tasks
        try {
            System.out.printf("Main: Waiting for the report generators.
");
            faceThread.join();
            onlineThread.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        // Shutdown the executor
        System.out.printf("Main: Shuting down the executor.
");
        executor.shutdown();
        try {
            executor.awaitTermination(1, TimeUnit.DAYS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // End the execution of the ReportSender
        processor.setEnd(true);
        System.out.printf("Main: Ends
");

    }

}

작업 원리
범례의 주 클래스에서, 우리는 Executors 공장 클래스의 new Cached Thread Pool () 방법을 호출하여Thread Pool Executor 집행기 대상을 만들었다.그런 다음 이 객체를 사용하여 Completion Service 객체를 초기화합니다. 서비스 완료(Completion Service)는 수행자를 사용하여 작업을 수행하기 때문입니다.그런 다음 ReportRequest 클래스의 submit () 방법을 호출하여 '서비스 완성' 을 이용하여 작업을 수행합니다.
'서비스 완료' 작업이 끝나면 이 작업 중 하나의 작업이 끝납니다. '서비스 완성' 에는 대기열에서 실행되는 것을 제어하기 위해 Future 대상이 저장되어 있습니다.
poll () 방법으로 이 대기열에 접근해서 작업이 완료되었는지 확인하고, 만약 완료되었는지 확인하면 대기열의 첫 번째 요소 (즉 실행이 완료된future 대상) 를 되돌려줍니다.poll()이 Future 객체를 반환하면 대기열에서 Future 객체를 삭제합니다.이 예시에서, 우리는poll () 방법을 호출할 때 두 개의 인자를 전달하여, 대기열에서 작업 결과가 비어 있을 때, 작업 실행이 끝날 시간을 기다리고 있음을 나타낸다.
Completion Service 대상이 만들어지면 Completion Service에서 lianggeReportGenerator 작업을 수행하는 데 사용할 두 개의 ReportRequest 대상이 만들어집니다.ReportProcessor 작업은 실행기에 전송된 두 개의 ReportRequest 작업에 대한 결과를 처리합니다.