• java ThreadPoolExecutor 分析

    前言:ThreadPoolExecutor(线程池) 是早些年的东西,但是很久时间不用就会忘记怎么用,仅此篇来记录一下!

    thread知识你还记得多少?

    要想搞懂ThreadPoolExecutor 那么就得从实际场景出发

    并且对比原始new Thread() 看看他们有什么区别

    假设现在需要下载100张图片:

    普通写法:

    public class TPEClient1 {
    private static final int COUNT = 100;

    public static void main(String[] args) {

        for (int i = 0; i < COUNT; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println("线程:" + Thread.currentThread().getName());
                    }
                }
            }).start();
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21

    可以看出,睡眠2秒之后, 确实是创建了100个线程,然后执行了任务

    image-20221021100157503

    但是这里线程的执行顺序为啥是无序的呢?

    那是因为在 Thread#start() 时候,线程不会立刻执行,

    而是告诉CPU我需要执行,然后等待CPU去调度

    简单的说就是: Thread#start() 只是通知CPU我准备好了, CPU什么时候用,怎么用,都是CPU说了算.

    直接new Thread()不足的地方:

    很明显, 直接new Thread() 每次都会创建一个单独的线程来执行相同的任务,

    看起来代码十分的丑陋,并且也很浪费资源,

    最重要的是不好管理, 如果代码中到处是new thread(),维护起来肯定十分的不爽

    那么来看看ThreadPoolExecutor是怎么做的

    ThreadPoolExecutor

    先来看如何创建一个ThreadPoolExecutor()

    LinkedBlockingQueue<Runnable> blockQueue = new LinkedBlockingQueue<>(100); // 最多存放100个数据

ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(4, // 核心线程数
                10, // 最大线程数
                10, // 失效时间
                TimeUnit.SECONDS, // 失效时间单位
                blockQueue, // 队列
                new ThreadFactory() { // 线程工厂
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r);
                    }
                },
                // 失败策略
                new ThreadPoolExecutor.AbortPolicy()
        );

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    这里有最大争议的恐怕是

    • 核心线程数
    • 最大线程数

    还是上面的需求: 假设现在需要下载100张图片,来看看线程池是怎么做的:

    首先需要100个任务

    image-20221021111425684

    创建线程池,正常情况下线程池中的线程是由核心线程数决定的, 我们设置核心线程数为4,那么就创建4个线程

    image-20221021111446053

    在构建线程池的时候,还需要传入一个队列

    这个队列的作用就是来管理100个任务,

    image-20221021131939008

    当任务来临的时候,将任务加入到队列中,因为有100个任务,

    并且在创建队列的时候,我们设置队列的容量为100,刚好可以存放这100个任务

    加入队列

    因为在创建线程时只创建4了个线程,所以只能从队列中取出4个任务来执行,

    当有任务执行完成后,在判断队列中是否还有任务,如果还有任务,那么就继续执行新的任务

    执行

    从此以来,反复的利用这4个线程,并不会创建额外的线程!

    这是比较理想状态的的线程池,再来回顾一下构建ThreadPoolExecutor的参数

    // 只能存放100个任务
LinkedBlockingQueue<Runnable> blockQueue = new LinkedBlockingQueue<>(100);

ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(4, // 核心线程数
                10, // 最大线程数
                10, //
                TimeUnit.SECONDS,
                blockQueue,
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r);
                    }
                });

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15

    @param corePoolSize(核心线程数) 4: 在线程池中创建4个线程

    @param maxMumPoolSize(最大线程数) 10: 最多可以使用10个线程

    @param keepAliveTime(线程等待时间):10

    @param TimeUnit(TimeUnit.SECONDS): 线程等待单位

    @param BlockQueue: 存放任务的队列

    @param ThreadFactory:线程工厂: 用来创建线程

    在上面图解中,我们使用到了核心线程数和 队列,知道了线程池运行的基础逻辑

    那么先来看一眼上面图解对应的代码:

    public class TPEClient2 {
    private static final int COUNT = 10;

    public static void main(String[] args) {
        // 只能存放100个任务
        LinkedBlockingQueue<Runnable> blockQueue = new LinkedBlockingQueue<>(10);

        ThreadPoolExecutor threadPoolExecutor =
                new ThreadPoolExecutor(
                        4, // 核心线程数
                        10, // 最大线程数
                        3, // 等待3秒
                        TimeUnit.SECONDS,
                        blockQueue,
                        new ThreadFactory() {
                            @Override
                            public Thread newThread(Runnable r) {
                                return new Thread(r);
                            }
                        });

        for (int i = 0; i < COUNT; i++) {
            int finalI = i;
            Runnable runnable = () -> {
                try {
                    Thread.sleep(2000);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("i:" + finalI + "\t线程:" + Thread.currentThread().getName());
                }
            };
            threadPoolExecutor.execute(runnable);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    tips: 这里就不创建100个任务,创建10个任务也是同样的道理

    Oct-21-2022 13-59-41

    可以看出,这里线程是每4个一起执行的,因为 Thread.sleep(2000);

    并且通过Thread.currentThread().getName() 可以看出,只创建了4个线程, 线程对象都在重复利用

    但是有一个奇怪的一点,为什么我设置了3秒等待时间,10个任务结束之后,为什么我等了3秒,他还是在运行状态呢?

    在线程池执行过程中分别有5种状态,(在源码中可以搜索到)

    • RUNNING：接受新任务并处理排队的任务。
    • SHUTDOWN：不接受新任务，但处理排队的任务。
    • STOP：不接受新任务，不处理排队的任务，中断正在进行的任务。 (隐含：此状态下，队列已为空)
    • TIDYING：所有任务都已终止，workerCount 为零，转换到状态 TIDYING 的线程将运行 terminate() 钩子方法。
    • TERMINATED：终止（）已完成。

    如果当队列中没有没有任务,并且等待keepAliveTime后

    需要通过 ThreadPoolExecutor#allowCoreThreadTimeOut(true); 将状态设置为 终止状态(TERMINATED)

    Oct-21-2022 14-13-10

    这样一来,当任务执行完成之后,并且等待keepAliveTime后就会自动终止了

    最大线程数和核心线程数以及失败策略

    首先吧最大线程数和核心线程数放到一边, 先来看看失败策略

    创建一个线程池

    LinkedBlockingQueue<Runnable> blockQueue = new LinkedBlockingQueue<>(100); // 最多存放100个数据

ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(4, // 核心线程数
                10, // 最大线程数
                10, // 失效时间
                TimeUnit.SECONDS, // 失效时间单位
                blockQueue, // 队列
                new ThreadFactory() { // 线程工厂
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r);
                    }
                },
                // 失败策略
                new ThreadPoolExecutor.AbortPolicy()
        );	
        

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    在线程池中的最后一个参数,就是失败策略 即 ThreadPoolExecutor.AbortPolicy()

    首先要考虑的当然是什么东西失败了? 为什么会失败?

    在这个ThreadPoolExecutor中, 队列最多可存放100个任务

    假设当前需要执行200个任务, 那么200个任务就放不到100的队列中, 这时候失败策略就起作用了!

    如果一旦失败,最大线程数也会发挥作用

    我们知道,

    • 核心线程数为 4
    • 最大线程数为10

    那么非核心线程数 = 6

    一旦当前需要执行的任务 > 队列最大容量,那么就会创建所有的线程来共同执行任务,也就是会创建10个线程(4个核心线程,6个非核心线程)来共同执行任务

    即使是创建10个线程,也不会吧这200个任务全部执行完, 因为这200个任务没有完全添加到队列中

    来看一眼代码:

    public class TPEClient3 {
    private static final int COUNT = 200;

    public static void main(String[] args) {

        LinkedBlockingQueue<Runnable> blockQueue = new LinkedBlockingQueue<>(100);


        ThreadPoolExecutor threadPoolExecutor =
                new ThreadPoolExecutor(4, // 核心线程数
                        10, // 最大线程数
                        10, // 失效时间
                        TimeUnit.SECONDS, // 失效单位
                        blockQueue, // 队列
                        new ThreadFactory() { // 线程工厂
                            @Override
                            public Thread newThread(Runnable r) {
                                return new Thread(r);
                            }
                        },
                        // 失败策略
                        new ThreadPoolExecutor.AbortPolicy()
                );


        for (int i = 0; i < COUNT; i++) {
            int finalI = i;
            Runnable runnable = new Runnable() {
                @Override
                public void run() {

                    try {
                        Thread.sleep(2000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println("i:" + finalI + "\t线程:" + Thread.currentThread().getName());
                    }
                }
            };

            threadPoolExecutor.execute(runnable);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45

    Oct-21-2022 14-30-57

    常见的队列

    • BlockingQueue() 先进先出队列
    • SynchronousQueue() 线程安全队列
    • PriorityBlockingQueue() 优先阻塞队列 无序队列,可以根据优先级排序

    总结

    其实ThreadPoolExecutor 很简单,

    通过核心线程数来创建默认的线程, 如果任务过多,超出队列的个数,那么就会触发失败策略

    触发失败策略的同时 就会创建最大线程数个线程, 唤醒所有的线程来执行任务!

    完整代码

    原创不易,您的点赞就是对我最大的支持!

    推荐文章:

  • 相关阅读:
    Qt富文本语法高亮
    js文字逐个显示
    笔记本电脑的摄像头找不到黑屏解决办法
    某银行客户Db2下移分布式数据库OceanBase案例
    Pytorch学习笔记（一）官方60min入门教程之张量
    Windows通过RDP异地远程桌面Ubuntu【内网穿透】
    简述分布式链路追踪工具——Jaeger
    IO学习系列之使用多线程复制同一个文件内容
    PB的扩展DLL开发（超级篇）（七）
    Adobe Premiere Pro：掌控视频剪辑的魔法之手，让你的创作腾飞！
  • 原文地址：https://blog.csdn.net/weixin_44819566/article/details/127446410