FutureTask-详解(二)-ThreadPollExecutor-并发编程(Java)

1 FutureTask

1.1 简介

public class FutureTask<V> implements RunnableFuture<V> 
public interface RunnableFuture<V> extends Runnable, Future<V>
1
2

从继承关系不难看出，该类用于任务执行后的结果获取，任务可以是同步的也可以是异步，主要用于异步获取，调用线程不必阻塞等待任务执行结果，从而提高系统的吞吐量。

1.2 成员变量

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;

private Callable<V> callable;
private Object outcome; 
private volatile Thread runner;
private volatile WaitNode waiters;
1
2
3
4
5
6
7
8
9
10
11
12
13

• state：状态
• NEW~INTERRUPTED：为状态取值
  • NEW=0：新建
  • COMPLETING：正在停止
    • 停止之前还要做一些善后工作
  • NORMAL：正常执行任务得到结果
  • EXCEPTIONAL：产生异常
  • CANCELLED：任务已取消
  • INTERRUNPTING：正在打断
    • 打断之前还要做一些善后工作
  • INTERRUPTED：以打断
• callable：任务
• outcome：任务执行的结果
• runner：当前线程
• waiters：等待执行结果的线程单链表集合

1.3 构造方法

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}

public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}
1
2
3
4
5
6
7
8
9
10
11

我们知道Callable有返回值，而Runnable线程执行并没有返回值，那么它是怎么将执行结果返回的呢？这里看下Executors.callable(runnable, result)是怎么处理的。

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

这样子看来，它的执行结果是我们自己传进去的？是地，如果想要任务自己的执行结果调用Callable线程就好了。

1.4 主要执行流程分析

这里我们主要分析:

• run()：任务执行
• get(),get(long timout, Timeunit unit)：结果获取
• cancel(boolean mayInterruptIfRunning)：取消任务执行

这几个方法，其他自行查看相关文档或者源码。

1.4.1 run任务执行

1.4.1.1 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();
                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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

执行流程分4部分：

• 前期检测
• 正常执行
• 执行异常
• 最终处理

执行流程图如下图1.4.1所示：[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

1.4.1.2 set(result)方法

任务正常执行，返回结果后，会执行set()方法把任务赋值给outcome,上源码：

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        finishCompletion();
    }
}
1
2
3
4
5
6
7

• 执行流程
  • cas操作尝试吧状态由NEW更换为COMPETING
    • 把结果赋值给outcome
    • case操作把state状态设置为NORMAL
    • 执行完成操作finishCompletion()

1.4.1.3 setException(ex)方法

如果在run执行任务的过程中产生异常，catch捕获异常后，会执行该方法，源码如下：

protected void setException(Throwable t) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = t;
        UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
        finishCompletion();
    }
}
1
2
3
4
5
6
7

执行流程

• cas操作尝试吧状态由NEW更换为COMPETING
  • 把异常对象赋值给outcome
  • case操作把state状态设置为EXCEPTIONAL
  • 执行完成操作finishCompletion()

1.4.1.4 finishCompletion()方法

不管是正常执行后set()方法还是异常执行setException()方法，方法内部都会执行该方法，源码如下：

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;        // to reduce footprint
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

该方法主要任务就是清理waitNode上等待的线程，具体过程如下：

• 双层循环的目的就是防止等待线程出现空和不空混杂的情况
• cas尝试吧waiters置为空
  • 因为waiters是单链表结构，通过for内循环清理
  • 获取q.thread赋值给线程变量t
  • 如果t不为空
    • t值为空，同时unpark()唤醒线程，之前调用的park()阻塞
  • 继续获取后继节点，为空中断循环
• done()方法当前类为空，啥也没干，根据应用场景可由子类继承覆盖
• callable置空

1.4.2 get() 执行流程

1.4.2.1 get()方法

源码如下：

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

• 执行步骤
  • 判断状态<=COMPLETING
    • 等等执行结束awaitDone(false,0L)：无时限等等
  • 然后结果,report(s)

1.4.2.2 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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

执行流程图如下：

主要职责就是等待任务执行完毕，且设置结果；否则就阻塞等待。如果中间被中断或取消执行，也会结束。

1.4.2.3 removeWaiter()方法

源码如下：

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                if (q.thread != null)
                    pred = q;
                else if (pred != null) {
                    pred.next = s;
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

主要任务就是如果参数给定的等待节点不为空，把给定节点 thread赋值null。然后移除阻塞等待队列中等待节点thread为空的节点。

1.4.2.4 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);
}
1
2
3
4
5
6
7
8

方法用于返回执行结果：

• 正常执行任务，设置结果后state状态为NORMAL，此时返回结果outcome
• 如果此时状态>=CANCELLED，此时任务被取消，抛出取消任务异常
• 其他状态抛出执行异常。

2 submit

线程池执行任务，处理直接执行execute()方法，如果任务为异常且需要返回值，可以调用submit方法，源码如下：

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}
1
2
3
4
5
6

execute()和submit()执行任务主要区别就是前者不需要返回值而后者有返回值。

3 invokeAll()和invokeAny()

前面讲解的主要是执行单个任务的情况，如果需要批量执行任务，这可以调用invokeAll()或者invokeAny()方法。

• invokeAll()：等待全部任务执行完毕，返回全部执行结果，如果产生异常，把任务取消
• invokeAny()：任意一任务执行完毕，返回该任务执行结果，其余任务取消

invokeAll()和invokeAny()执行流程相对简单，这里不在详述。

4 Executors

在构建线程池对象的时候，建议使用该类。该类通过静态方法返回多种线程池对象，常用三种线程池简单使用可参考之前写的文章详解-ThreadPollExecutor-并发编程(Java)或者自行查阅相关文档。其他的线程池类型，等以后用到的时候在讲解。

我们来看下其中一种线程池的构建，源代码如下：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
1
2
3
4
5

它是直接new ThreadPoolExecutor的对象，那么我们为什么不直接自己new一个，还要在包装一下呢？这里用到了一种设计模式，具体的等后面我们学习设计模式的时候在讲解。