前言

今天我们来分析实现多线程的一种方式，实现Callable接口。这种方式有种特殊的地方，就是可以拿到线程返回值。具体怎么实现的呢？我们来研究下。

栗子

我们先举个简单的栗子来看下Callable接口吧。
要首先明白，线程池提交实现Callable接口的线程后会返回一个Future对象，这个对象里包含程序的运行结果。

public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException,TimeoutException{
        //创建一个线程池
        ExecutorService executor = Executors.newCachedThreadPool();
        Future<String> future = executor.submit(()-> {
                //System.out.println("CallableTest");
                TimeUnit.SECONDS.sleep(5);
                return "CallableTest";
        });
        System.out.println(future.get());
    }
}

可以看到我们可以通过future.get()拿到结果”CallableTest”。
我们也可以设置指定时间后拿到结果，如指定6s后拿到结果。

1	System.out.println(future.get(4,TimeUnit.SECONDS));

可以看到也拿到了返回结果，如果我们设置4s拿到结果，小于程序运行时间5s，可以看到它抛出了超时异常。java.util.concurrent.TimeoutException。

原理

是不是很神奇？

我们来研究下Callable接口获取返回值的原理。

我们先来看看ExecutorService的submit方法，它接受一个Callable对象，返回一个Future对象。

1	<T> Future<T> submit(Callable<T> task);

及它的实现。AbstractExecutorService的submit方法。

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new FutureTask<T>(callable);
}
void execute(Runnable command);

ThreadPoolExecutor的execute是对其实现。

可以看到创建了一个FutureTask对象并执行。

FutureTask对象实现了Runable接口，我们来看下它。

看一下它的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 {
            	//拿到结果设置ran为true
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
            //异常设置结果为空ran为false并设置异常
                result = null;
                ran = false;
                setException(ex);
            }
            //ran为true时放入结果
            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 interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    //会一直挂起知道处理业务的线程完成，唤醒等待线程
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

我们调用get方法时，他先查询线程状态，如果未完成，就调用awaitDone方法。

private int awaitDone(boolean timed, long nanos)
     throws InterruptedException {
     //如果设置了超时时间就获取截止时间
     final long deadline = timed ? System.nanoTime() + nanos : 0L;
     WaitNode q = null;
     boolean queued = false;
     /循环监视线程的状态
     for (;;) {
         if (Thread.interrupted()) {
             removeWaiter(q);
             throw new InterruptedException();
         }

         int s = state;
         if (s > COMPLETING) {
             if (q != null)
                 q.thread = null;
             return s;
         }
         else if (s == COMPLETING) // cannot time out yet
             Thread.yield();
         else if (q == null)
             q = new WaitNode();
         else if (!queued)
             queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                  q.next = waiters, q);
         else if (timed) {
             nanos = deadline - System.nanoTime();
             if (nanos <= 0L) {
                 removeWaiter(q);
                 return state;
             }
             LockSupport.parkNanos(this, nanos);
         }
         else
             LockSupport.park(this);
     }
 }

带有超时时间的get，到达时间后，会判断线程状态，如果未完成，抛出超时异常。

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

总结

因此，带有返回值得异步线程基本上可以这样理解。

由于有返回值，如果未设置等待时间，会等线程执行完成后返回，基本类似同步。其原理是线程运行后，如果未完成，会放入等待队列。直到线程状态变化（完成或者异常等）。如果设置了等待时间，则到时间后无论线程状态是否完成都会返回线程状态。

SakuraTears的博客

Callable原理分析

前言

栗子

原理

总结