Callable 源码分析

首先写一个小的示例，源码如下：

public class TestThread {
	public static void main(String[] args) throws Exception{
		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(1);
		Future future = newFixedThreadPool.submit(new TestThread().new MyCallable());
		System.out.println(future.get());
        newFixedThreadPool.shutdown();
	}
	public class MyCallable implements Callable {
		@Override
		public String call() throws Exception {
			return "1111";
		}
	}
}

先拿到线程池的对象，MyCallable内部类实现callable接口 重写了call方法，然后通过submit调用，能通过future对象拿到返回值 源码实现如下：

1. java.util.concurrent.AbstractExecutorService.submit(Callable)方法执行时候

   1.执行 newTaskFor(task)方法 传入callable对象 将该对象包装成RunnableFuture对象 

其实就是创建了一个FutureTask对象 该类实现了Runnable 和Future接口，分别重写了run和

get方法

run方法总结（不考虑异常）：

1.直接调用callable对象的call方法 拿到返回值

2.set值到全局变量outcome中

3.cas设置state为COMPLETING状态

4.唤醒waiter中的所有阻塞进程 （get时候放进去的）

get方法总结

 1. 判断标志位state是否是线程处理结束 没有的话加入阻塞队列 waiter中 park阻塞住

 2. 自旋 当线程被unpark唤醒之后，判断state如果为COMPLETING，直接返回outcome对象

private Object outcome;
.
.
.
//重写run方法
public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();//直接调用Callable对象的call方法 获取返回值result
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);//将异常信息设置到outcome中
                }
                if (ran)
                    set(result);//设置返回值到outcome中
            }
        } finally {
           ...
        }
    }

//设置outcome对象 唤醒waiter链表中的阻塞对象
protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;//设置outcome对象
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();//unpark所有链表中阻塞的线程
        }
    }
.
.
.
private volatile WaitNode waiters;
.
.
.
//重写了get方法
public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);//等待的线程加入链表waiters中  park住所有线程
        return report(s);//当等待的线程被唤醒之后返回outcome的值
    }
.
.
.

2.执行execute方法

  实现类在java.util.concurrent.ThreadPoolExecutor.execute(Runnable)中

 里面就是线程池的知识，抛开线程池的创建过程，简单总结 大概就是

1、创建一个Worker类 传入了FutureTask对象。 该类也实现Runnable接口，继承了AbstractQueuedSynchronizer类，重写了run方法，和lock unlock方法

    run方法里面调用了runWorker  方法 总结就是如下几步

   1.自旋 从阻塞队列中take数据 没有就阻塞  有就执行传入的FutureTask的run方法

   2.执行FutureTask的run方法

细节会在解析会在线程池里面详细分析。