线程池相关的类学习

Executor

public interface Executor {
   
	//执行任务
    void execute(Runnable command);
}

ExecutorService

public interface ExecutorService extends Executor {
   

   //关闭线程池，不能再向线程池中提交任务，已存在与线程池中的任务会继续执行，直到完成
    void shutdown();

   //立刻关闭线程池，不能再向线程池中提交任务，已存在与线程池中的任务会被终止执行
    List<Runnable> shutdownNow();

	//判断线程池是否已关闭
    boolean isShutdown();

	//判断线程池是否已终止，只有调用了shutdown()或shutdownNow()之后该方法才会返回true
    boolean isTerminated();

	//等待线程池中所有任务都执行完成，并设置超时时间
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

	//向线程池中提交一个Callable类型的任务，并返回一个Future类型的结果
    <T> Future<T> submit(Callable<T> task);

	//向线程池中提交一个Runnable类型的任务，并且给定一个T类型的收集结果集的参数，并返回一个Future类型的结果
    <T> Future<T> submit(Runnable task, T result);

	//向线程池中提交一个Runnable类型的任务，并返回一个Future类型的结果
    Future<?> submit(Runnable task);

	//执行全部提交Callable类型的tasks任务集合，并返回一个Future类型的结果集集合
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

    //执行全部提交Callable类型的tasks任务集合，并且设置超时时间，并返回一个Future类型的结果集集合
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

     //执行提交Callable类型的tasks任务集合，并返回一个已经执行成功的任务结果
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

   //执行提交Callable类型的tasks任务集合，并设置超时时间，并返回一个已经执行成功的任务结果
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

AbstractExecutorService

public abstract class AbstractExecutorService implements ExecutorService {
   

	//将Runnable类型的任务包装成FutureTask
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
   
        return new FutureTask<T>(runnable, value);
    }

   //将Callable类型的任务包装成FutureTask
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
   
        return new FutureTask<T>(callable);
    }

   //向线程池中提交一个Runnable类型的任务，并把该任务包装成RunnableFuture类型，并执行该任务，并且返回一个Future类型的结果
    public Future<?> submit(Runnable task) {
   
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

   //向线程池中提交一个Runnable类型的任务，并设定一个T类型的参数用于包装返回值结果，并把该任务包装成RunnableFuture类型，并执行该任务，并且返回一个Future类型的结果
    public <T> Future<T> submit(Runnable task, T result) {
   
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }

     //向线程池中提交一个Callable类型的任务，并把该任务包装成RunnableFuture类型，并执行该任务，并且返回一个Future类型的结果
    public <T> Future<T> submit(Callable<T> task) {
   
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

    //向线程池中提交一个tasks任务集合，并设置是否超时及超时时间，并得到一个已经执行完毕任务的结果
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                              boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
   
        //集合是null或空抛出异常
        if (tasks == null)
            throw new NullPointerException();
        //拿到任务数量
        int ntasks = tasks.size();
        if (ntasks == 0)
            throw new IllegalArgumentException();
        //存放任务执行结果
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
        //用于执行提交的任务
        ExecutorCompletionService<T> ecs =
            new ExecutorCompletionService<T>(this);

        try {
   
			//可能抛出的异常
            ExecutionException ee = null;
            //超时时间
            final long deadline = timed ? System.nanoTime() + nanos : 0L;
            //获取一个任务
            Iterator<? extends Callable<T>> it = tasks.iterator();
			//再循环之前先提交指定一个任务，保证循环之前任务已经开始执行
            futures.add(ecs.submit(it.next()));
            --ntasks;//任务数量减一
            int active = 1;//记录正在执行任务的数量

            for (;;) {
   
            	//从完成任务的BlockingQueue队列中获取并移除下一个将要完成的任务的结果。 poll()为非阻塞方法
                Future<T> f = ecs.poll();
                if (f == null) {
   
                	//还有未完成的任务
                    if (ntasks > 0) {
   
                        --ntasks;
                        //继续执行任务
                        futures.add(ecs.submit(it.next()));
                        ++active;
                    }
                    else if (active == 0)
                    //如果没有正在执行的任务则跳出循环
                    //这里加这个判断是因为poll()方法是非阻塞的	可能active==0,但结果集还没有返回
                        break;
                    else if (timed) {
   
                    //超时，则设置超时时间
                        f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null)
                            throw new TimeoutException();
                        nanos = deadline - System.nanoTime();
                    }
                    else
                        f = ecs.take();
                }
                if (f != null) {
   
                    --active;
                    try {
   
                    	//只要有一个结果集不为空，则直接返回，不会继续向下执行
                        return f.get();
                    } catch (ExecutionException eex) {
   
                        ee = eex;
                    } catch (RuntimeException rex) {
   
                        ee = new ExecutionException(rex);
                    }
                }
            }

            if (ee == null)
                ee = new ExecutionException();
            throw ee;

        } finally {
   
        	//判断存在还未执行的任务，则直接取消
            for (int i = 0, size = futures.size(); i < size; i++)
                futures.get(i).cancel(true);
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
   
        try {
   
            return doInvokeAny(tasks, false, 0);
        } catch (TimeoutException cannotHappen) {
   
            assert false;
            return null;
        }
    }

    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
   
        return doInvokeAny(tasks, true, unit.toNanos(timeout));
    }

	//执行提交的所有任务，并返回结果集
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
   
        if (tasks == null)
            throw new NullPointerException();
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        //这个标识代表当前所有任务是否都已经执行完成
        //无论是正常执行，还是异常，都算是已完成
        boolean done = false;
        try {
   
        	//遍历所有任务执行
            for (Callable<T> t : tasks) {
   
                RunnableFuture<T> f = newTaskFor(t);
                futures.add(f);
                execute(f);
            }
            //遍历结果集
            for (int i = 0, size = futures.size(); i < size; i++) {
   
                Future<T> f = futures.get(i);
                //还没有执行完的任务，阻塞继续执行，直至返回结果
                if (!f.isDone()) {
   
                    try {
   
                        f.get();
                    } catch (CancellationException ignore) {
   
                    } catch (ExecutionException ignore) {
   
                    }
                }
            }
            //标志所有任务都已完成
            done = true;
            return futures;
        } finally {
   
        	//如果存在还没有完成的任务，则直接取消
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }
	//同上，只不过在执行任务时和获取结果时设置了超时时间
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
   
        if (tasks == null)
            throw new NullPointerException();
        long nanos = unit.toNanos(timeout);
        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
        boolean done = false;
        try {
   
            for (Callable<T> t : tasks)
                futures.add(newTaskFor(t));

            final long deadline = System.nanoTime() + nanos;
            final int size = futures.size();

            // Interleave time checks and calls to execute in case
            // executor doesn't have any/much parallelism.
            for (int i = 0; i < size; i++) {
   
                execute((Runnable)futures.get(i));
                nanos = deadline - System.nanoTime();
                if (nanos <= 0L)
                    return futures;
            }

            for (int i = 0; i < size; i++) {
   
                Future<T> f = futures.get(i);
                if (!f.isDone()) {
   
                    if (nanos <= 0L)
                        return futures;
                    try {
   
                        f.get(nanos, TimeUnit.NANOSECONDS);
                    } catch (CancellationException ignore) {
   
                    } catch (ExecutionException ignore) {
   
                    } catch (TimeoutException toe) {
   
                        return futures;
                    }
                    nanos = deadline - System.nanoTime();
                }
            }
            done = true;
            return futures;
        } finally {
   
            if (!done)
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
        }
    }

}

ScheduledExecutorService

public interface ScheduledExecutorService extends ExecutorService {
   

  	//延时delay时间来执行command任务，只执行一次
    public ScheduledFuture<?> schedule(Runnable command,long delay, TimeUnit unit);
	//延时delay时间来执行callable任务，只执行一次
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,long delay, TimeUnit unit);
	//延时initialDelay时间首次执行command任务，之后每隔period时间执行一次
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit);
  	//延时initialDelay时间首次执行command任务，之后每延时delay时间执行一次
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay,long delay,TimeUnit unit);

}