在之前的例子中勉痴，我們使用執(zhí)行器框架都是在主類中提交任務(wù)利朵，等待任務(wù)執(zhí)行完畢后再去處理任務(wù)執(zhí)行的結(jié)果片仿。接下來(lái)我們打算將任務(wù)的提交和結(jié)果的處理都放置到線程中去執(zhí)行。在每個(gè)任務(wù)內(nèi)部提交自己到執(zhí)行器毛俏，然后通過(guò)一個(gè)統(tǒng)一的結(jié)果處理線程來(lái)處理所有任務(wù)執(zhí)行的結(jié)果。

為了解決這個(gè)問(wèn)題饲窿，執(zhí)行器框架為我們提供了一個(gè)CompletionService類煌寇，任務(wù)執(zhí)行線程和結(jié)果處理線程能夠共享這個(gè)類，結(jié)果處理線程便可以在這里渠道已經(jīng)執(zhí)行完畢的任務(wù)的結(jié)果逾雄。CompletionService類的內(nèi)部也是通過(guò)一個(gè)ExecutorService來(lái)提交任務(wù)的阀溶。

首先，創(chuàng)建任務(wù)線程鸦泳，實(shí)現(xiàn)Callable接口银锻。模擬報(bào)表生成過(guò)程。

/**
 * 模擬生成報(bào)告
 *
 * Created by hadoop on 2016/11/3.
 */
public class ReportGenerator implements Callable<String> {
    private String sender;
    private String title;

    public ReportGenerator(String sender, String title) {
        this.sender = sender;
        this.title = title;
    }

    @Override
    public String call() throws Exception {
        long duration = (long)(Math.random() * 10);

        System.out.printf("ReportGenerator: Generator report %s_%s duration %d seconds.\n", sender, title, duration);

        TimeUnit.SECONDS.sleep(duration);

        return sender + "_" + title;
    }
}

然后我們創(chuàng)建任務(wù)提交線程做鹰，這個(gè)線程的構(gòu)造方法接受兩個(gè)參數(shù)击纬，分別是報(bào)表名稱和CompletionService對(duì)象。將報(bào)表生成任務(wù)提交到CompletionService去執(zhí)行钾麸。

import java.util.concurrent.CompletionService;

/**
 * 模擬請(qǐng)求獲取報(bào)告
 *
 * Created by hadoop on 2016/11/3.
 */
public class ReportRequest implements Runnable {
    private String name;
    private CompletionService<String> service;

    public ReportRequest(String name, CompletionService<String> service) {
        this.name = name;
        this.service = service;
    }

    @Override
    public void run() {
        ReportGenerator generator = new ReportGenerator(name, "Report");
        service.submit(generator);
    }
}

下面我們創(chuàng)建任務(wù)結(jié)果處理類更振，來(lái)打印生成的報(bào)表炕桨。這個(gè)類同樣會(huì)拿到CompletionService的引用，然后循環(huán)調(diào)用CompletionService.poll()方法來(lái)從任務(wù)結(jié)果隊(duì)列中獲取執(zhí)行的結(jié)果殃饿，這個(gè)方法接受一個(gè)時(shí)間參數(shù)谋作，如果當(dāng)前結(jié)果隊(duì)列為空，那么則等待這個(gè)時(shí)間乎芳，超時(shí)返回null遵蚜。不帶參數(shù)的poll()方法，如果對(duì)別為空則直接返回null奈惑。

/**
 * 處理報(bào)表結(jié)果
 *
 * Created by hadoop on 2016/11/3.
 */
public class ReportProcessor implements Runnable {
    private boolean end;
    private CompletionService<String> service;

    public ReportProcessor(boolean end, CompletionService<String> service) {
        this.end = end;
        this.service = service;
    }

    @Override
    public void run() {
        while (!end) {
            try {
                Future<String> future = service.poll(20, TimeUnit.SECONDS);

                if (future != null) {
                    System.out.printf("ReportReceiver: received %s\n", future.get());
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

    public void setEnd(boolean end) {
        this.end = end;
    }
}

最后我們創(chuàng)建主方法類吭净。在這里我們創(chuàng)建ExecutorServer并把它賦值給ExecutorCompletionService。之后創(chuàng)建兩個(gè)報(bào)表請(qǐng)求任務(wù)和一個(gè)報(bào)表處理任務(wù)肴甸，同時(shí)持有ExecutorCompletionService的引用寂殉。

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;

/**
 * 在執(zhí)行器中分離任務(wù)執(zhí)行和結(jié)果處理
 *
 * 我們?nèi)绾翁幚碓谝粋€(gè)對(duì)象里發(fā)送任務(wù)給執(zhí)行器，在另一個(gè)對(duì)象里處理任務(wù)執(zhí)行結(jié)果原在。
 * 對(duì)于這種情況Java API提供了CompletionService類
 *
 * CompletionService使用Executor對(duì)象類執(zhí)行任務(wù)友扰。
 *   優(yōu)勢(shì)在于：可以共享CompletionService。
 *   缺點(diǎn)在于：CompletionService獲取的Future對(duì)象只能是已經(jīng)執(zhí)行完畢的任務(wù)庶柿，他沒有辦法控制任務(wù)狀態(tài)村怪，只能處理任務(wù)結(jié)果。
 *
 * 我們創(chuàng)建了一個(gè)ExecutorService浮庐，然后使用這個(gè)ExecutorService來(lái)初始化一個(gè)ExecutorCompletionService<String>(executor)甚负。
 *
 * 首先創(chuàng)建了兩個(gè)ReportRequest任務(wù)，然后在任務(wù)內(nèi)部使用service.submit(generator)方法調(diào)用報(bào)表生成任務(wù)审残。
 *
 * 然后再ReportProcessor中不斷調(diào)用service.poll(20, TimeUnit.SECONDS);方法獲取已經(jīng)執(zhí)行完的結(jié)果梭域，如果當(dāng)前沒有結(jié)果那么等待20秒。
 *
 * CompletionService還提供了兩個(gè)人方法：
 *   poll()：如果沒有任何Future直接返回null搅轿。
 *   take()：如若任務(wù)隊(duì)列中沒有Future那么阻塞知道有可用的Future病涨。
 *
 * Created by hadoop on 2016/11/3.
 */
public class Main {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();

        CompletionService<String> service = new ExecutorCompletionService<String>(executor);

        ReportRequest request1 = new ReportRequest("Face", service);
        ReportRequest request2 = new ReportRequest("Online", service);

        ReportProcessor processor = new ReportProcessor(false, service);

        Thread thread1 = new Thread(request1);
        Thread thread2 = new Thread(request2);
        Thread thread3 = new Thread(processor);

        thread1.start();
        thread2.start();
        thread3.start();

        try {
            thread1.join();
            thread2.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        executor.shutdown();

        try {
            executor.awaitTermination(1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        try {
            TimeUnit.SECONDS.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        processor.setEnd(true);
    }
}

控制臺(tái)中，我們可以看到兩個(gè)任務(wù)的提交信息和結(jié)果處理信息璧坟。

ReportGenerator: Generator report Online_Report duration 4 seconds.
ReportGenerator: Generator report Face_Report duration 1 seconds.
ReportReceiver: received Face_Report
ReportReceiver: received Online_Report