线程池相关的类学习

Executor

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

ExecutorService

public interface ExecutorService extends Executor {//关闭线程池，不能再向线程池中提交任务，已存在与线程池中的任务会继续执行，直到完成void shutdown();//立刻关闭线程池，不能再向线程池中提交任务，已存在与线程池中的任务会被终止执行List<Runnable> shutdownNow();//判断线程池是否已关闭boolean isShutdown();//判断线程池是否已终止，只有调用了shutdown()或shutdownNow()之后该方法才会返回trueboolean 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类型的任务包装成FutureTaskprotected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {return new FutureTask<T>(runnable, value);}//将Callable类型的任务包装成FutureTaskprotected <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();}elsef = 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);}