死磕ThreadPoolExecutor线程池

Rothschil 2020-05-14 15:53:00

Java,ThreadPoolExecutor

1. 线程池的优势

节省资源开销：重复利用线程池中的线程，不需要每次都创建
提升对线程的管理能力：统一对线程分配和监控，避免无限创建，造成资源内存溢出和CPU耗尽
提高响应，降低系统开销：减少了创建线程的时间消耗，提高应对任务的响应

线程空间大小

线程空间大小和具体JDK版本有很大关系，JDK8将近1.9M、JDK11差不多1.5M多。具体大小的查看可以执行命令java -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -version。

C:\Users\WONGS> java -XX:+UnlockDiagnosticVMOptions -XX:NativeMemoryTracking=summary -XX:+PrintNMTStatistics -version
java version "11.0.2" 2019-01-15 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.2+9-LTS)
Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.2+9-LTS, mixed mode)

Native Memory Tracking:

Total: reserved=7849030KB, committed=465994KB
-                 Java Heap (reserved=6248448KB, committed=391168KB)
                            (mmap: reserved=6248448KB, committed=391168KB)

-                     Class (reserved=1056866KB, committed=4578KB)
                            (classes #472)
                            (  instance classes #407, array classes #65)
                            (malloc=98KB #502)
                            (mmap: reserved=1056768KB, committed=4480KB)
                            (  Metadata:   )
                            (    reserved=8192KB, committed=4096KB)
                            (    used=3120KB)
                            (    free=976KB)
                            (    waste=0KB =0.00%)
                            (  Class space:)
                            (    reserved=1048576KB, committed=384KB)
                            (    used=297KB)
                            (    free=87KB)
                            (    waste=0KB =0.00%)

-                    Thread (reserved=16455KB, committed=591KB)
                            (thread #16)
                            (stack: reserved=16384KB, committed=520KB)
                            (malloc=52KB #89)
                            (arena=19KB #30)
......

2. 几种常见线程池

newCachedThreadPool：数量无上限，该线程池会根据需要创建，但是优先使用之前构造的线程。这些池通常将提高执行许多短期异步任务的程序的性能，如果没有可用的现有线程，则将创建一个新线程并将其添加到池中。当60S内未使用的线程将被终止并从缓存中删除。因此，保持空闲时间足够长的池不会消耗任何资源。

构造函数：

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

newFixedThreadPool：数量固定大小，该线程池重用在共享的无边界队列上运行的固定数量的线程。在任何时候，最多nThreads（构造函数的参数，核心线程数与最大线程数相等）个线程都是活动的处理任务。如果在所有线程都处于活动状态时提交了其他任务，则它们将在队列中等待，直到某个线程可用为止。如果在关闭之前执行过程中由于执行失败导致任何线程终止，则在执行后续任务时将使用新线程代替。池中的线程将一直存在，直到明确将其关闭。

构造函数：

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

newSingleThreadExecutor：单线程，保证提交线程执行任务的FIFO，（但是请注意，如果该单个线程由于在关闭前执行期间由于执行失败而终止，则在需要执行新任务时将使用新线程代替。），在任何给定时间活动的任务不超过一个。与newFixedThreadPool不同，保证返回的执行程序不可重新配置为使用其他线程。

构造函数：

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

newScheduledThreadPool：该线程池可以安排命令在给定的延迟后运行或定期执行。

构造函数：

1
2
3

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

ScheduledThreadPoolExecutor类结构图

通过类图，我们分析，其实 ScheduledThreadPoolExecutor 是 ThreadPoolExecutor 子类。

newWorkStealingPool：使用所有可用处理器作为目标并行度级别，创建工作窃取线程池,在实际中很少使用，不细讲。

综上所述，我们可以看到这些线程池底层实现都依靠 ThreadPoolExecutor 类的构造器，它是构造线程池的核心实现。但是现实在开发过程中避免利用 Executors 去创建线程池，这容易让人疑惑，JDK命名自带实现，为什么避免用，看完下一章节后，我们再谈这个话题。

3. 解析ThreadPoolExecutor

corePoolSize：初始化大小，即使没有空闲，保留在池中的线程数，除非设置了allowCoreThreadTimeOut
maximumPoolSize：允许线程池同时并行的线程数量
keepAliveTime：当线程数大于内核数时，这是多余的空闲线程将在终止之前等待新任务的最长时间
unit：TimeUnit类型，这没什么好说
workQueue：在执行任务之前用于保留任务的队列，此队列将仅保存execute方法提交的Runnable任务。
threadFactory：执行程序创建新线程时要使用的工厂
handler：当线等待队列中的数量超过既定容量，所需要处理策略

构造函数：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          @NotNull TimeUnit unit,
                          @NotNull BlockingQueue<Runnable> workQueue,
                          @NotNull ThreadFactory threadFactory,
                          @NotNull RejectedExecutionHandler handler)

ThreadPoolExecutor结构类图

3.1. corePoolSize、maximumPoolSize

corePoolSize、maximumPoolSize 线程池中初始化的线程数量，初始化太多或者太少，都有可能造成资源的浪费，具体实际情况根据所需要处理的任务特征决定。

3.2. workQueue

将待处理的任务放入一个队列，这是一个阻塞队列，该队列可以是有界也可以是无界。

3.3. handler

AbortPolicy：拒绝执行处理程序，这是默认策略。
CallerRunsPolicy：线程池未关闭，被拒绝任务，它直接在调用线程中运行被丢弃的任务
DiscardOldestPolicy：丢弃最老，然后重试执行当前任务
DiscardPolicy：比较粗暴，直接丢弃。

其实上述四种策略都不够友好，在实际应用场景中，肯定要记录日志或者通过RPC框架触发通知补偿措施，否则会造成数据丢失或者处理过程不够严谨。一般情况下，我们需要自己实现 RejectedExecutionHandler 接口，在接口中记录日志或者持久化不能处理的任务信息。再通过定时任务，进行补偿重试。

3.4. 线程池执行顺序

线程池执行逻辑

判断核心线程数（corePoolSize）是否已满，未满则创建核心线程，用来执行任务
判断 workQueue 队列类型是否是有界队列，如果否，则maximumPoolSize 参数的配置无效；当是有界队列，则判断有界队列是否已满
判断有界队列是否已满，当队列还有空间，还要进一步判定线程池是否已满即线程池中线程数量是否已达到maximumPoolSize，未超过则直接创建非核心线程；超过则根据拒绝策略执行相关操作

下面将举几个例子。

3.4.1. 无界队列样例

定义核心线程数，corePoolSize 为 1；无界队列LinkedBlockingQueue同时为展示更好的效果，我让每个线程执行后都sleep 秒钟。

public class ThreadPoolExecutorDemo2 {

    public static ExecutorService getExecutorService(int coreSize,int maxSize,BlockingQueue queue){
        RejectedExecutionHandler policy = new ThreadPoolExecutor.AbortPolicy();
        return new ThreadPoolExecutor(coreSize, maxSize,0, TimeUnit.SECONDS, queue, policy);
    }

    public static void main(String[] args) {
//        ArrayBlockingQueue queue = new ArrayBlockingQueue(3);
        LinkedBlockingQueue queue = new LinkedBlockingQueue();
        ExecutorService es = getExecutorService(1,5,queue);

        ThreadTaskDemo t1 = new ThreadTaskDemo("t1");
        ThreadTaskDemo t2 = new ThreadTaskDemo("t2");
        ThreadTaskDemo t3 = new ThreadTaskDemo("t3");
        ThreadTaskDemo t4 = new ThreadTaskDemo("t4");
        ThreadTaskDemo t5 = new ThreadTaskDemo("t4");

        es.execute(t1);
        es.execute(t2);
        es.execute(t3);
        es.execute(t4);
        es.execute(t5);

        System.out.println("执行完毕！");
        es.shutdown();
    }
}

public class ThreadTaskDemo implements Runnable{

    @Getter
    @Setter
    private String value;

    public ThreadTaskDemo(){

    }

    public ThreadTaskDemo(String value){
        this.value=value;
    }

    @Override
    public void run() {
        System.out.println("当前时间 "+LocalDateTime.now().getSecond()+" 当前线程名: "+Thread.currentThread().getName()+" BEGIN "+value );
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

通过控制台输出，我们可以看到，任务只在一个线程中有序执行，说明 maximumPoolSize 参数配置无意义，并未有创建线程的操作。

当前时间 35 当前线程名: pool-1-thread-1 BEGIN t1
当前时间 37 当前线程名: pool-1-thread-1 BEGIN t2
当前时间 39 当前线程名: pool-1-thread-1 BEGIN t3
当前时间 41 当前线程名: pool-1-thread-1 BEGIN t4
当前时间 43 当前线程名: pool-1-thread-1 BEGIN t5

演示效果

3.4.2. 有界队列样例

定义核心线程数，corePoolSize 为 1；有界队列ArrayBlockingQueue 设置 3、同时为展示更好的效果，我也让每个线程执行后都sleep 秒钟。

任务数 为 5，和 maximumPoolSize 设置 6 的情况下：我们看到控制台打印显示有两个线程 pool-1-thread-1 和 pool-1-thread-2。


当前时间 30:14 当前线程名: pool-1-thread-2 BEGIN t5
当前时间 30:14 当前线程名: pool-1-thread-1 BEGIN t1
当前时间 30:16 当前线程名: pool-1-thread-1 BEGIN t2
当前时间 30:16 当前线程名: pool-1-thread-2 BEGIN t3
当前时间 30:18 当前线程名: pool-1-thread-2 BEGIN t4

任务数 为 7，和 最大线程池容量 设置 1 的情况下：控制台打印有异常，这是因为我们任务数超出 队列容量、最大线程池容量的之和，所以应用执行 RejectedExecutionHandler 策略。并且这时候应用状态时挂起，非常不友好，在下一节，写个案例自定义handler。


Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task xyz.wongs.interview.thread.pool.ThreadTaskDemo@4dfa3a9d rejected from java.util.concurrent.ThreadPoolExecutor@6eebc39e[Running, pool size = 1, active threads = 1, queued tasks = 3, completed tasks = 0]
    at java.base/java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2055)
    at java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
    at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
    at xyz.wongs.interview.thread.pool.ThreadPoolExecutorDemo2.main(ThreadPoolExecutorDemo2.java:29)
当前时间 18:16 当前线程名: pool-1-thread-1 BEGIN t1
当前时间 18:18 当前线程名: pool-1-thread-1 BEGIN t2
当前时间 18:20 当前线程名: pool-1-thread-1 BEGIN t3
当前时间 18:22 当前线程名: pool-1-thread-1 BEGIN t4

有界队列执行顺序

综上所述，在 有界队列实现中我们要注意，任务数 、 最大线程池容量、队列容量三者之间的关系。

3.4.3. 自定义`handler`

编写一个 Java类，实现接口 RejectedExecutionHandler，重写 rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法，具体如下

public class CoustomRejectedExecutionHandler implements RejectedExecutionHandler {


    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r instanceof ThreadTaskDemo) {
            ThreadTaskDemo thTk = (ThreadTaskDemo) r;
            //为了演示用，所以直接打印，勿模仿。正式场景下应该持久化或者写入日志！！！
            System.out.println("当前任务 "+ thTk.getValue()+" 执行失败！");
        }
    }
}

我们再运行下例子，我们可以发现并没抛出异常，而且控制应用也关闭。

自定义拒绝策略后验证效果

3.5. 禁用`Executors`创建线程池

通过上面我们简单了解线程池的构造函数参数的意义，我们线程池再线程创建时，其构造函数中指定的队列 LinkedBlockingQueue，这是一种无界的队列，最大值 Integer.MAX_VALUE 即214748364，这队列堆积数量过大，在实际生产中可能直接OOM，不信的话。好奇同学也会说不是还有 newCachedThreadPool，但是它的最大线程数量是 Integer.MAX_VALUE，道理一样，容易造成OOM。

所以很多大型公司在编码规范上都禁止利用 Executors创建线程池。

3.6. 第三方常见创建线程池的方式

3.6.1. 引入 commons-lang3 包方式【不推荐】

1
2

ThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d").daemon(true).build();
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,threadFactory);

依然给最大线程池赋值无上限。
DelayedWorkQueue 延迟阻塞队列，不推荐

3.6.2. 引入 com.google.guava 包方式【一般推荐】

ThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d").daemon(true).build();

ExecutorService pool = new ThreadPoolExecutor(5, 200,
        0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());

4. 常见问题

4.1. newFixedThreadPool(1) 与 newSingleThreadExecutor 区别

newSingleThreadExecutor采用FIFO，保证线程执行顺序，先提交的任务先执行，而newFixedThreadPool(1)不保证。
在newSingleThreadExecutor方法中，当线程执行出现异常时，它会重新创建一个线程替换之前的线程继续执行，而newFixedThreadPool(1)不行。

死磕Java队列

死磕synchronized

死磕ThreadPoolExecutor线程池

1. 线程池的优势

线程空间大小

2. 几种常见线程池

3. 解析ThreadPoolExecutor

3.1. corePoolSize、maximumPoolSize

3.2. workQueue

3.3. handler

3.4. 线程池执行顺序

3.4.1. 无界队列样例

3.4.2. 有界队列样例

3.4.3. 自定义handler

3.5. 禁用Executors创建线程池

3.6. 第三方常见创建线程池的方式

3.6.1. 引入 commons-lang3 包方式【不推荐】

3.6.2. 引入 com.google.guava 包方式【一般推荐】

4. 常见问题

4.1. newFixedThreadPool(1) 与 newSingleThreadExecutor 区别

3.4.3. 自定义`handler`

3.5. 禁用`Executors`创建线程池