JAVA 多线程（一） ThreadPoolExecutor 原理分析

àì夳堔傛蜴生んèń 2022-06-08 09:27 222阅读 0赞

## 工作原理 ##

当一个新的任务被提交到ThreadPoolExecutor的execute()方法处理，如果当前池中正在运行的线程少于corePoolSize,就会创建一个新的线程来处理任务。  
如果线程池中的线程大于等于corePoolSize，但是小于最大容量maximumPoolSize时，队列未满，任务加入到队列中  
如果队列满了，正在运行的线程数等于maximumPoolSize，该任务会被拒绝

参数详细说明

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }

**corePoolSize和maximumPoolSize**  
如果二者相等，那么线程池的数量就是固定的。  
默认情况下，只有当一个新的任务到达时，才会创建和启动core threads，但是可以通过prestartCoreThread和prestartAllCoreThreads来改变。

**ThreadFactory**  
通过使用java.util.concurrent.ThreadFactory可以创建新的线程  
如果不额外指定ThreadFactory，默认使用Executors.defaultThreadFactory  
通过该默认的线程工厂，所有创建的线程都会被加入到同一个ThreadGroup中去，并且这些线程都会有相同的优先级（NORM\_PRIORITY），并且都是non-daemon线程

**keepAliveTime**  
如果pool当前拥有的线程超过了corePoolSize,超出的线程如果在大于keepAliveTime的时间外限制（idle）,这些线程就会被终止  
该机制在pool没有被活跃的使用的时候，可以减少资源浪费；  
默认情况下，keep-alive机制仅仅会在线程数超过corePoolSizeThreads时才会被使用；  
当然，通过使用ThreadPoolExecutor\#allowCoreThreadTimeOut(boolean)也可以将这种keep-alive机制应用在core threads上去（只要keepAliveTime>0即可）

**Queue**  
任何一种BlockingQueue都可以被用来传递和存储递交到线程池中的任务

有三种队列策略：  
1.synchronousQueue 默认：

*  直接把任务交给线程而不是入队，如果已经没有线程立即来处理提交到线程池的任务，会创建一个新的线程。
 *  这种策略需要maximumPoolSize无界来确保新提交的任务不会被拒绝
 *  缺点是：当任务到来的属于大于任务被处理的速度，线程数会疯涨

2.LinkedBlockingQueue 无界队列：

*  由于队列无界，当运行的线程等于corePoolSize时，新到来的任务会入队而不会创建新的线程来执行（即pool中的线程数永远不会大于corePoolSize）
 *  这种方式的缺点：当任务到来的速度大于任务被处理的速度时，队列长度会疯长。

3.ArrayBlockingQueue 有界队列：

*  这种方式是非常难处理好的一种方式，要考虑好ArrayBlockingQueue的大小和maximumPoolSize的大小；
 *  当ArrayBlockingQueue较大而maximumPoolSize较小时，会降低CPU使用率、减少OS资源、减少上下文切换，但是吞吐量会降低。–>线程较少的特点就是这样
 *  如果任务频繁的被阻塞（例如，they are I/O bound），就需要更多的线程了；
 *  当ArrayBlockingQueue较小而maximumPoolSize较大时，会使CPU使用繁忙但也会遇到一些不可接受的scheduling，吞吐量也会降低。

**回绝策略**

在回绝任务的时候，execute()方法会调用RejectedExecutionHandler.rejectedExecution，有四种拒绝策略：

*  ThreadPoolExecutor.CallerRunsPolicy:调用execute()的线程自己来处理该任务，绝大部分情况下是主线程。

由于主线程执行这个任务，那么新到来的任务不会被提交到线程池中执行，而是提交到TCP层的队列，TCP层队列满了，就开始拒绝，性能很低，直到主线程执行完这个任务

*  ThreadPoolExecutor.DiscardPolicy:不能被执行的任务直接被扔掉
 *  ThreadPoolExecutor.DiscardOldestPolicy:如果executor没有被关闭，队列头部的任务将会被丢弃，然后将该任务加到队尾
 *  ThreadPoolExecutor.AbortPolicy（默认）：回绝任务并抛出异常

**AOP应用**

ThreadPoolExecutor提供了两个方法在每个任务执行前后进行调用ThreadPoolExecutor.beforeExecute和ThreadPoolExecutor.afterExecute

**例子**

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /** * ThreadPoolExecutorTest * Created by heqianqian on 2017/9/21. */
    public class ThreadPoolExecutorTest { 
    
        private static ThreadPoolExecutor executor =
                new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
    
        /*执行任务*/
        public void executeTask() throws InterruptedException {
            Task1 task1 = new Task1();
            Task2 task2 = new Task2();
            executor.execute(task1);
            executor.execute(task2);
            executor.awaitTermination(2,TimeUnit.SECONDS);
            executor.shutdown();
        }
    
        class Task1 implements Runnable {
    
            @Override
            public void run() {
                for (int i = 0; i < 1000; i++) {
                    System.out.println("hello!");
                }
    
            }
        }
    
        class Task2 implements Runnable {
    
            @Override
            public void run() {
                for (int i = 0; i < 1000; i++) {
                    System.out.println("World!");
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            ThreadPoolExecutorTest threadPoolExecutorTest = new ThreadPoolExecutorTest();
            threadPoolExecutorTest.executeTask();
        }
    
    }

初始化的参数：

corePoolSize==5    
    maximumPoolSize==10  
    keepAliveTime==30s 
    队列：ArrayBlockingQueue，大小为10
    线程工厂：defaultThreadFactory（默认）
    回绝策略：AbortPolicy（默认）

*  当提交的任务数小于corePoolSize也就是5时，executor会使用5个核心线程去执行这些任务
 *  这是又提交一个新任务，如果5个核心线程有空闲线程的话，就会使用空闲的线程去处理，如果都不空闲，该任务进队列。
 *  之后再来任务，还是像第二步那样去执行，直到任务将队列放满了，这时候，如果再来一个任务，如果5个核心线程有空闲线程，直接去执行该任务，如果5个核心线程都在忙，这时候就创建一个新的线程来执行该任务
 *  如果通过上边的流程，最后5个线程都在忙，并且队列满了，并且pool中的线程数已经是10个了（池中的线程总数==maximumPoolSize了），这时候就要执行回绝策略了，在这里，使用了默认的AbortPolicy，即直接放弃该任务并抛出异常

在代码的执行过程中，如果发现后来创建的5个线程有超过30秒都没被调用过的，该线程就被回收掉了。

**线程池的生命周期**

*  创建之初，状态是RUNNING
 *  调用了ExecutorService.shutdown，把之前提交的任务进行处理，但是不接受新的任务（使用回绝策略回绝任务），状态是SHUTDOWN
 *  如果调用的是ExecutorService.shutdownNode的话，会取消所有运行中的任务包括队列中的，并且不再接受新任务，状态是STOP/TERMINATED

--------------------

当队列满了之后，这时候来了一个任务，恰好5个核心线程有一个空闲了，那么下面两种情况哪一个正确：

1）这个空闲的核心线程直接执行刚刚到来的任务

2）这个空闲的核心线程直接执行队列头部的任务，而将刚刚到来的任务放入队尾

解答：这个问题的答案就一句话，有空闲核心线程，就是用核心线程去执行任务；没有空闲的核心线程，任务才会入队。所以选1）