线程池--相关接口

　　线程池整个体系中涉及到三个和线程有关的接口和一个和线程池有关的接口。

Executor

public interface Executor {

    void execute(Runnable command);
}

 　　线程池中最顶层的接口，只有一个方法。execute方法接受Runnable类型的task并在未来的某个时刻执行task，执行task有可能用线程池里的线程也有可能用新建的线程。

• 方法是execute并非submit
• 参数的类型是Runnable而非和Future相关的接口

Future

　　Future接口是Java世界中所有异步执行结果的占位符的顶层接口，不仅仅在线程池中在Netty中也有广泛的应用。对于一个异步的方法，方法调用会立即返回，无论方法是否执行完毕，所以需要Future作为方法结果的占位符，并借助Future获得异步方法是否执行完毕以及在执行完毕的时候获得结果，以及取消异步方法的执行。

• boolean cancel(boolean mayInterruptIfRunning) 尝试取消一个正在执行的task，如果task已经执行完毕或者由于某种方法无法取消，则返回false。如果task正在执行，则根据 mayInterruptIfRunning的取值来决定是否中断task。其实cancel是通过中断线程来实现的，所以需要被中断的线程提供相应的支持。 当一个task被成功cancel后，isDone isCancelled返回true
• V get() throws InterruptedException, ExecutionException; 等到task执行完毕取回结果，这应该是一个阻塞的方法
• V get(long timeout, TimeUnit unit) 和get类似，只不过加入了超时机制
• boolean isCancelled(); task是否被取消
• boolean isDone(); task是否结束，并不能理解成完成，因为一个task被cancel后isDone也会返回true，即使该task没有完成

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

 Runnable

　　老朋友了，没啥好说的

• Runnable没有泛型
• run方法没有返回值，也没有提供类似Future#get方法获得方法的结果，如果想获得线程的结果必须通过类似共享变量的方式

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

Callable

　　新面孔，和Runnable类似都是用来在新建线程的时候使用

• Callable有返回值
• Callable会抛出异常

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

FutureTask

　　在线程池中真正使用到的一个类，注意这是一个实现了Future接口的类而非接口，间接的实现了Callable和Future接口。它的作用主要有

• 提供了Future接口的实现，即可以通过get方法获取结果，公共cancel设置中断标志位。并且get方法是阻塞的，直到方法执行完毕才会返回。
• 包裹(wrap)一个Callable和Runnable任务对象，在线程池中submit的task并没有直接实现Future接口，但线程池执行任务的策略是异步的，必然需要使用到Future接口，所以在真正执行方法前需要把Runnable和Callable对象包裹为FutureTask对象实现异步性

状态变量

　　每一个FutureTask对象都代表一个submit的task，因为FutureTask在执行的过程中可能会发生很多变化，如执行完毕，异常退出，被cancel等，所以需要一个state来存储FutureTask在执行过程中的状态变化。该state是volatile的，所以对state的修改肯定是CAS的。

  　private volatile int state;
    private static final int NEW          = 0;
    private static final int COMPLETING   = 1;
    private static final int NORMAL       = 2;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED  = 6;

　　而且这些state的所有可能取值都是int的，所以可以通过比较大小的方式来确定FutureTask的执行状态。

 

public boolean isCancelled() {

    return state >= CANCELLED;

}

public boolean isDone() {

    return state != NEW;

}

执行任务

　　执行FutureTask中的任务，把任务的结果放到outcome中。

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();
                    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 interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

获取FutureTask结果

　　如果FutureTask的state<= COMPLETING,说明FutureTask还未执行完毕，调用get方法的线程应该阻塞并等待，否则调用report返回结果。　　

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

　　阻塞调用get方法的线程和AQS阻塞线程思路类似，维护一个队列，线程被阻塞的时候创建Node加入队列，然后park当前线程。

• awaitDone可以在带有超时的get中复用，所以入参中有和时间相关的参数，如果传入的timed=false则不考虑超时。
• awaitDone是中断敏感的，即调用get被挂起的线程如果被中断是会立即返回的，体现在for循环的第一段代码，检查Thread的中断标志位，如果被被设置则抛异常并返回。
• 加入队列的方式是采用CAS，因为可能存在多个线程同时调用get方法被挂起并加入队列，需要考虑并发的安全性。

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);
        }
    }

　　如果state>=COMPLETING，说明FutureTask执行完毕，可以直接返回结果。其实就是吧outcome强转一下。

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);
}

取消任务

• 首先修改FutureTask的状态。只有在state为NEW的时候才会根据 mayInterruptIfRunning的取值把state修改成 INTERRUPTING或者 CANCELLED。如果state不为NEW或者CAS失败都会返回false，CAS失败只能是因为在判断state为NEW和CAS修改这之间state发生了变化。总之只能cancel状态为NEW的FutureTask。
• 在CAS修改state成功后，在 mayInterruptIfRunning为true的情况下设置线程中断标志位。this.runner是FutureTask的一个属性，代表当前正在执行Callable任务的线程。java里不推荐直接使用shutdown线程的方式来停止一个线程的执行，因为会带来并发安全问题，所以在这里取消任务其实是设置线程的中断位。
• 调用 finishCompletion删除所有因调用get而等待在该task上的线程。

    public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {    // in case call to interrupt throws exception
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally {
            finishCompletion();
        }
        return true;
    }