java.util.concurrent包学习笔记（一）Executor框架

 

类图：

---

 

其实从类图咱们能发现concurrent包（除去java.util.concurrent.atomic 和 java.util.concurrent.locks）中的内容并无特别多，大概分为四类：BlockingQueue阻塞队列体系、Executor线程组执行框架、Future线程返回值体系、其余各类单独的并发工具等。

首先学习的是Executor体系，是咱们处理多线程最常接触的内容。首先咱们单独看下继承体系：

Executor是顶级接口，里面只有一个方法：

---

 

public interface Executor {
    void execute(Runnable command);－－－执行一个Runnable对象，Runnable在前面的文章里面已经讲到，是线程的顶级接口，里面有个run方法
}

ExecutorService是咱们常常用到的多线程执行框架的声明，源码也不多，咱们逐个方法进行解释：

public interface ExecutorService extends Executor {
    void shutdown();－－－－关闭线程执行框架中的线程，但效果是再也不接受新线程加入，而且等待线程执行结束后关闭Executor

    List<Runnable> shutdownNow();－－－大致同上的，可是该命令会尝试关闭正在运行中的线程，可是也仅仅是调用terminate方法而后让jdk去决定是否结束，同时该方法返回那些awaiting状态的线程组

    boolean isShutdown();－－－是否已经被关闭的状态

    boolean isTerminated();－－－是否已经被停止的状态，在shutdown和shutdownnow被调用后，而且线程所有执行结束，该状态才是true，不然都是false

    boolean awaitTermination(long timeout, TimeUnit unit) －－－若是线程组terminate了，返回true,超时时间到了返回false。
        throws InterruptedException;

    <T> Future<T> submit(Callable<T> task);－－提交一个Callable的回调，而后执行完成后将结果放入Future对象。

    <T> Future<T> submit(Runnable task, T result);－－提交一个Runnable接口实现，而后result是Future返回值

    Future<?> submit(Runnable task);－－提交一个Runnable接口实现，若是执行完成，future.get()能够返回一个null

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)－－执行提交的Callable集合，而后返回各自的执行结果的Future对象列表，每一个元素isDone都是true
        throws InterruptedException;

    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)－－执行提交的task集合，执行完成或者timeout以后返回结果，isDone为true
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)－－执行提交的task集合，有一个执行完成就返回结果
        throws InterruptedException, ExecutionException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)－－同上，有一个执行完成或者有timeout出现
        throws InterruptedException, ExecutionException, TimeoutException;

　AbstractExecutorService实现了ExecutorService接口，而且声明为抽象类，然而其中一个抽象方法都没有。。.

---

 

public abstract class AbstractExecutorService implements ExecutorService {

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {---为子类定义了一个建立RunnableFuture对象的快捷方法
    return new FutureTask<T>(runnable, value);
}
－－－其他方法都是接口的实现方法

}

　还有一个继承了ExecutorService的接口的接口：

---

 

public interface ScheduledExecutorService extends ExecutorService { －－－定义延迟执行的动做
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit); －－－延迟delay个时间单位后开始执行

    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);－－－延迟delay个时间单位后执行并返回Future对象

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);在initialDelay、initialDelay＋N*period分别出发
　　public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　long initialDelay,
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　 long delay,
 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　TimeUnit unit); }－－initialDelay开始进行，后面每次执行成功后的dealy时间单位开始

 

---

接下来是重磅的内容ThreadPoolExecutor 也就是Executor框架的核心逻辑所在，ThreadPoolExecutor类算上注释有2100多行，但里面有不少一部分是方法的详细说明，还有大量的变量和private/protected的方法，真正开放出来的public方法不多，咱们只要逐个分析这些public方法就能够了。

咱们首先来看构造函数，构造函数有多个重载方法，目前只看参数最全的状况：

public ThreadPoolExecutor(int corePoolSize, －－－－池子里面的线程数量，即使idle状态的线程也会保留这个数量
                              int maximumPoolSize,－－－－池子里面能盛放最大线程数量
                              long keepAliveTime,－－－－当线程数量大于core核心数量的时候，而且里有idle状态的线程，那么最大能够被terminate的的等待时间
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　（就是在cpu借的线程资源若是闲置多久就必须还回去）
                              TimeUnit unit,－－timeUnit
                              BlockingQueue<Runnable> workQueue,－－－任务的队列
                              ThreadFactory threadFactory,－－－线程工厂
                              RejectedExecutionHandler handler) {－－－线程池处理不过来任务队列的时候的默认处理方法
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

public void execute(Runnable command) {}---执行一个任务

public void shutdown() {－－－使用可重入锁实现，关闭线程池执行框架，再也不接受新的任务，关闭idle状态的线程，等待正在执行的执行完成。
    final ReentrantLock mainLock = this.mainLock;
}

public List<Runnable> shutdownNow() {}--- 同ExecutorService中的接口说明，与shuwdown的区别在于会尝试关闭正在执行的线程

public boolean isShutdown() {}－－－是否已经关闭（没有running状态的线程了）

－－－－－－－一堆get和set方法

public int getActiveCount() {}－－－得到活动的线程数量

public long getCompletedTaskCount() {}－－－或者完成的任务数量

 其实咱们只要知道ThreadPoolExecutor的构造参数就能够明白它的工做方式，并且咱们须要作的也是把这些内容构造好。

咱们最经常使用的Executors里面的更加方便的Pool的类型其实都是为咱们加了一些default参数的ThreadPoolExecutor：

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}

也就是其实最后都是会到ThreadPoolExecutor上面去实现，区别就是一个用着方便，一个你能够控制的粒度更小，进而可能效率更高。

 

 咱们在类图的继承体系里面还能发现一个叫作ForkJoinPool的类，这个类实际上是不少人口中所谓将来java并发方向的类，因为涉及的内容较多，后面单独随笔进行学习和理解吧。