FutureTask与Callable运用

Posted 2020-10-01

　　最近项目需要用到FutureTask异步获取执行结果，并与Callable结合起来运用。

　　首先，看下FutureTask执行过程：FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或Callable的任务给FutureTask，

直接调用其run方法或者放入线程池执行，然后通过FutureTask的get方法异步获取执行结果。

       下面直接看代码。

       Service.java   ---服务类

//1、new一个线程池，属于成员变量，在项目启动时就初始化
private static final ExecutorService exec =  Executors.newFixedThreadPool(100);
public FutureTask<List> getFutureTask() {
        //2、创建FutureTask任务对象，传参Callable
        FutureTask<List> ft = new FutureTask<List>(new Callable<List>() {
            //实现Callable的call（）方法
            @Override
            public List call() throws Exception {
                //3、进行业务操作
                List re = "";
                //4、返回业务结果
                return re;
            }
        });
        //5、把FutureTask任务提交到线程池
        exec.submit(ft);
        //6、并返回FutureTask
        return ft;
    }

Action.java----业务类

//6、获取任务，传参objList（业务参数）
FutureTask<List> futureTask = taskService.getFutureTask();
//8、get()获取业务操作结果。
List taskList = futureTask.get(20, TimeUnit.SECONDS);

    下面看源码实现

   

public class FutureTask<V> implements RunnableFuture<V>
    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; // non-volatile, protected by state reads/writes

     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
    }
    //。。。。。。省略
}

FutureTask类实现了RunnableFuture接口，同时RunnableFuture继承了Runnable接口和Future接口。

state修饰变量是volatie，对整个线程是可见的。

outcome ，属于non-valatie，是为了在读写时候做了自我保护

FutureTask提供了2个构造器：FutureTask(Callable<V> callable)和FutureTask(Runnable runnable, V result) 分别实现了Callable和Runnable方法。

调用这个两个构造器时，state状态为NEW

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

FutureTask(Callable<V> callable)实现了Callable的call（）方法，并返回V结果

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);
    }
    
    /**
     * 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);
    }
 protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

futureTask.get(20, TimeUnit.SECONDS)，通过get（）方法获取结果，并设定阻塞时间