JVM虚拟机&字节码底层术篇幅&JAVA进阶互联网科技Java学习笔记

FutureTask源码分析

2016-12-11  本文已影响329人  一只小哈

对于java的并发编程来说,我们都知道Thread和runnable,这是创建一个线程最基本的两种方法,但是这两种方法创建的线程是不支持对线程的执行结果进行返回的。虽然我们可以通过传递引用的方式实现,但是实现起来未免太复杂。这个时候我们可能要用到Callable,callable是一个JDK提供的一个支持线程返回结果的一个接口,通过实现call方法,能返回指定泛型的变量。

class CallableTask implements Callable<Integer>{

    @Override
    public Integer call() throws Exception {
        System.out.println("call runing");
        Thread.sleep(5000);
        return 1;
    }
}
public class CallableTest {
    
    public static void main(String args[]){
        
        CallableTask task = new CallableTask();
        try {
            System.out.println("call start");
            ExecutorService service = Executors.newSingleThreadExecutor();
            Future fu = service.submit(task);
            System.out.println(fu.get());
            service.shutdown();
            System.out.println("call end");
        } catch (Exception e) {
            e.printStackTrace();
        }
}
}

可以通过线程池去实现任务的提交,任务提交后会返回future对象,通过get方法即可获得返回值。
注意:这里其实是不推荐调用call方法的,实际上直接调用call方法和runnable的run方法效果是一样的。

其实JDK提供了一种更好的提交方式,它可以将Runnable和Callable进行封装,以便于提交到线程池。并且可以对线程有更好的控制,比如取消线程的执行,它就是FutureTask。

FutureTask只是简单的对Callable以及Runnable进行了封装,提供了额外的对线程控制的功能以及阻塞获取请求结果的功能,其实对于线程池的submit方法,对于每一个任务都会封装成一个FutureTask来运行。

    /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

    /**
     * Returns a <tt>RunnableFuture</tt> for the given callable task.
     *
     * @param callable the callable task being wrapped
     * @return a <tt>RunnableFuture</tt> which when run will call the
     * underlying callable and which, as a <tt>Future</tt>, will yield
     * the callable's result as its result and provide for
     * cancellation of the underlying task.
     * @since 1.6
     */
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

那么FutureTask到底是怎么实现的呢?

首先看构造方法:

    /**
     * Creates a <tt>FutureTask</tt> that will, upon running, execute the
     * given <tt>Callable</tt>.
     *
     * @param  callable the callable task
     * @throws NullPointerException if callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
    }

    /**
     * Creates a <tt>FutureTask</tt> that will, upon running, execute the
     * given <tt>Runnable</tt>, and arrange that <tt>get</tt> will return the
     * given result on successful completion.
     *
     * @param runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        sync = new Sync(Executors.callable(runnable, result));
    }

FutureTask可以接受Runnable以及Callable两种类型的参数,在初始化的时候内部构造了一个Sync的AQS实现类的实例,对于runnable类型的线程需要转化成Callable,同时可以指定返回值。
当我们再观察其他方法的时候,几乎都是委托Sync去处理的,那么重点就放在了Sync上。
首先看看Sync里面有几个状态:

        /** State value representing that task is ready to run */
        private static final int READY     = 0;//准备就绪
        /** State value representing that task is running */
        private static final int RUNNING   = 1;//正在运行
        /** State value representing that task ran */
        private static final int RAN       = 2;//运行完毕
        /** State value representing that task was cancelled */
        private static final int CANCELLED = 4;//任务取消

一个FutureTask的实例就在上面几个状态之间进行轮转,当执行线程时调用run方法,run方法又委托Syn的innerRun方法:

    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    public void run() {
        sync.innerRun();
    }
   //首先CAS将status置为RUNING,可以防止结束前重复提交
        void innerRun() {
            if (!compareAndSetState(READY, RUNNING))
                return;

            runner = Thread.currentThread();
            //double check 防止在此之前被cancel
            if (getState() == RUNNING) { // recheck after setting thread
                V result;
                try {
                    result = callable.call();
                } catch (Throwable ex) {
                    setException(ex);
                    return;
                }
                //设置结果
                set(result);
            } else {
                //清除runner,唤醒阻塞线程
                releaseShared(0); // cancel
            }
        }

当执行线程的时候,首先做的是将AQS的状态由READY变成RUNNING,因为Sync是AQS的实现类,这个也是改变AQS的状态,改变状态之后进行double check,此时是为了防止在这之前有Cancel的请求。如果Cancel了,那么releaseShared清除状态并且唤醒get等待的线程。如果为Running状态,接下来调用call方法,这里也就是为什么要提交到线程池执行了,注意call方法调用只是一个方法调用,而不像Thread.start那样会直接返回,并且开启新线程执行。当执行完毕之后,调用Set,Set其实也是委托给Sync的innerSet:

    /**
     * Sets the result of this Future to the given value unless
     * this future has already been set or has been cancelled.
     * This method is invoked internally by the <tt>run</tt> method
     * upon successful completion of the computation.
     * @param v the value
     */
    protected void set(V v) {
        sync.innerSet(v);
    }

        void innerSet(V v) {
            for (;;) {
                int s = getState();
                if (s == RAN)
                    return;
                //收到取消信号,不设置结果,直接返回
                if (s == CANCELLED) {
                    // aggressively release to set runner to null,
                    // in case we are racing with a cancel request
                    // that will try to interrupt runner
                    releaseShared(0);
                    return;
                }
                //设置结果,并设置当前的状态为RAN
                if (compareAndSetState(s, RAN)) {
                    //设置内容
                    result = v;
                    //唤醒阻塞线程
                    releaseShared(0);
                    done();
                    return;
                }
            }
        }

这里在Set的时候呢,首先也是判断状态如果是RAN直接返回,如果取消了,那么唤醒get等待的线程,并且返回。如果都没有,那么设置FutureTask状态为RAN,表示线程执行完了,同时设置restult为返回值,唤醒所有的等待线程。
上面其实在执行前和执行后都做了Cancel的检查,如果取消,无论执行前后都是没有结果set给result的。
接下来看看是怎么实现阻塞等待结果的,首先看get方法:

    /**
     * @throws CancellationException {@inheritDoc}
     */
    public V get() throws InterruptedException, ExecutionException {
        return sync.innerGet();
    }

            V innerGet() throws InterruptedException, ExecutionException {
            //共享锁,没有完成会阻塞在这
            acquireSharedInterruptibly(0);
            //如果已经取消,那么抛出异常
            if (getState() == CANCELLED)
                throw new CancellationException();
            if (exception != null)
                throw new ExecutionException(exception);
            return result;
        }

同样是委托机制,其实关键在于acquireSharedInterruptibly方法。

    /**
     * Acquires in shared mode, aborting if interrupted.  Implemented
     * by first checking interrupt status, then invoking at least once
     * {@link #tryAcquireShared}, returning on success.  Otherwise the
     * thread is queued, possibly repeatedly blocking and unblocking,
     * invoking {@link #tryAcquireShared} until success or the thread
     * is interrupted.
     * @param arg the acquire argument
     * This value is conveyed to {@link #tryAcquireShared} but is
     * otherwise uninterpreted and can represent anything
     * you like.
     * @throws InterruptedException if the current thread is interrupted
     */
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0) //如果目前是RAN状态或者是Cancel状态的话标识已经完成或者结束
            doAcquireSharedInterruptibly(arg);//等待Task运行结束,唤醒阻塞队列
    }
       /**
         * Implements AQS base acquire to succeed if ran or cancelled
         */
       protected int tryAcquireShared(int ignore) {
            return innerIsDone() ? 1 : -1;
        }
        boolean innerIsDone() {
            return ranOrCancelled(getState()) && runner == null;
        }
        private boolean ranOrCancelled(int state) {
            return (state & (RAN | CANCELLED)) != 0;
        }

其实这里还是使用了委托的机制,同时呢采用了一个共享锁去实现同步,共享锁有一个特点就是允许多个线程获取锁,其实这里对于get操作,其实多个线程同时get是没有问题的,并且如果使用独占锁会降低性能,这里引入共享锁感觉是比较巧妙的。

上面代码将的是,首先线程回去check当前FutureTask的状态,如果是RAN或者Cancel,表示线程已经结束,那么直接返回,如果当前不是上面状态,证明此时线程没执行或者没执行完,那么需要阻塞等待,所以执行doAcquireSharedInterruptibly,让线程等待,等待innerSet之后或者Cancel之后的releaseShared。releaseShared会逐步的唤醒所有阻塞在get上的线程,这样所以线程都能get到结果。提高了效率。

FutureTask实现不但简单而且巧妙(比如巧妙的运用了共享锁),最重要的是使用的也是十分广泛:

  1. 做异步处理,对于下载,或者生成PDF这种比较重的场景,我们可以通过将请求异步化,抽象成FutureTask提交到线程池中运行,从而避免占用大量的Worker线程(Tomcat或者RPC框架),导致后面的请求阻塞。

  2. 对于服务的同步调用,我们可以利用FutureTask进行服务的并行调用,而在最后进行结果的汇总,这样就能变串行调用为并行调用,大大的减小请求的时间(类似于Fork-Join)。

最后,异步线程处理和并行处理是个好东西,需要用起来!!!。

上一篇下一篇

猜你喜欢

热点阅读