Java 使用 Callable 接口创建线程

使用 Callable 接口创建线程相较于实现 Runnable 接口方式的优点：可以有返回值，可以抛出异常。

 1. 概述

Callable 接口的源码如下所示：@FunctionalInterface 表名是一个函数式的接口，可以有返回值 V （泛型），可以抛出异常 throws Exception。call 方法类似  Runnable 接口 中的 run 方法，作为线程的 执行体。

@FunctionalInterface
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;
}

2. FutureTask

FutureTask 类实现 RunnableFuture接口。

public class FutureTask<V> implements RunnableFuture<V> 

 RunnableFuture接口 实现 Runnable 和 Future。

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

 FutureTask 的 run 方法调用 call 方法 ，等待  call 方法  执行完毕，存储 call 方法的返回值，然后  FutureTask  可以通过 get 方法 获取  call 方法的返回值。接下来看一下 FutureTask 的 源码。

 FutureTask 的 run 方法如下所示：

 public void run() {if (state != NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))return;try {Callable<V> c = callable;if (c != null && state == NEW) {V result;boolean ran;try {result = c.call(); // 调用 call 方法ran = true; // 成功调用则会有返回值} catch (Throwable ex) {result = null;ran = false;setException(ex);}if (ran)set(result); // 返回值}} finally {// runner must be non-null until state is settled to// prevent concurrent calls to run()runner = null;// state must be re-read after nulling runner to prevent// leaked interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}}

 set 方法如下所示，使用 CAS（乐观锁的一种形式） 设置返回值到outcome。

    protected void set(V v) {if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {outcome = v;UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final statefinishCompletion();}}

 get 方法获取 outcome。

    public V get() throws InterruptedException, ExecutionException {int s = state;if (s <= COMPLETING)s = awaitDone(false, 0L);return report(s);}

report 方法如下所示：

    private V report(int s) throws ExecutionException {Object x = outcome;if (s == NORMAL)return (V)x;if (s >= CANCELLED)throw new CancellationException();throw new ExecutionException((Throwable)x);}

3. 代码示例

重写 call 方法：

    @Overridepublic Integer call() throws Exception {int times = 1000;for (int i=0; i<times; i++) {if(i%100 == 0)System.out.println(Thread.currentThread().getName());}return times;}

 使用 FutureTask 类包装 Callable 接口的实现类。

 FutureTask<Integer> task = new FutureTask<Integer>(new MyCallableStudy());

启动线程。

new Thread(task, "callable").start();

获取返回值。

task.get();

完整代码如下所示： 

package callable;import java.util.concurrent.*;public class MyCallableStudy implements Callable<Integer> {@Overridepublic Integer call() throws Exception {int times = 1000;for (int i=0; i<times; i++) {if(i%100 == 0)System.out.println(Thread.currentThread().getName());}return times;}public static void main(String[] args) throws ExecutionException, InterruptedException {FutureTask<Integer> task = new FutureTask<Integer>(new MyCallableStudy());new Thread(task, "callable").start();System.out.println(task.get());}
}

运行截图：