搜索

Java中如何优雅的使用线程池？

愿天堂没有BUG

共 1431字，需浏览 3分钟

 · 2021-12-13

为什么要用线程池？

线程是不是越多越好？

  1. 线程在java中是一个对象，更是操作系统的资源，线程创建、销毁需要时间。如果创建时间+小会时间>执行任务时间就很不合算。

  2. java对象占用堆内存，操作系统线程占用系统内存，根据jvm规范，一个线程默认最大栈大小1M，这个栈空间是需要从系统内存中分配的。线程过多，会消耗很多的内存。

  3. 操作系统需要频繁切换线程上下文（每个线都想被运行），影响性能。

线程池的推出，就是为了方便边的控制线程数量。

线程池

线程池基本概念

线程池包括以下四个基本组成部分：

  1. 线程池管理器：用于创建并管理线程池，包括创建线程池，销毁线程池，添加新任务；

  2. 工作线程：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；

  3. 任务接口：每个任务必须实现的借口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；

  4. 任务队列：用于存放没有处理的任务。提供一种缓冲机制。

线程池接口定义和实现类

可以认为ScheduledThreadPoolExector是最丰富的实现类。

ExecutorService

public interface ExecutorService extends Executor {
    /**
     * 优雅关闭线程池，之前提交的任务将被执行，但是不会接受新的任务。
     */
    void shutdown();

    /**
     * 尝试停止所有正在执行的任务，停止等待任务的处理，并返回等待执行任务的列表。
     */
    List shutdownNow();

    /**
     * 如果此线程池已关闭，则返回true.
     */
    boolean isShutdown();

    /**
     * 如果关闭后的所有任务都已完成，则返回true
     */
    boolean isTerminated();

    /**
     * 监测ExecutorService是否已经关闭，直到所有任务完成执行，或超时发生，或当前线程被中断。
     */
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

    /**
     * 提交一个用于执行的Callable返回任务，并返回一个Future，用于获取Callable执行结果。
     */
     Future submit(Callable task);

    /**
     * 提交可运行任务以执行，并返回Future，执行结果为传入的result
     */
     Future submit(Runnable task, T result);

    /**
     * 提交可运行任务以执行，并返回Future对象，执行结果为null
     */
    Future submit(Runnable task);

    /**
     * 执行给定的任务集合，执行完毕后，则返回结果。
     */
     List> invokeAll(Collection> tasks) throws InterruptedException;

    /**
     * 执行给定的任务集合，执行完毕或者超时后，则返回结果，其他任务终止。
     */
     List> invokeAll(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException;

    /**
     * 执行给定的任务，任意一个执行成功则返回结果，其他任务终止。
     */
     T invokeAny(Collection> tasks) throws InterruptedException, ExecutionException;

    /**
     * 执行给定的任务，任意一个执行成功或者超时后，则返回结果，其他任务终止
     */
     T invokeAny(Collection> tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
}

复制代码

ScheduledExecutorService

public interface ScheduledExecutorService extends ExecutorService {

    /**
     * 创建并执行一个一次性任务，过了延迟时间就会被执行
     */
    public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit);

    /**
     * 创建并执行一个一次性任务，过了延迟时间就会被执行
     */
    public  ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit);

    /**
     * 创建并执行一个周期性任务，过了给定的初始化延迟时间，会第一次被执行。执行过程中发生了异常，那么任务停止
     * 一次任务执行时长超过了周期时间，下一次任务会等到该次任务执行结束后，立刻执行，这也是它和scheduleWithTixedDelay的重要区别
     */
    public ScheduledFuture scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

    /**
     * 创建并执行一个周期性任务，过了给定的初始化延迟时间，会第一次被执行。执行过程中发生了异常，那么任务停止
     * 一次任务执行时长超过了周期时间，下一次任务会在该次任务执行结束的时间基础上，计算执行延时。
     * 对于超时周期的长时间处理任务的不同处理方式，这是它和scheduleAtFixedRate的重要区别
     */
    public ScheduledFuture scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

}
复制代码

线程池工具类

在使用过程中，可以自己实例化线程池，也可以用Executors创建线程池的工厂累，常用方法如下：

newFixedThreadPool(int nThreads)

创建一个固定大小、任务队列容量误解的线程池。核心线程数=最大线程数。

newCachedThreadPool()

创建的是一个大小无界的缓冲线程池。它的任务队列是一个同步队列。任务加入到池中，如果池中有空闲线程，则用空闲线程执行，如无则创建新线程执行。池中的线程空闲时间超过60秒，将被销毁释放。线程数随任务的多少变化。适用于执行耗时较小的异步任务。池的核心线程数=0，最大线程=Integer.MAX_VALUE

newSingleThreadExecutor()

只有一个线程来执行无界任务队列的单一线程池。该线程池确保任务加入的顺序一个一个一次执行。当唯一的线程因任务异常中止时，将创建一个新的线程来继续执行后续的任务。与newFixedThreadPool(1)的区别在于，单一线程池的池大小在newSingleThreadExecutor方法中硬编码，不能再改变的。

newScheduledThreadPool(int corePoolSize)

能定时执行任务的线程池。该池的核心线程数由参数指定，最大线程数=Integer.MAX_VALUE

任务线程池执行过程

如何确认合适的线程数量？

  • 如果是CPU密集型应用，则线程池大小设置为N+1 （N为CPU总核数）

  • 如果是IO密集型应用，则线程池大小设置为2N+1 （N为CPU总核数）

  • 线程等待时间(IO)所占比例越高，需要越多线程。

  • 线程CPU时间所占比例越高，需要越少线程。

一个系统最快的部分是CPU，所以决定一个系统吞吐量上限的是CPU。增强CPU处理能力，可以提高系统吞吐量上限。但根据短板效应，真实的系统吞吐量并不能单纯根据CPU来计算。那要提高系统吞吐量，就需要从“系统短板”（比如网络延迟、IO）着手：

  1. 尽量提高短板操作的并行化比率，比如多线程下载技术；

  2. 增强短板能力，比如用NIO替代IO；

线程池的使用分析

public class ExecutorsUse {
    /**
     * 测试：提交15 个执行时间需要3秒的任务,看线程池的状况
     *
     * @param threadPoolExecutor 传入不同的线程池，看不同的结果
     * @throws Exception
     */
    public void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
        // 测试：提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(() -> {
                        try {
                            System.out.println("开始执行：" + n);
                            Thread.sleep(3000L);
                            System.err.println("执行结束:" + n);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
            );
            System.out.println("任务提交成功 :" + i);
        }
        // 查看线程数量，查看队列等待数量
        Thread.sleep(500L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
        // 等待15秒，查看线程数量和队列数量（理论上，会被超出核心线程数量的线程自动销毁）
        Thread.sleep(15000L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
    }

    /**
     * 1、线程池信息：核心线程数量5，最大数量10，无界队列，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest1() throws Exception {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue());
        testCommon(threadPoolExecutor);
        // 预计结果：线程池线程数量为：5,超出数量的任务，其他的进入队列中等待被执行
    }

    /**
     * 2、 线程池信息：核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest2() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue(3), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("有任务被拒绝执行了");
            }
        });
        testCommon(threadPoolExecutor);
        // 预计结果：
        // 1、 5个任务直接分配线程开始执行
        // 2、 3个任务进入等待队列
        // 3、 队列不够用，临时加开5个线程来执行任务(5秒没活干就销毁)
        // 4、 队列和线程池都满了，剩下2个任务，没资源了，被拒绝执行。
        // 5、 任务执行，5秒后，如果无任务可执行，销毁临时创建的5个线程
    }

    /**
     * 3、 线程池信息：核心线程数量5，最大数量5，无界队列，超出核心线程数量的线程存活时间：5秒
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest3() throws Exception {
        // 和Executors.newFixedThreadPool(int nThreads)一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue());
        testCommon(threadPoolExecutor);
        // 预计结：线程池线程数量为：5，超出数量的任务，其他的进入队列中等待被执行
    }

    /**
     * 4、 线程池信息：
     * 核心线程数量0，最大数量Integer.MAX_VALUE，SynchronousQueue队列，超出核心线程数量的线程存活时间：60秒
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest4() throws Exception {

        // SynchronousQueue，实际上它不是一个真正的队列，因为它不会为队列中元素维护存储空间。与其他队列不同的是，它维护一组线程，这些线程在等待着把元素加入或移出队列。
        // 在使用SynchronousQueue作为工作队列的前提下，客户端代码向线程池提交任务时，
        // 而线程池中又没有空闲的线程能够从SynchronousQueue队列实例中取一个任务，
        // 那么相应的offer方法调用就会失败（即任务没有被存入工作队列）。
        // 此时，ThreadPoolExecutor会新建一个新的工作者线程用于对这个入队列失败的任务进行处理（假设此时线程池的大小还未达到其最大线程池大小maximumPoolSize）。

        // 和Executors.newCachedThreadPool()一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new SynchronousQueue());
        testCommon(threadPoolExecutor);
        // 预计结果：
        // 1、 线程池线程数量为：15，超出数量的任务，其他的进入队列中等待被执行
        // 2、 所有任务执行结束，60秒后，如果无任务可执行，所有线程全部被销毁，池的大小恢复为0
        Thread.sleep(60000L);
        System.out.println("60秒后，再看线程池中的数量：" + threadPoolExecutor.getPoolSize());
    }

    /**
     * 5、 定时执行线程池信息：3秒后执行，一次性任务，到点就执行 

     * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest5() throws Exception {
        // 和Executors.newScheduledThreadPool()一样的
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        threadPoolExecutor.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("任务被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 3000, TimeUnit.MILLISECONDS);
        System.out.println(
                "定时任务，提交成功，时间是：" + System.currentTimeMillis() + ", 当前线程池中线程数量：" + threadPoolExecutor.getPoolSize());
        // 预计结果：任务在3秒后被执行一次
    }

    /**
     * 6、 定时执行线程池信息：线程固定数量5 ，

     * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest6() throws Exception {
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        // 周期性执行某一个任务，线程池提供了两种调度方式，这里单独演示一下。测试场景一样。
        // 测试场景：提交的任务需要3秒才能执行完毕。看两种不同调度方式的区别
        // 效果1：提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，完毕后立刻执行)。
        // 也就是说这个代码中是，3秒钟执行一次（计算方式：每次执行三秒，间隔时间1秒，执行结束后马上开始下一次执行，无需等待）
        threadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务-1 被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 2000, 1000, TimeUnit.MILLISECONDS);

        // 效果2：提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，等上一次执行完毕后再开始计时，等待1秒)。
        // 也就是说这个代码钟的效果看到的是：4秒执行一次。（计算方式：每次执行3秒，间隔时间1秒，执行完以后再等待1秒，所以是 3+1）
        threadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务-2 被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 2000, 1000, TimeUnit.MILLISECONDS);
    }

    /**
     * 7、 终止线程：线程池信息：核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest7() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue(3), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("有任务被拒绝执行了");
            }
        });
        // 测试：提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        threadPoolExecutor.shutdown();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdown后，不接收新的任务，等待13任务执行结束
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }

    /**
     * 8、 立刻终止线程：线程池信息：核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
     *
     * @throws Exception
     */
    private void threadPoolExecutorTest8() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue(3), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("有任务被拒绝执行了");
            }
        });
        // 测试：提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        List shutdownNow = threadPoolExecutor.shutdownNow();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        System.out.println("未结束的任务有：" + shutdownNow.size());

        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdownnow后，队列中的3个线程不再执行，10个线程被终止
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }

    public static void main(String[] args) throws Exception {
//        new ExecutorsUse().threadPoolExecutorTest1();
//        new ExecutorsUse().threadPoolExecutorTest2();
//          new ExecutorsUse().threadPoolExecutorTest3();
        new ExecutorsUse().threadPoolExecutorTest4();
//        new ExecutorsUse().threadPoolExecutorTest5();
//        new ExecutorsUse().threadPoolExecutorTest6();
//        new ExecutorsUse().threadPoolExecutorTest7();
//        new ExecutorsUse().threadPoolExecutorTest8();
    }
}


作者：细思极恐
链接：https://juejin.cn/post/6986863481764970510
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。



浏览 15
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
了解加密货币到加密货币的互换
1、什么是加密货币互换？加密货币到加密货币的互换是指以现行市场汇率将一种加密货币直接兑换为另一种加密货币。与需要法定货币存款和较长流程的传统交易所不同，加密货币到加密货币的互换可以无缝地促进交换。掉期在提高加密货币的流动性和效率方面发挥着重要作用。该功能使用户能够将他们的加密货币与钱包中的其他代币进
区块链头条
0
李彦宏：开源大模型不如闭源，后者会持续领先；周鸿祎：“开源不如闭源” 的言论是胡说八道
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁0、李彦宏：开源大模型不如闭源，后者会持续领先当今
源码共读
0
【第129期】程序员的新宠：三款终端工具，让你告别Xshell！
概述 WindTerm：跨平台的SSH利器 首先介绍的是WindTerm，这是一款使用C语言开发的跨平台SSH客户端。它不仅完全免费，而且没有商业使用的限制。WindTerm支持SSH v2、Telnet、Raw Tcp等协议，而且性能出色，甚至超过了FinalShell和Electerm。功能
前端微服务
0
字节员工：35岁以后被裁员的，后来都走了哪条路？现在2-2，要不要利用最后一年拼命上个岸。
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁在当今竞争激烈的职场环境中，年龄并不总是一个决定性
源码共读
0
【送书福利】《Java面试八股文：高频面试题与求职攻略一本通》
先来唠唠最近粉丝面试回来跟我聊天，基本上都提到一个点，在面试过程中八股文占比很高（八股文70%、项目20%、10%算法）除了一些搞算法突出的厂除外。其实现在很多厂八股都是逐渐深入的方式来问，所以大家在学习的过程中，针对一些重点的内容，最好深入去学习，不然还是比较难应对这种追问式的问题。最近刚好从一位
Java后端技术
0
上班的时候，有一群摸鱼搭子非常重要...
上班的时候，有一群摸鱼搭子非常重要！一到上班时间，他们就从四面八方涌进群里冒泡...从八卦聊到股市、从职场聊到乌X兰局势，偶尔还会复读、相亲、battle...然后，下午6点钟准时消失不见...所以你要不要加入我们一起摸鱼？我们有北京、上海、深圳、广州、杭州、武汉、成都、南京等8个城市的摸鱼群，还有
产品经理日记
0
周四002 瑞超：同样落寞的境遇——北雪平vs埃尔夫斯堡
上赛季最终排名联赛第9的北雪平本赛季伊始表现不佳，4轮战罢他们仅以1胜1平2负的战绩排在倒数第三，这支历史上曾夺得13次联赛冠军、6次杯赛冠军老牌劲旅，正如英格兰赛场上的一众百年俱乐部，在低谷中不断探索着出路。球队主教练安德烈亚斯·阿尔姆曾是AIK索尔纳及赫根队的主教练，他于今年年初刚刚拿起球队教鞭
产品与体验
0
雷军辟谣了！不是高考状元，卡里也没有冰冷的 40 亿
架构师大咖 架构师大咖，打造有价值的架构师交流平台。分享架构师干货、教程、课程、资讯。架构师大咖，每日推送。 公众号该公众号已被封禁最近很火的雷军简历，听说落魄时卡里只有冰冷的 40
源码共读
0
日本影山优佳最新杂志照，展现充满透明感的美丽
今天的图文分享的是影山优佳的杂志写真。元日向坂46的影山优佳，登上了写真杂志《周刊FLASH》5/7和‬5/14合并号的封面。影山优佳是日本艺人、女演员、前偶像。身高155厘米。2001年‬5月‬8日‬出生‬于‬东京都。2023年7月从组合日向坂46毕业，之后作为演员活跃的影山优佳，在《周刊FLAS
python教程
0
盘点一个使用超级鹰识别验证码并自动登录的案例
点击上方“Python共享之家”，进行关注回复“资源”即可获赠Python学习资料今日鸡汤江上几人在，天涯孤棹还。大家好，我是皮皮。一、前言前几天在Python钻石交流群【静惜】问了一个Python实现识别验证码并自动登录的问题，提问截图如下：验证码的截图如下所示：二、实现过程这里大家激烈的探讨，【
IT共享之家
0
点赞
评论
收藏
分享

手机扫一扫分享

举报