• 深入理解Java Callable接口


    概述
    Callable和Runnbale一样代表着任务,区别在于Callable有返回值并且可以抛出异常。其使用如下:

    public class CallableDemo {

    static class SumTask implements Callable<Long> {

    @Override
    public Long call() throws Exception {

    long sum = 0;
    for (int i = 0; i < 9000; i++) {
    sum += i;
    }

    return sum;
    }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("Start:" + System.nanoTime());
    FutureTask<Long> futureTask = new FutureTask<Long>(new SumTask());
    Executor executor=Executors.newSingleThreadExecutor();
    executor.execute(futureTask);
    System.out.println(futureTask.get());
    System.out.println("End:" + System.nanoTime());
    }

    }
    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
    从上面的代码可以看到,使用到了一个FutureTask的变量并且还可以得到Callable执行的结果,那么这个FutureTask是什么呢?

    分析
    Future接口
    Future是一个接口,代表了一个异步计算的结果。接口中的方法用来检查计算是否完成、等待完成和得到计算的结果。当计算完成后,只能通过get()方法得到结果,get方法会阻塞直到结果准备好了。如果想取消,那么调用cancel()方法。其他方法用于确定任务是正常完成还是取消了。一旦计算完成了,那么这个计算就不能被取消。

    FutureTask类
    FutureTask类实现了RunnableFuture接口,而RunnnableFuture接口继承了Runnable和Future接口,所以说FutureTask是一个提供异步计算的结果的任务。
    FutureTask可以用来包装Callable或者Runnbale对象。因为FutureTask实现了Runnable接口,所以FutureTask也可以被提交给Executor(如上面例子那样)。

    FutureTask的状态
    FutureTask中有一个表示任务状态的int值,初始为NEW。定义如下:

    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;
    1
    2
    3
    4
    5
    6
    7
    8
    可能的状态转换包括:
    - NEW -> COMPLETING -> NORMAL
    - NEW -> COMPLETING -> EXCEPTIONAL
    - NEW -> CANCELLED
    - NEW -> INTERRUPTING -> INTERRUPTED

    构造方法
    FutureTask一共有两个构造方法,如下:

    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
    12
    第一个构造方法好理解;第二个方法是将Runnbale和结果组合成一个Callable,这个可以通过Excutors.callable()方法得出结论:

    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
    从上面可以看到RunnableAdapter实现了Callable并且在call方法中调用了Runnable的run方法,然后将结果返回,这其实就是一个适配器模式啊。
    所以说两个构造方法最终都是得到了一个Callable以及设置了初始状态为NEW。

    run方法
    当将FutureTask提交给Executor后,Executor执行FutureTask时会执行其run方法,下面看一下run方法中做了哪些事情。

    public void run() {
    //如果状态不为NEW或者CAS当前执行线程失败,直接返回
    if (state != NEW ||
    !UNSAFE.compareAndSwapObject(this, runnerOffset,
    null, Thread.currentThread()))
    return;
    //尝试调用Callable.call
    try {
    Callable<V> c = callable;
    if (c != null && state == NEW) {
    V result;
    boolean ran;
    try {
    result = c.call();
    ran = true;
    } catch (Throwable ex) {
    //出现异常了,调用setException方法
    result = null;
    ran = false;
    setException(ex);
    }
    //如果成功了,调用set方法
    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
    33
    34
    35
    36
    37
    从上面可以看到,任务可以被执行的前提是当前状态为NEW以及CAS当前执行线程成功,也就是runner值,代表执行Callable的线程。从这个看到run方法就是调用Callable的call方法,然后如果出现异常了就调用setException方法,如果成功执行了,那么调用set方法,下面我们分别来看这几种情况。

    set方法
    当Callable成功执行后,会调用set方法将结果传出。源码如下:

    protected void set(V v) {
    //完成NEW->COMPLETING->NORMAL状态转换
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    outcome = v;
    UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
    finishCompletion();
    }
    }
    1
    2
    3
    4
    5
    6
    7
    8
    从上面可以看到,将outcome变量赋值为结果,并将state状态更新,最后调用finishCompletion()方法。finishCompletion()方法将移除和通知所有等待线程,这个方法后面再说。下面先看setException方法。

    setException方法
    setException方法如下:

    //完成NEW->COMPLETING->EXCEPTIONAL状态转换
    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
    8
    从上面看到,该方法和set方法类似,完成状态转换,将结果设置为Throwable并调用finishCompletion通知和移除等待线程。

    get方法
    当想得到FutureTask的结算结果时,调用get方法,get方法可以允许多个线程调用,下面的例子展示了多个线程调用get的情况。

    public static void main(String[] args) throws ExecutionException, InterruptedException {
    System.out.println("Start:" + System.nanoTime());
    FutureTask<Long> futureTask = new FutureTask<Long>(new SumTask());
    Executor executor=Executors.newSingleThreadExecutor();
    executor.execute(futureTask);
    for(int i=0;i<5;i++){
    executor.execute(new Runnable() {
    @Override
    public void run() {
    try {
    System.out.println("get result "+futureTask.get());
    } catch (InterruptedException e) {
    e.printStackTrace();
    } catch (ExecutionException e) {
    e.printStackTrace();
    }
    }
    });
    }
    System.out.println(futureTask.get());
    System.out.println("End:" + System.nanoTime());
    }
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    该例子展示了一共有5个线程想得到FutureTask的结果,一旦调用get,那么该线程就会阻塞。
    FutureTask的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方法,这个方法会将调用的线程挂起;否则直接调用report方法返回结果。
    在前面set和setException方法中可以得出结论:当状态从NEW变为COMPLETING后,才会将outcome赋值,也就是状态是NEW或者COMPLETING时,outcome都还未赋值,也就意味着计算仍在进行,那么此时想要get到结果,就必须等待。下面先看下awaitDone方法是如何将调用线程阻塞的。awaitDone的两个参数分别表示是否定时,以及定时的时间多少。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;
    //如果状态大于COMPLETING,说明已经计算已经完成了
    if (s > COMPLETING) {
    if (q != null)
    q.thread = null;
    return s;
    }
    //状态是COMPLETING,在set和setException方法中可以看到处于该状态马上就会进入下一个状态
    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
    37
    38
    39
    40
    41
    42
    43
    44
    上面的代码中有一个WaitNode类,该类表示等待节点,保存等待的线程以及下一个节点,是一个单链表结构,其定义如下:

    static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
    }
    1
    2
    3
    4
    5
    awaitDone方法中进入死循环后,主要有几步:
    1. 如果线程被中断了,移除节点,抛出异常
    2. 如果状态大于COMPLETING,那么直接返回
    3. 如果状态是COMPLETING,在set和setException可以看到,处于COMPLETING是一个暂时状态,很快就会进入下一个状态,所以这儿就调用了Thread.yield()方法让步一下
    4. 如果状态是NEW且节点为null,那么创建一个节点
    5. 如果还没有将当前线程加入队列,那么将当前线程加入到等待队列中。由于WaitNode是一个单链表结构,FutureTask中保存了waiters的变量,就可以沿着该变量得到所有等待的线程
    6. 如果限制了时间,那么计算出生出超出时间,挂起指定时间。当解除挂起时,如果计算还未完成,那么将会由于没有时间了,调用removeWaiter方法移除节点。
    7. 如果没有限制时间,那么将线程无限挂起

    上面几种情况下,都涉及了移除节点,removeWaiter方法就是删除单链表中一个节点的实现。
    当线程被解除挂起,或计算已经完成后,将会get方法中将会调用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
    9
    10
    11
    从上面可以看到report会根据任务的状态不同返回不同的结果。
    - 如果计算正常结束,即状态是NORMAL,那么返回正确的计算结果
    - 如果计算被取消了,即状态大于等于CANCELLED,那么抛出CancellationException
    - 如果计算以异常结束,即状态是EXCEPTIONAL,那么抛出ExecutionException

    finishCompletion方法
    在set方法和setException方法中,当将结果赋值后,都调用了finishCompletion方法来移除和通知等待线程。由于get方法中可以挂起了一群等待节点,那么当结果被计算出来了,自然应该通知那些等待线程。finishCompletion的实现如下:

    private void finishCompletion() {
    //如果有等待线程,从头开始解除挂起
    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
    25
    finishCompletion的实现比较简单,就是遍历等待线程的单链表,释放那些等待线程。当线程被释放后,那么在awaitDone的死循环中就会进入下一个循环,由于状态已经变成了NORMAL或者EXCEPTIONAL,将会直接跳出循环。
    释放了所有线程后,将会调用done()方法,FutureTask的done()方法默认没有任何实现,子类可以在该方法中调用完成回调以及记录操作等等。
    上面的方法分析完了FutureTask的主要流程,包括调用get线程的阻塞、run方法执行、计算结果的返回。下面再来看一些取消、查看状态的方法。

    cancel方法
    cancel方法用于取消Callable的计算。参数mayInterruptIfRunning指明是否应该中断正在运行的任务,返回值表示取消是否成功了。其源码如下:

    public boolean cancel(boolean mayInterruptIfRunning) {
    if (!(state == NEW &&
    UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
    mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
    return false;
    try {
    //如果需要中断
    if (mayInterruptIfRunning) {
    try {
    Thread t = runner;
    if (t != null)
    t.interrupt();
    } finally {
    //最终状态INTERRUPTED
    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
    }
    }
    } finally {
    //释放等待线程
    finishCompletion();
    }
    return true;
    }
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    从上面可以看到如果是需要中断正在执行的任务,那么状态转换将会是NEW->INTERRPUTING->INTERRUPTED;如果不需要中断正在执行的任务,那么状态转换将会是NEW->CANCELD。不管是否中断,最终都会调用finishCompletion()完成对等待线程的释放。
    当这些线程释放后,再进入到awaitDone中的循环时,返回的状态将会是大于等于CANCELD,在report方法中将会得到CancellationException异常。

    isDone方法
    Future接口中isDone方法表明任务是否已经完成了,如果完成了,那么返回true,否则false。下面是FutureTask的实现:

    public boolean isDone() {
    return state != NEW;
    }
    1
    2
    3
    可以看到只要状态从初始状态NEW完成了一次转换,那么就说明任务已经被完成了。

    总结
    Callable是一种可以返回结果的任务,这是它与Runnable的区别,但是通过适配器模式可以使Runnable与Callable类似。Future代表了一个异步的计算,可以从中得到计算结果、查看计算状态,其实现FutureTask可以被提交给Executor执行,多个线程可以从中得到计算结果。Callable和Future是配合使用的,当从Future中get结果时,如果结果还没被计算出来,那么线程将会被挂起,FutureTak内部使用一个单链表维持等待的线程;当计算结果出来后,将会对等待线程解除挂起,等待线程就都可以得到计算结果了。
    ---------------------
    作者:xingfeng_coder
    来源:CSDN
    原文:https://blog.csdn.net/qq_19431333/article/details/77483763
    版权声明:本文为博主原创文章,转载请附上博文链接!

  • 相关阅读:
    Delphi的属性Property
    Delphi数组
    Delphi中record和packed record的区别
    Delphi基本数据类型---枚举、子界、集合、数组
    Delphi的枚举类型
    Java创建线程的细节分析
    java concurrency: ThreadLocal及其实现机制
    java concurrency: daemon线程
    java volatile关键字的理解
    Java动态代理
  • 原文地址:https://www.cnblogs.com/GarfieldEr007/p/10166044.html
走看看 - 开发者的网上家园