JAVA多線程提高七:Callable與Future的應用

Callable與Runnable

先說一下java.lang.Runnable吧，它是一個接口，在它里面只聲明了一個run()方法：

public interface Runnable { public abstract void run(); }

由於run()方法返回值為void類型，所以在執行完任務之后無法返回任何結果。 
Callable位於java.util.concurrent包下，它也是一個接口，在它里面也只聲明了一個方法call()：

public interface Callable<V> { /** * Computes a result, or throws an exception if unable to do so. * * @return computed result * @throws Exception if unable to compute a result */ V call() throws Exception; }

 

可以看到，這是一個泛型接口，call()函數返回的類型就是傳遞進來的V類型。 
一般情況下Callable是配合ExecutorService來使用的。ExecutorService接口中聲明了若干個submit方法的重載版本：

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

一般情況下我們使用第一個submit方法和第三個submit方法，第二個submit方法很少使用。第一個submit方法里面的參數類型就是Callable。

Future

Future就是對於具體的Runnable或者Callable任務的執行結果進行取消、查詢是否完成、獲取結果。必要時可以通過get方法獲取執行結果，該方法會阻塞直到任務返回結果。

Future類位於java.util.concurrent包下，它是一個接口：

public interface Future<V> { boolean cancel(boolean mayInterruptIfRunning); boolean isCancelled(); boolean isDone(); V get() throws InterruptedException, ExecutionException; V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

在Future接口中聲明了5個方法，下面依次解釋每個方法的作用：

• cancel方法用來取消任務:

當你想要取消你已提交給執行者的任務，使用Future接口的cancel()方法。根據cancel()方法參數和任務的狀態不同，這個方法的行為將不同：

• 1、如果這個任務已經完成或之前的已被取消或由於其他原因不能被取消，那么這個方法將會返回false並且這個任務不會被取消。

• 2、 如果這個任務正在等待執行者獲取執行它的線程，那么這個任務將被取消而且不會開始它的執行。如果這個任務已經正在運行，則視方法的參數情況而定。 cancel()方法接收一個Boolean值參數(mayInterruptIfRunning)。如果參數為true並且任務正在運行，那么這個任務將被取消。如果參數為false並且任務正在運行，那么這個任務將不會被取消。

• isCancelled方法表示任務是否被取消成功，如果在任務正常完成前被取消成功，則返回 true。

• isDone方法表示任務是否已經完成，若任務完成，則返回true；

• get()方法用來獲取執行結果，這個方法會產生阻塞，會一直等到任務執行完畢才返回；

• get(long timeout, TimeUnit unit)用來獲取執行結果，如果在指定時間內，還沒獲取到結果，就直接返回null。

也就是說Future提供了三種功能：

1）判斷任務是否完成；
2）能夠中斷任務；
3）能夠獲取任務執行結果。

因為Future只是一個接口，所以是無法直接用來創建對象使用的，因此就有了下面的FutureTask。

FutureTask

集Runnable、Callable、Future於一身，它首先實現了Runnable與Future接口，然后在構造函數中還要注入Callable對象（或者變形的Callable對象：Runnable + Result），所以FutureTask類既可以使用new Thread(Runnable r)放到一個新線程中跑，也可以使用ExecutorService.submit(Runnable r)放到線程池中跑，而且兩種方式都可以獲取返回結果，但實質是一樣的，即如果要有返回結果那么構造函數一定要注入一個Callable對象，或者注入一個Runnable對象加一個預先給定的結果　

public class FutureTask<V> implements RunnableFuture<V> public interface RunnableFuture<V> extends Runnable, Future<V> { void run(); }

FutureTask提供了2個構造器，可以看到其實實現方式是一樣的

public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; } public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; this.state = NEW; }

事實上，FutureTask只是Future接口的實現類。

使用示例

1.使用Callable+Future獲取執行結果

ExecutorService threadPool =  Executors.newSingleThreadExecutor();
Future<String> future =
    threadPool.submit(
        new Callable<String>() { public String call() throws Exception { Thread.sleep(1000); return "hello"; }; } ); System.out.println("...."); try { System.out.println("get value:" + future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (Exception e) { e.printStackTrace(); }

2.使用Callable+FutureTask獲取執行結果

Executor框架利用FutureTask來完成異步任務，並可以用來進行任何潛在的耗時的計算。一般FutureTask多用於耗時的計算，主線程可以在完成自己的任務后，再去獲取結果。

import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; public class TestThreadPool { public static void main(String[] args) { // 創建線程池 ExecutorService threadPool = Executors.newFixedThreadPool(3); List<FutureTask<String>> tasks = new ArrayList<FutureTask<String>>(); for (int i = 0; i < 10; i++) { FutureTask<String> futureTask = new FutureTask<String>(new ThreadPoolTask(i)); threadPool.submit(futureTask); tasks.add(futureTask); } for (FutureTask<String> futureTask : tasks) { try { // 阻塞一直等待執行完成拿到結果 System.out.println("future result:" + futureTask.get()); // 阻塞一直等待執行完成拿到結果，如果在超時時間內，沒有拿到則拋出異常 // System.out.println("future result:"+futureTask.get(1,TimeUnit.SECONDS)); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } // 捕獲超時異常 // catch (TimeoutException e) { // e.printStackTrace(); // } } System.out.println("--------------------------"); threadPool.shutdown(); } public static class ThreadPoolTask implements Callable<String> { private int value; public ThreadPoolTask(int value) { this.value = value; } public String call() throws Exception { System.out.println("value-----" + value++); Thread.sleep(2000); return String.valueOf(value); } } }

對比使用Callable+Future的實現可知道，使用FutureTask不用去接收submit的返回值，而是自身就繼承了Future，相對方便些。

CompletionService

示例

這里先給出個示例

import java.util.Random; import java.util.concurrent.*; public class CallableAndFuture { ExecutorService threadPool2 = Executors.newFixedThreadPool(10); CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(threadPool2); for(int i=1;i<=10;i++){ final int seq = i; completionService.submit(new Callable<Integer>() { public Integer call() throws Exception { Thread.sleep(new Random().nextInt(2000)); return seq; } }); } for(int i=0;i<10;i++){ try { System.out.println( completionService.take().get()); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (ExecutionException e) { // TODO Auto-generated catch block e.printStackTrace(); } } System.out.println("======"); threadPool2.shutdown(); } }

使用Callable+Future和FutureTask的實現需要定義一個List來保存Future的值，取值得時候還需要通過for循環遍歷，相對來說使用CompletionService更方便。

---

CompletionService方法

ExecutorCompletionService構造方法：

1、ExecutorCompletionService(Executor executor) 使用為執行基本任務而提供的執行程序創建一個 ExecutorCompletionService，並將 LinkedBlockingQueue 作為完成隊列。

2、ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>>>completionQueue) 使用為執行基本任務而提供的執行程序創建一個 ExecutorCompletionService，並將所提供的隊列作為其完成隊列。

返回值	方法
Future<V>	take()獲取並移除表示下一個已完成任務的 Future，如果目前不存在這樣的任務，則等待。
Future<V>	submit(Callable task)提交要執行的值返回任務，並返回表示掛起的任務結果的 Future。
Future<V>	submit(Runnable task, V result)提交要執行的 Runnable 任務，並返回一個表示任務完成的 Future，可以提取或輪詢此任務。
Future<V>	poll()獲取並移除表示下一個已完成任務的 Future，如果不存在這樣的任務，則返回 null。
Future<V>	poll(long timeout, TimeUnit unit)獲取並移除表示下一個已完成任務的 Future，如果目前不存在這樣的任務，則將等待指定的時間（如果有必要）。

---

原理

CompletionService整合了Executor和BlockingQueue的功能。你可以將Callable任務提交給它去執行，然后使用類似於隊列中的take和poll方法，在結果完整可用時獲得這個結果，像一個打包的Future。ExecutorCompletionService是實現CompletionService接口的一個類，並將計算任務委托給一個Executor。

ExecutorCompletionService的實現相當直觀。它在構造函數中創建一個BlockingQueue，用它去保持完成的結果。計算完成時會調用FutureTask中的done方法。當提交一個任務后，首先把這個任務包裝為一個QueueingFuture，它是 FutureTask的一個子類，然后覆寫done方法，將結果置入BlockingQueue中，take和poll方法委托給了BlockingQueue，它會在結果不可用時阻塞。

ExecutorCompletionService的一個構造函數，整合了Executor和BlockingQueue的功能：

public ExecutorCompletionService(Executor executor) { if (executor == null) throw new NullPointerException(); this.executor = executor; this.aes = (executor instanceof AbstractExecutorService) ? (AbstractExecutorService) executor : null; this.completionQueue = new LinkedBlockingQueue<Future<V>>(); }

任務的提交和執行都是委托給Executor來完成。當提交某個任務時，該任務首先將被包裝為一個QueueingFuture，

public Future<V> submit(Callable<V> task) { if (task == null) throw new NullPointerException(); RunnableFuture<V> f = newTaskFor(task); executor.execute(new QueueingFuture(f)); return f; }

 

QueueingFuture是FutureTask的一個子類，通過改寫該子類的done方法，可以實現當任務完成時，將結果放入到BlockingQueue中。

private class QueueingFuture extends FutureTask<Void> { QueueingFuture(RunnableFuture<V> task) { super(task, null); this.task = task; } protected void done() { completionQueue.add(task); } private final Future<V> task; }

 

而通過使用BlockingQueue的take或poll方法，則可以得到結果。在BlockingQueue不存在元素時，這兩個操作會阻塞，一旦有結果加入，則立即返回。public Future<V> take() throws InterruptedException {

 return completionQueue.take(); } public Future<V> poll() { return completionQueue.poll(); }

轉載:java並發編程-Callable與Future