ThreadPoolExecutor线程池原理解读

        当我看到”你知道线程池的增长策略和拒绝策略吗“这个问题的时候好像似曾相识但又很模糊，跟着阅读了下源码和看了网上很多关于ThreadPoolExecutor线程池的文章，我在这里给自己做个记录，总结下自己的收获和自己理解。

写一个简单的例子

         先看一段代码和注释，基本上描述了一个线程池基本的参数

   /**
     * 线程池初始化方法
     * corePoolSize核心线程池大小----1
     * maximumPoolSize最大线程池大小----3
     * keepAliveTime线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit
     * TimeUnitKeepAliveTime时间单位----TimeUnit.MINUTES
     * workQueue阻塞队列----newArrayBlockingQueue<Runnable>(5)====5容量的阻塞队列
     * threadFactory新建线程工厂----newCustomThreadFactory()====定制的线程工厂
     * rejectedExecutionHandler当提交任务数超过maxmumPoolSize+workQueue之和时,
     * 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),
     * 任务会交给RejectedExecutionHandler来处理
     */
    public void init(int corePoolSize, int maximumPollSize, int capacity) {
        pool = new ThreadPoolExecutor(
                corePoolSize,
                maximumPollSize,
                30,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(capacity),
                new CustomThreadFactory(),
                new CustomRejectedExecutionHandler());

    }

         了解了每个参数的意义，那我还有一个问题，什么场景会受到最大线程池大小限制，什么情况会进阻塞队列呢？

          经过一番折腾，终于弄明白了这个问题，只要线程的数量大于核心线程数，并且为超过最大线程池大小，阻塞队列未满就放入队列中，最大线程池大小相当于一个阀门，一个能否进入阻塞队列的阀门，而核心线程池大小是控制当前就处理还是放入队列之后慢慢处理，如果都不满足条件，那拒绝策略就起到作用了，通过我们的配置去做相应的处理。

那什么是线程池的增长策略呢？

           后来我猜线程池的增长策略就是它的排队队列吧，我只是猜想，不喜勿喷哦。那就说说它的排队策略吧。通过看源码可以看到初始化方法的阻塞队列参数，BlockingQueue<Runnable> workQueue，workQueue是一个BlockingQueue接口的对象，仅用于存放Runnable对象。我们可以看到BlockingQueue的实现类：

（1）直接提交（如SynchronousQueue）
         直接提交策略表示线程池不对任务进行缓存。新进任务直接提交给线程池，当线程池中没有空闲线程时，创建一个新的线程处理此任务。这种策略需要线程池具有无限增长的可能性。
          Executors.newCachedThreadPool()使用SynchronousQueue创建线程池。
（2）*队列（如不具有预定义容量的LinkedBlockingQueue）
          LinkedBlockingQueue将导致当所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用*队列。
          Executors.newFixedThreadPool(3)使用LinkedBlockingQueue创建线程池。
          Executors.newSingleThreadExecutor()使用LinkedBlockingQueue创建线程池。
（3）有界队列（如ArrayBlockingQueue）
          有界队列（如ArrayBlockingQueue）有助于防止资源耗尽当最大线程数有限时，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷。
（4）优先级队列（如PriorityBlockingQueue）
（5）DelayedWorkQueue
          DelayedWorkQueue是ScheduledThreadPoolExecutor的静态内部类。
          Executors.newScheduledThreadPool(3)使用DelayedWorkQueue创建线程池。

线程池的拒绝策略

         线程池的拒绝策略主要是核心线程池处理不过来，又超过了阻塞队列的大小，并且线程池中的数量登录最大线程池大小，那么久通过handler指定的策略来处理此任务。RejectedExecutionHandler接口有四个实现类，分别是四种拒绝策略。

        （1）AbortPolicy：直接抛出异常，丢弃任务。

/**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }

        （2） CallerRunsPolicy:直接调用execute方法，除非这个executor被shut down，这个任务才被丢弃。这个策略不想放弃任何任务，只有在线程池被杀掉时才不得不放弃。

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }

           （3）DiscardPolicy：不能执行的任务将被删除

/**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }

           （4）DiscardOldestPolicy：这种策略通过翻译丢弃之前未被处理的请求然后重试，除非executor被中断，不会抛出异常。在pool没有关闭的前提下，首先丢掉缓存队列中的最早的任务，然后重新尝试该任务。

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

看看源码

         上面已经了解了一个线程池的初始化方法以及各个参数和一些策略问题。看看源码也许会更理解一些。

1. 核心方法execute，该方法用于向线程池中添加一个任务。

    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.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        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);
    }

         通过翻译，excute方法分为3步，（1）如果当前运行中的线程数小于corePoolSize，则试着开始一个新线程，执行addWorker方法。（2）判断线程池是否在运行，如果在，任务队列是否允许插入，插入成功再次验证线程池是否运行，如果不在运行，移除插入的任务，然后抛出拒绝策略。如果在运行，没有线程了，就启用一个线程。（3）如果添加非核心线程失败，就直接拒绝了。

2. 添加一个工作线程addWorker

        主要是创建新的线程并执行任务

• 第一个红框：做是否能够添加工作线程条件过滤：

  • 判断线程池的状态，如果线程池的状态值大于或等SHUTDOWN，则不处理提交的任务，直接返回；

• 第二个红框：做自旋，更新创建线程数量：

  • 通过参数core判断当前需要创建的线程是否为核心线程，如果core为true，且当前线程数小于corePoolSize，则跳出循环，开始创建新的线程

• 第一个红框：获取线程池主锁。

  • 线程池的工作线程通过Woker类实现，通过ReentrantLock锁保证线程安全。

• 第二个红框：添加线程到workers中（线程池中）。

• 第三个红框：启动新建的线程。

3. 线程池的底层数据存储结构是HashSet

4. worker线程处理队列任务

• 第一个红框：是否是第一次执行任务，或者从队列中可以获取到任务。

• 第二个红框：获取到任务后，执行任务开始前操作钩子。

• 第三个红框：执行任务。

• 第四个红框：执行任务后钩子。

这两个钩子（beforeExecute，afterExecute）允许我们自己继承线程池，做任务执行前后处理。

        源码的分析过程参考的博客来看的，特此说明https://mp.weixin.qq.com/s/-89-CcDnSLBYy3THmcLEdQ

总结

        了解了线程池的增长和拒绝策略，我们可以根据不同的情况去实现自己想要的线程池。 

         线程池的源码，存储任务的底层结构，源码已经给我们提供了入口，在创建任务之前通过钩子实现自己想要的逻辑。

         线程池目的是利用更高效友好的方式实现我们的业务