JAVA線程12 - 新特性：有返回值的線程

1、概述

在Java5以前，線程是沒有返回值的，要實現子線程完成任務後返回值給主線程須要藉助第三方轉存。

在JAVA5開始，有返回值的任務能夠利用Callable接口來實現。
 
執行Callable任務後，能夠獲取一個Future的對象，在該對象上調用get就能夠獲取到Callable任務返回的Object了。

2、示例

import java.util.concurrent.*; 

public class CallableDemo { 
    public static void main(String[] args) throws ExecutionException, InterruptedException { 
        //建立一個線程池 
        ExecutorService pool = Executors.newSingleThreadExecutor(); 
        //建立有返回值的任務 
        Callable callable = new MyCallable("zhangsan"); 
        //執行任務並獲取Future對象 
        Future future = pool.submit(callable); 
        //從Future對象上獲取任務的返回值，並輸出到控制檯 
        System.out.println("等待結果...");
        System.out.println("獲得結果："+future.get().toString()); 
        //關閉線程池 
        pool.shutdown(); 
    } 
} 

class MyCallable implements Callable{ 
    private String name; 

    MyCallable(String name) { 
        this.name = name; 
    } 

    @Override 
    public Object call() throws Exception {
        Thread.sleep(2000);
        return name + "返回的內容"; 
    } 
}

3、CompletionService

1. 簡介

CompletionService用於提交一組Callable任務，其take(）方法返回已完成的一個Callable任務對應的Feture對象。

CompletionService採起的是BlockingQueue<Future<V>>無界隊列來管理Future。則有一個線程執行完畢把返回結果放到BlockingQueue<Future<V>>裏面。就能夠經過completionServcie.take().get()取出結果，同時還有一個poll()方法，這兩個方法的區別以下：
take 獲取並移除表示下一個已完成任務的 Future，若是目前不存在這樣的任務，則等待。(若是須要用到返回值建議用take)
poll 獲取並移除表示下一個已完成任務的 Future，若是不存在這樣的任務，則返回null。

2. CompletionService解決的問題

當向Executor提交批處理任務時，而且但願在它們完成後得到結果，若是用FutureTask，你能夠循環獲取task，並用future.get()去獲取結果，可是若是這個task沒有完成，你就得阻塞在這裏，這個實效性不高，其實在不少場合，其實你拿第一個任務結果時，此時結果並無生成並阻塞，其實在阻塞在第一個任務時，第二個task的任務已經早就完成了，顯然這種狀況用future task不合適的，效率也不高。

3. 本身維護list<Future>和CompletionService的區別

從list中遍歷的每一個Future對象並不必定處於完成狀態，這時調用get()方法就會被阻塞住，若是系統是設計成每一個線程完成後就能根據其結果繼續作後面的事，這樣對於處於list後面的可是先完成的線程就會增長了額外的等待時間。
而CompletionService的實現是維護一個保存Future對象的BlockingQueue。只有當這個Future對象狀態是結束的時候，纔會加入到這個Queue中，take()方法其實就是Producer-Consumer中的Consumer。它會從Queue中取出Future對象，若是Queue是空的，就會阻塞在那裏，直到有完成的Future對象加入到Queue中。

4. 示例

import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletionServiceDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(10);
        CompletionService<Integer> completionServcie = new ExecutorCompletionService<Integer>(pool);
        try {
            for (int i = 0; i < 10; i++) {
                completionServcie.submit(new MyCallable(i));
            }
            for (int i = 0; i < 10; i++) {
                // take 方法等待下一個結果並返回 Future 對象。
                // poll 不等待，有結果就返回一個 Future 對象，不然返回 null。
                System.out.println(completionServcie.take().get());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            pool.shutdown();
        }
    }
}

class MyCallable implements Callable<Integer>{ 
    private int i;

    Task(int i) {
        this.i = i;
    }

    @Override 
    public Object call() throws Exception {
        Thread.sleep(new Random().nextInt(5000));
        System.out.println(Thread.currentThread().getName() + "   " + i);
        return i;
    } 
}

4、參考資料

http://my.oschina.net/jielucky/blog/158839