zoukankan      html  css  js  c++  java
  • 并发编程学习笔记(11)----FutureTask的使用及实现

    1. Future的使用

      Future模式解决的问题是。在实际的运用场景中,可能某一个任务执行起来非常耗时,如果我们线程一直等着该任务执行完成再去执行其他的代码,就会损耗很大的性能,而Future接口就是Future的实现,它可以让当前线程将任务交给Future去执行,然后当前线程就可以去干别的事,知道耗时任务执行完成之后,当前线程直接获取结果即可。FutureTask的使用比较简单,只需要先实例化一个Callable对象,重写call方法,再创建一个FutureTask对象,将Callable对象传给FutureTask,创建一个线程,将FutureTask对象传入Thread对象,启动线程即可,接下来就直接上源码:

    package com.company.thread.t5;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    public class FutureDemo {
    
        public static void main(String[] args) {
    
            Callable<Integer> callable = new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    System.out.println("正在计算结果");
                    Thread.sleep(3000);
                    return 1;
                }
            };
            FutureTask<Integer> futureTask = new FutureTask<>(callable);
            Thread thread = new Thread(futureTask);
            thread.start();
            //做点别的
            System.out.println("干点别的");
            try {
                System.out.println("拿到的结果为:" + futureTask.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

    2. FutureTask源码

      FutureTask实现了RunnableFuture,而RunnableFuture同时继承了Runnable和Future接口,所以它既是一个任务,也实现了Future接口的功能。

      从上面使用的步骤来看,我们需要先实例化一个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;
    }

      这是一个接口,具体的实现在重写的call方法中,是需要做的任务的方法所在。

      在使用方法中我们需要将Callable传给FutureTask的构造方法,所以我们先看FutureTask的构造方法:

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

      如果callable为空,则抛出异常,否则执行赋值操作,并将state状态改成NEW,即是初始化状态,这里说一下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的值为0,表示当前为初始化状态。

      FutureTask实现了Runnable,所以必然又run方法去供线程使用,接下来就看FutureTask重写的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);
            }
        }

      第一步判断state是否为初始化状态,不为初始化状态将将当前线程设置为FutureTask的执行线程,设置失败停止方法,经过第一步还能往下执行,则当前的state必然是初始化状态,得到Callable实例,及真正需要执行的任务代码,判断是否为空,且状态是否为初始化状态,判断成功,则执行callable的call()方法,并将执行结果赋值给result,将执行标记ran赋值为ture,执行成功时,ran为true时调用set()方法保存结果,

      set()方法:

      protected void set(V v) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                outcome = v;
                UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
                finishCompletion();
            }
        }

      将状态从NEW偏移到COMPLETING(正在执行中),将结果保存到outcome中,(outcome是一个Object类型,不止可以保存结果,还可以保存异常信息),保存完成后将偏移量改为NORMAL(正常执行完毕的状态),调用finishCompletion()方法。

      finishCompletion()方法:

     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
        }

      因为在执行过程中可能会未执行完成就调用了get方法,此时get方法会将想要获取结果的线程都等待在一个等待队列中,所以finishCompletion()的功能就是遍历等待队列,并将等带队列中所有的有效线程唤醒(unpark)。此时等待get()的线程将会继续执行,

      当调用call方法出现异常时,result未空,ran赋值未false,调用setException()方法;

      setException(Throwable t):

     protected void setException(Throwable t) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                outcome = t;
                UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
                finishCompletion();
            }
        }

      首先更改状态为COMPLETING,将异常信息保存到outcome中,更改状态为EXCEPTIONAL(表示异常执行结束),调用finishCompletion()唤醒等待线程。run执行完成;

      FutureTask还有一个重要的方法就是get()方法,用于获取Callable的call方法返回的结果或执行过程中出现的异常信息。

      get():

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

      当状态小于等于COMPLETING时,表示当前任务还没有处于终止状态,即可能正在初始化或者是正在执行中,调用awaitDone()方法将线程放入等待队列,并等待起来,直到被唤醒,最后返回report(s)执行结果,

    report(s:

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

      如果正常执行结束,返回执行结果,s >= CANCELLED表示线程不可用状态,抛出异常CancellationException,否则就是异常执行结束,抛出ExecutionException(),这样就可以拿到执行结果。

      FutureTask其核心的原理也是等待/唤醒,当调用get方法没有结果时,线程等待,当set()方法赋值成功,表示任务执行完成,有值可以返回时,唤醒等待的线程,get()方法上等待的线程被唤醒,继续执行,最后返回结果。

      

  • 相关阅读:
    EntityFramework之领域驱动设计实践(二)
    博客园开始对X++语言语法高亮的支持
    一种来源于Microsoft Dynamics AX的权限管理设计思想
    使用InternalsVisibleToAttribute控制internal成员的访问
    EntityFramework之领域驱动设计实践:总结
    EntityFramework之领域驱动设计实践(九)
    EntityFramework之领域驱动设计实践(三)
    EntityFramework之领域驱动设计实践(七)
    EntityFramework之领域驱动设计实践(五)
    EntityFramework之领域驱动设计实践 (一)
  • 原文地址:https://www.cnblogs.com/Eternally-dream/p/9755358.html
Copyright © 2011-2022 走看看