网站链接: element-ui dtcms
当前位置: 首页 > 技术博文  > 技术博文

java中Future的实现原理

2021/5/7 2:55:39 人评论

文章目录执行结果是如何通过Future返回的Callable是如何被执行的总结Callable任务的执行流程执行结果是如何通过Future返回的 上代码: package com.jswdwsx;import java.util.concurrent.Executors; import java.util.concurrent.Future;public class FutureStudy…

文章目录

  • 执行结果是如何通过Future返回的
  • Callable是如何被执行的
  • 总结Callable任务的执行流程

执行结果是如何通过Future返回的

上代码:

package com.jswdwsx;

import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class FutureStudy {

    public static void main(String[] args) {
        Future<Integer> f = Executors.newSingleThreadExecutor().submit(() -> 1);

        System.out.println(f.getClass());
    }
}

输出如下

class java.util.concurrent.FutureTask

由运行结果可以看出我们拿到的Future对象的具体类是FutureTask。
那么我们基于JDK1.8看get方法的实现:

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

    /**
     * @throws CancellationException {@inheritDoc}
     */
    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);
    }

report方法如下:

    /**
     * Returns result or throws exception for completed task.
     *
     * @param s completed state value
     */
    @SuppressWarnings("unchecked")
    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);
    }

总的来说就是根据state变量判断任务是否执行完成(包括正常和异常),如果未完成就等,正常完成就返回结果,其他情况抛出对应异常。

Callable是如何被执行的

我们知道Thread只能执行实现了Runnable接口的任务,那么Callable是如何被执行的呢?
上代码:

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

可以看到在submit方法中以task为参数构造了一个FutureTask对象,然后以此为参数调用execute方法,最后返回该对象。
看到这里我们可以敏锐地察觉到,这个FutureTask类,必然实现Runnable接口和Future接口。因为execute方法只接收Runnable对象,而submit方法的返回值就是一个Future对象。
继续看源码:

public class FutureTask<V> implements RunnableFuture<V> {
	
	//...省略其他
	private Callable<V> callable;
    private Object outcome;
    
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }
    
    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);
        }
    }
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }
    //...省略其他
}
public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}

可以看到FutureTask实现了RunnableFuture接口,而RunnableFuture接口继承了Runnable和Future两个接口。
再看上面FutureTask的run方法,就是执行了通过构造参数参数传入的Callable对象的call方法,并把结果赋值给outcome变量。
所以Thread是执行了FutureTask的run方法,而FutureTask的run方法执行了Callable的call方法。

总结Callable任务的执行流程

可以看到FutureTask类在Callable任务的执行中扮演着十分重要的角色,正是FutureTask整合了三个关键的部分,Callable接口(作为构造方法的参数和成员变量)、Runnable接口(间接实现)和Future接口(间接实现)
至此我们可以梳理出Callable任务的执行流程:

  1. 我们通过实现Callable接口定义一个带返回值的任务
  2. 通过线程池submit方法提交Callable对象
  3. submit方法构造FutureTask交给线程池执行,同时返回FutureTask用于结果的获取
  4. 线程池将FutureTask交给线程执行
  5. 线程执行FutureTask的run方法
  6. FutureTask的run方法执行Callable对象的call方法,并将结果保存在outcome中
  7. 我们通过返回的FutureTask获取outcome中执行结果

相关资讯

    暂无相关的数据...

共有条评论 网友评论

验证码: 看不清楚?