java基础-Runnable与Callable

下面这段测试代码，如果取消第一个future.get的注释，也就是说等第一次submit提交的任务执行完成后，再提交同一个任务，再次提交的任务并不会得到执行，因为此时callable的status已经不是NEW了。但如果将第一个future.get注释掉，也就是在第一次submit提交的任务完成前，再提交同一个任务，则再次提交的任务依旧会被执行，因为此时callable的status仍然是NEW。这点与使用execute提交Runnable任务有区别，Runnable本身是没有任何状态标识的，注意，Callable本身也没有状态，但是在jdk Executor框架中提交的Callable会被包装为java.util.concurrent.FutureTask。

    @Test
    public void testCallable() throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
//        int i = 0;
        final AtomicInteger i = new AtomicInteger(0);
        Callable<Object> callable = new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                Thread.sleep(1000);
                System.out.println(i.incrementAndGet());
                return null;
            }
        };
        Future<Object> future = executor.submit(callable);
//        future.get();
        executor.submit(callable);
        executor.shutdown();
//        executor.shutdownNow();
        executor.awaitTermination(20000, TimeUnit.MILLISECONDS);
    }

先来看Executor提交（execute）Runnable任务的主要逻辑

        int c = ctl.get();
        //如果worker（执行线程）数小于corePoolSize，则增加worker，并将command作为其第一个work（Runnable）
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //将command入队，并检查Executor状态，如果在运行但worker数为0（因为woker到达空闲时间就会被取消，就可能出现降为0的情况），则（如果worker数小于maximumPoolSize）增加woker
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //最后再尝试添加worker，如果不成功说明Executor被关闭了或者worker数达到最大了
        else if (!addWorker(command, false))
            reject(command);    

再来Executor提交（submit）Callable

    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        //newTaskFor就是将Callable包装成一个FutureTask（RunnableFuture的实现，Future代表了一个异步计算的结果，RunnableFuture就是Future继承了Runnable，表示可执行的Future，可见FutureTask可执行（真正的执行逻辑call被run包装）、可获得计算结果（通过wait、notify），并且还通过interrupt提供了可取消）
        RunnableFuture<T> ftask = newTaskFor(task);
        //execute就是提交Runnable的方法
        execute(ftask);
        return ftask;
    }

FutureTask类

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;

    /** The underlying callable; nulled out after running */
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
    private volatile Thread runner;
    /** Treiber stack of waiting threads，可能会有多个线程get同一个future，Callable执行完成后就需要notify所有的这些等待线程*/
    private volatile WaitNode waiters;

    ......

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

    ......
}

其实我们完全可以自己调用FutureTask

    @Test
    public void testFutureTask() throws ExecutionException, InterruptedException {
        FutureTask<Integer> futureTask = new FutureTask<>(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(1000);
                return 1;
            }
        });
        new Thread(futureTask).start();
        Integer result = futureTask.get();
        System.out.println(result);//1
    }

Executor的shutdown就是interrupt所有未被执行的work，shutdownNow则是interrupt所有work，awaitTermination并不对work进行操作，只是判断Executor是否已经terminated（shutdown了，并且没有work在执行了，就是terminated了）。