Java 线程池

前言：

    在Java中，使用线程来执行异步任务；Java线程的创建和销毁需要消耗一定的系统计算开销（在HotSpot VM的线程模型中，Java线程（java.lang.Thread）被一对一映射为本地操作系统线程，Java线程启动时会创建一个本地操作系统线程，当该Java线程终止时，这个操作系统线程也会被回收，操作系统会调度所有线程并将他们分配给可用的CPU），为了减少频繁地创建和销毁线程带来额外的系统开销以及提高系统处理任务的性能，由此引入线程池的概念。

为什么要使用线程池：

<1>降低资源消耗：通过重复利用已创建的线程降低线程的创建和销毁带来的系统开销；
<2>提高相应速度：当任务到达时，可以立即执行；
<3>提高线程的可管理性：使用线程池进行线程的统一分配，调优和监控。

线程池实现原理：

    当向线程池提交一个任务后，线程池执行流程如下：
1）：线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务，否则，进入第二步；
2）：线程池判断工作队列是否已经满。如果工作队列没有满，则将新提交的任务存储在工作队列中；否则，进入第三步；
3）：线程池判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务；否则，则交给饱和策略来处理任务。

image.png

线程池总体框架（Executor框架）：

    Java里面线程池的顶级接口是Executor，但是严格意义上讲Executor并不是一个线程池，而只是一个执行线程的工具。真正的线程池接口是ExecutorService。

【1】Executor框架主要由3大部分组成：

<1>任务。包括被执行任务需要实现的接口（Runnable接口和Callable接口）；
<2>任务的执行。包括任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口，Executor框架有两个关键类实现了ExecutorSercice接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor）。
<3>异步计算的结果。包括接口Future和实现Future接口的FutureTask类。

【2】Executor框架的类和接口

image.png

其中主要包括：
<1>Executor接口：是Executor框架的基础，它将任务的提交和执行分离开来；
<2>ThreadPoolExecutor：线程池的核心实现类，用来执行被提交的任务；
<3>ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。
<4>Future接口和Future接口实现类FutureTask，代表异步计算的结果；
<5>Runnable接口和Callbale接口的实现类，都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。

【3】Executor框架使用示意图

image.png

具体步骤如下：
1）主线程首先要创建实现Runnable或者Callable接口的任务对象。
2）然后把Runnable对象直接交给ExecutorService执行（ExecutorService.execute(Runnable command)），或者把Runnable对象或Callable对象提交给ExecutorService执行（ExecutorService.submit(Runnable task)或者ExecutorService.submit(Callable<T> task)）.
3)如果执行ExecutorService.submit(...)，ExecutorService将返回一个实现了Future接口的对象（FutureTask）。
4）最后，主线程可以执行FutureTask.get()方法来等待任务执行完成，也可以执行FutureTask.cancel(boolean mayInterruptIfRunning)方法来取消此任务的执行。

线程池的核心实现类：

要配置一个线程池是比较复杂的，尤其是对于线程池的原理不是很清楚的情况下，很有可能配置的线程池不是较优的，因此在Executors类里面提供了一些静态工厂，用于创建一些常用的线程池。

【1】ThreadPoolExecutor

【2】ScheduledThreadPoolExecutor

    这两个核心实现类通常均由静态工厂类Executors来创建。其中，Executors可以创建3种类型的ThreadPoolExecutor，分别为SingleThreadExecutor、FixedThreadPool、CachedThreadPool；Executors可以创建两种类型的ScheduledThreadPoolExecutor，分别为：ScheduledThreadPoolExecutor、SingleThreadScheduledExecutor。

    ThreadPoolExecutor执行executor()方法的示意图，如下图所示：

image.png

    图中1,2,3,4分别表示如下：
1）如果当前运行的线程数少于corePoolSize，则创建新线程来执行任务（执行这一操作需要获取全局锁）；
2）如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
3）如果队列已满，即无法将任务加入BlockingQueue中，则创建新的线程来处理任务（同理，需要获取全局锁）；
4）如果创建新的线程将使运行的线程超过maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

源码分析：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.
         *The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        //  如果运行的线程数少于corePoolSize，则尝试使用给定的命令创建一个新的线程
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 如果任务可以成功排队，那么我们仍然需要仔细检查是否应该添加一个线程（因为自上次检查后现有的线程已经死亡），或者自任务进入此方法后线程池就关闭了， 所以我们需要重新检查状态，如果有必要，回滚入队的ifstopped，或者如果没有，则启动一个新的线程。
        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);
        }
        // 如果我们无法将任务入队列，那么我们尝试添加一个新的线程。 如果失败，我们知道我们已经关闭或使用饱和策略来拒绝任务。
        else if (!addWorker(command, false))
            reject(command);
    }

线程池的创建：

    Executor框架最核心的类便是ThreadPoolExecutor，它是线程池的实现类，创建线程池时主要包含以下组件：
【1】corePoolSize：核心线程池大小。
【2】maximumPoolSize：最大线程池的大小。
【3】BlockingQueue：暂时保存任务的工作队列，主要由以下几种：
        1）ArrayBlockingQueue：一个基于数组结构的有界阻塞队列，按FIFO原则对元素进行排序；
        2）LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
        3）：SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等待另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常高于LinkedBlockingQueue,静态工厂方法Executors.newCachedThreadPool使用了这个队列。
        4）PriorityBlockingQueue：一个具有优先级的无线阻塞队列。

【4】RejectedExecutionHandler（饱和策略）：当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时（达到最大线程池的大小且工作队列已满），execute()方法将要调用的Handler，默认情况下AbortPolicy，表示无法处理新任务时抛出异常。饱和策略主要有4种策略：
        1）AbortPolicy：直接抛出异常；
        2）：CallerRunsPolicy：只用调用者所在线程来运行任务；
        3）：DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务；
        4）：DiscardPolicy：不处理，丢弃掉。
【5】ThreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个线程设置更有意义的名字，如使用开源框架guava提供的ThreadFactoryBuilder可以快速给线程池里的线程设置更有意义的名字：new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build()；
【6】keepAliveTime：线程池的工作线程空闲后，保持存活的时间。
【7】TimeUnit：线程活动保持时间的单位。

向线程池的提交任务：

    通过execute()或者submit()方法向线程池提交任务。
        <1>：execute()方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行完成。
        <2>：submit()方法用于提交需要返回值的任务。线程池会返回一个furture类型的对象，通过这个future对象可以判断任务是否执行完成，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get(long time, TimeUnit unit)方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

关闭线程池：

    通过shutdown()或shutdownNow()方法来关闭线程池。原理：遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。但shutdownNow首先将线程池的状态设置成STOP,然后尝试停止所有的正在执行或者暂停任务的线程，并返回等待执行任务的列表，而shutdown只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。通常调用shutdown方法来关闭线程，如果任务不一定要执行完，则可以调用shutdownNow方法。