Java异步 call future 解析Java异步之call future

public interface Callable<V> {
    V call() throws Exception;
}


public interface Runnable {
    public abstract void run();
}

public interface Executor {
    void execute(Runnable command);
}

public interface ExecutorService extends Executor {

    void shutdown();                // 现有任务完成后停止线程池
 
    List<Runnable> shutdownNow();   // 立即停止线程池

    boolean isShutdown();           // 判断是否已停止

    boolean isTerminated();

    <T> Future<T> submit(Callable<T> task);        // 提交Callale任务

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);

    // 针对Callable集合的invokeAll()等方法
}

// 1.第一个重载方法，参数为Callable
public <T> Future<T> submit(Callable<T> task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<T> ftask = newTaskFor(task);
  // return new FutureTask<T>(callable);
  execute(ftask);
  return ftask;
}

// 2.第二个重载方法，参数为Runnable
public Future<?> submit(Runnable task) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<Void> ftask = newTaskFor(task, null);
  // return new FutureTask<T>(task, null);
  execute(ftask);
  return ftask;
}

// 3.第三个重载方法，参数为Runnable + 返回对象
public <T> Future<T> submit(Runnable task, T result) {
  if (task == null) throw new NullPointerException();
  RunnableFuture<T> ftask = newTaskFor(task, result);
  // return new FutureTask<T>(task, result);
  execute(ftask);
  return ftask;
}

public FutureTask(Callable<V> callable) {
  this.callable = callable;
  this.state = NEW;
}

public FutureTask(Runnable runnable, V result) {
  this.callable = Executors.callable(runnable, result); // 通过适配器将runnable在call()中执行并返回result
  this.state = NEW;
}

static final class RunnableAdapter<T> implements Callable<T> {
  final Runnable task;
  final T result;                // 返回的结果；显然：需要在run()中赋值

  RunnableAdapter(Runnable task, T result) {
    this.task = task;
    this.result = result;
  }
  public T call() {
    task.run();
    return result;
  }
}

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation unless it has been cancelled.
     */
    void run();
}

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;

public class FutureTask<V> implements RunnableFuture<V> {
    
    private Callable<V> callable;         // target，待执行的任务
    
    /** 保存执行结果或异常，在get()方法中返回/抛出 */
    private Object outcome; // 非volatile，通过CAS保证线程安全
    
    
    public void run() {
        ......
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();            // 调用call()执行用户任务并获取结果
                ran = true;                   // 执行完成，ran置为true
            } catch (Throwable ex) {          // 调用call()出现异常，而run()方法继续执行
                 result = null;
                 ran = false;
                 setException(ex);            
                 // setException(Throwable t): compareAndSwapInt(NEW, COMPLETING);  outcome = t;      
            }
            if (ran)
                set(result);                  
            	// set(V v): compareAndSwapInt(NEW, COMPLETING);  outcome = v;
        }
    }
    
    
    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);         // 加入队列等待COMPLETING完成，可响应超时、中断
        return report(s);
    }

    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        // 超时等待
    }
    
    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)                              // 将outcome作为执行结果返回
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);   // 将outcome作为捕获的返回
    }
}

class MyTask implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        System.out.println("子线程开始计算...");
        for (int i=0;i<3;++i){
            Thread.sleep(1000);
            System.out.println("子线程计算中，用时 "+(i+1)+" 秒");
        }
        System.out.println("子线程计算完成，返回：100");
        return 100;
    }
}

ExecutorService executor = Executors.newCachedThreadPool();
MyTask task = new MyTask();

System.out.println("主线程启动线程池");
Future<Integer> future = executor.submit(task);
System.out.println("主线程得到返回结果："+future.get());
executor.shutdown();

System.out.println("主线程启动线程池");
FutureTask<Integer> futureTask = new FutureTask<>(task);
executor.submit(futureTask);                                 // 作为Ruunable传入submit()中
System.out.println("主线程得到返回结果："+futureTask.get());    // 作为Future获取结果
executor.shutdown();