在上海做兼职去哪个网站搜索,怎么做关注网站,wordpress文章保存图片不显示图片,网站文章只被收录网站首页Java中的线程池是运用场景最多的并发框架#xff0c;几乎所有需要异步或并发执行任务的程序都可以使用线程池#xff0c;本篇文章就详细介绍一下。
一、什么是线程池 定义#xff1a;线程池是一种用于管理和重用线程的技术#xff08;池化技术#xff09;#xff0c;它主…Java中的线程池是运用场景最多的并发框架几乎所有需要异步或并发执行任务的程序都可以使用线程池本篇文章就详细介绍一下。
一、什么是线程池 定义线程池是一种用于管理和重用线程的技术池化技术它主要用于提高多线程应用程序的性能和效率。 ps线程池、连接池、内存池和对象池等都是编程领域中典型的池化技术。 首先有关线程的使用会出现两个问题 线程是宝贵的内存资源、单个线程约占1MB空间过多分配易造成内存溢出。频繁的创建及销毁线程会增加虚拟机回收频率、资源开销造成性能下降。 基于如上问题出现了线程池 线程容器可设定线程分配的数量。将预先创建的线程对象存入池中并重用线程池中的线程对象。避免频繁的创建和销毁。 Java中线程池的继承关系如下 二、ThreadPoolExecutor
Executor框架最核心的类是ThreadPoolExecutor我们可以通过ThreadPoolExecutor类来创建一个线程池。
2.1 ThreadPoolExecutor构造函数
ThreadPoolExecutor继承自AbstractExecutorService而AbstractExecutorService实现了ExecutorService接口。 2.2 线程池的7个核心参数
从ThreadPoolExecutor的构造函数可以看出创建一个线程池需要7个核心参数下面我们介绍一下这7个参数的含义 核心线程数数量corePoolSize当线程池被创建时在你池子中初始化多少个线程。 最大线程数maximumPoolSize当我的所有核心线程数都去干活时又来了一个任务如果我的当前线程数小于我最大线程数这时候可以再帮你创建一个线程去指定你的任务。 线程闲置时间keepAliveTime额外线程完成任务后存活的时间。 限制时间的单位unit)存活时间单位。 工作队列workQueue当没核心线程去处理任务时会把任务放在工作队列中当有闲下来的线程时再去执行队列的任务常见的工作队列有以下几种前三种用的最多 ArrayBlockingQueue列表形式的工作队列必须要有初始队列大小有界队列先进先出FIFO。 LinkedBlockingQueue链表形式的工作队列可以选择设置初始队列大小有界设置了初始大小/无界队列没设置 先进先出FIFO。 SynchronousQueue这不是一个真正的队列而是一种在线程之间移交的机制。要将一个元素放入SynchronousQueue中必须有另一个线程正在等待接受这个元素。如果没有线程等待并且线程池的当前大小小于最大值那么ThreadPoolExecutor将创建一个线程否则根据拒绝策略这个任务将被拒绝。使用直接移交更高效因为任务会直接移交给执行它的线程而不是被首先放在队列中然后由工作者线程从队列中提取任务。只有当线程是无界的或者可以拒绝任务时SynchronousQueue才有价值。 PriorityBlockingQueue优先级队列有界队列根据优先级来安排任务任务的优先级是通过自然顺序或Comparator来定义的。 DelayedWorkQueue延迟的工作队列无界队列。 创建线程的工厂threadFactory线程池不会帮你创建线程这时候就要用到线程工厂 DefaultThreadFactory默认线程工厂创建一个新的、非守护的线程并且不包括特殊的配置信息。 PrivilegedThreadFactory通过这种方式创建出来的线程将与创建privilegedThreadFactory的线程拥有相同的访问权限、AccessControlContext、ContextClassLoader。如果不使用privilegedThreadFactory线程池创建的线程将从在需要新线程时调用execute或submit的客户程序中继承访问权限。 自定义线程工厂可以自己实现ThreadFactory接口来自定义线程工厂。 拒绝策略handler当你的线程数达到最大工作队列任务也满了就执行拒绝策略 AbortPolicy抛出异常RejectedExecutionException默认的拒绝策略。调用者可以将异常进行捕获然后根据需求处理代码 CallerRunsPolicy调用者自己处理任务要把任务派发给我的线程池要有一个线程执行操作如果没有闲置的线程由调用者自己处理任务 DiscardOldestPolicy丢弃任务队列中最老的任务把自己放进去 DiscardPolicy丢弃掉当前任务 2.3 线程池Demo
我们可以使用ThreadExecutor创建一个线程池然后调用execute()或submit()的方法来向线程池中提交任务。
2.3.1 execute()方法自己的方法 public void execute(Runnable command) execute()方法没有返回值所以它适用于不需要返回值的任务当然也无法判断任务是否被线程池执行成功。
public class ThreadPoolExecutorTest {public static void main(String[] args) {//1.创建线程池ThreadPoolExecutor poolExecutor new ThreadPoolExecutor(2,3, 60L, TimeUnit.SECONDS,new LinkedBlockingQueue(1),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardOldestPolicy());//2.创建任务Runnable runnable () - {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() 执行了。。。);};//3.提交任务for (int i 0; i 5; i) {poolExecutor.execute(runnable);}//4.关闭线程池poolExecutor.shutdown();}
}
上述例子我们创建了一个核心线程数为2、最大线程数为3的线程池然后通过execute方法提交了5个线程给线程池执行。
2.3.2 submit()方法AbstractExecutorService父类的方法 public Future? submit(Runnable task) public T FutureT submit(Runnable task,T result) public T FutureT submit(CallableT task) sublmit()方法用于提交需要返回值的任务。线程池会返回一个future类型的对象通过这个对象可以判断任务是否执行成功并且可以通过get()方法以阻塞的方式来获取返回值。
public class ThreadPoolExecutorTest {public static void main(String[] args) throws ExecutionException, InterruptedException {//1.创建线程池ThreadPoolExecutor poolExecutor new ThreadPoolExecutor(1,1, 60L, TimeUnit.SECONDS,new LinkedBlockingQueue(1),Executors.defaultThreadFactory(),new ThreadPoolExecutor.DiscardOldestPolicy());//2.提交任务Future标识将要执行任务的结果submit可以传入一个CallableT对象FutureInteger future poolExecutor.submit(new CallableInteger() {private int sum 0;Overridepublic Integer call() throws Exception {for (int i 0; i 100; i) {sum i;Thread.sleep(10);}return sum;}});//3.获取任务的结果等待任务执行完毕才返回阻塞System.out.println(future.get());//4.关闭线程池poolExecutor.shutdown();}
}
2.4 当一个任务提交到线程池后 首先查看核心线程数是否达到最大 否利用线程工厂创建核心线程直接执行任务 是尝试将任务放到工作队列中 查看工作队列是否已满 否将任务放入到工作队列。 是尝试创建非核心线程最大线程数-核心线程数。 查看非核心线程数是否达到最大 否利用线程工厂创建非核心线程执行任务 是执行拒绝策略 2.5 常用方法
除了在创建线程池时指定上述参数值外还可在线程池创建后通过如下方法进行设置 此外还有一些方法 getCorePoolSize()返回线程池的核心线程数这个值是一直不变的返回在构造函数中设置的coreSize大小getMaximumPoolSize()返回线程池的最大线程数这个值是一直不变的返回在构造函数中设置的coreSize大小getLargestPoolSize()记录了曾经出现的最大线程个数水位线getPoolSize()线程池中当前线程的数量getActiveCount()Returns the approximate近似 number of threads that are actively executing tasksprestartAllCoreThreads()会启动所有核心线程无论是否有待执行的任务线程池都会创建新的线程直到池中线程数量达到 corePoolSizeprestartCoreThread()会启动一个核心线程同上allowCoreThreadTimeOut(true)允许核心线程在KeepAliveTime时间后退出 2.6 线程数设定多少合适
这个是个高频的面试点至于设定多少流传最多的就是“公式”所谓的公式一般情况下需要根据你的任务情况来设置线程数任务可能分为两种类型——CPU密集型和IO密集型 CPU密集型则线程池大小设置为N1。IO密集型则线程池大小设置为2N1。 上面的N为CPU总核数但在实际场景下公式只能是当作一个参考。 很多时候我们应用部署在云服务器上有时候我们的机器显示是8核的但实际上适用的只是虚拟机而已并不是物理机这就导致大多数情况下发挥不出8核的作用来。
而且上面的公式中前提要求是知道你的应用是IO密集型还是CPU密集型那么评判一个应用是CPU密集型还是IO密集型的标准是什么真的能明确区分出来吗
还有一点就是现在很多CPU都采用了超线程技术也就是利用特殊的硬件指令把两个逻辑内核模拟成两个物理芯片让单个处理器都能适用线程级并行计算。所以我们可以看到“4核8线程的CPU”也就是物理内核有4个逻辑内核有8个如果用上述的公式貌似按照4和8配置都不合理因为超线程技术的性能提升也并不是100%的。 所以在设定线程数的时候要考虑什么业务场景、什么机器配置、多大的并发量、一次业务处理整体耗时是多少最后在上线的时候可以根据公式大致设置一个数值然后再根据你自己的实际业务情况以及不断的压测结果再不断调整最终达到一个相对合理的值。 三、线程池的实现原理源码分析
下面我们通过源码来分析一下ThreadPoolExecutor类中三个比较核心的方法。
3.1 添加一个任务execute()方法
首先我们看一下线程池中的execute()方法该方法用于向线程池中添加一个任务。
源码 分析 第一个红框先检查是否有空闲的核心线程 workCountOf(c)方法根据ctl的低29位来得到有效的线程数。 判断有效的线程数是否小于核心线程数。 如果是是。则创建一个线程来处理任务核心线程。 第二个红框走到这说明核心线程数已满了 isRunning(c)方法判断当前是否是运行状态如果是则尝试能否将任务放入工作队列(work.offer(command)方法。 如果也添加成功了。则再次拿到ctl值再次检查状态如果不再运行并且移除添加的任务成功则抛出拒绝策略。 如果在运行则检查有效线程数是否为0如果是则新建一个线程非核心线程。 第三个红框这里说明工作队列也满了 addwork(command,flase)方法尝试新建一个线程来处理任务非核心。 如果失败则调用拒绝策略。 3.2 添加work线程addworker()
从方法execute的实现可以看出addWorker主要负责创建新的线程并执行任务。
这块代码比较长所以我们把它分成两段来介绍先看第一段。
源码 分析 第一个红框做是否能够添加工作线程条件过滤这里的情况比较多仅进行关键代码的解释 rs为运行状态源码中频繁适用大小关系来作为条件判断大小关系RUNNING运行SHUTDOWN关闭STOP停止TIDYING整理TERMINATED终止。 firstTask判断提交的任务是否为空。 return false表示不处理提交的任务直接返回。 第二个红框做自旋更新创建线程数量 第一个if校验有效线程数是否超过阈值如果超过则不处理提交的任务。 第二个if适用CAS讲workerCount1修改成功则跳出循环。 第三个if重新读取ctl判断当前运行状态如果不等于上面获取的运行状态rs说明rs被其它线程修改了跳到retry重新校验线程池的状态。 psretry是java中的goto语法只能运行在break和cotinue后面。 接着看后面的代码。
源码 分析先说下两个变量 workerStarted——Worker线程是否启动workerAdded——Worker线程是否成功增加。 第一个红框获取线程池主锁 用firstTask和当前线程创建一个Worker。 拿到Worker对应的线程t。 如果t不为空获取线程池主锁通过ReentrantLock锁来保证线程安全。 第二个红框添加线程到workers中线程池中。 第三个红框如果woker添加成功则启动新建的线程执行。 我们看看这个wokers是什么 一个HashSet所以线程池的存储结构其实就是一个HashSet。
3.3 worker线程处理队列任务runWorker
上文addWorker()方法里提到当Worker里的线程启动时就会调用该方法。
源码
分析 第一个红框取任务执行如果是第一次执行任务或者能从队列中取到任务。 第二个红框获取到任务后执行任务开始前操作钩子。 第三个红框执行任务。 第四个红框执行任务后钩子 ps这两个钩子beforeExecute、afterExecute允许我们自己继承线程池也就是我们自己可以重写这两个方法做任务执行前、后的处理。 四、Java中自带的5种线程池
这五种线程池都是由Executors工具类提供该类看起来功能还是比较强大的又用到了工厂模式扩展性很强重要的是用起来还特别方便如
//创建一个固定大小的线程池
ExecutorService executorService Executors.newFixedThreadPool(1);
接下来我们详细分析一下Java自带的五种线程池。
4.1 newFixedThreadPool
FixedThreadPool被称为可重用固定线程数的线程池源码如下
可以看到核心线程数corePoolSize和最大线程数maximumPoolSize都是传进来的nThreads这就意味着这个数量在其生命周期内不会变化当线程池中的线程都在工作时新提交的任务会被放置到一个无界队列中没有设置初始大小等待被执行。
4.2 newWorkStealingPool
用来执行大任务的线程池WorkStealingPool可以自适应地调整线程池大小它会根据处理器核数创建相应数量的线程池。源码如下
可以看出ForkJoinPool的构造函数被调用来创建WorkStealingPool线程池Runtime.getRuntime().availableProcessors()来获取处理器的核心数量来作为线程池的大小
为什么说它是执行大任务的线程池呢因为它本质是一个ForkJoinPool而ForkJoinPool有一个工作窃取的概念 工作窃取当我去执行一个特别大的任务时感觉我用一个线程去执行的时候周期会很长那么我可以通过自己的业务去把他拆分成很多很多个小的业务比如我的任务需要5分钟那么我把它拆分成10个可能就半分种就搞定了然后再把这10个任务交给newWorkStealingPool线程池找一些没事干的线程去帮我干这个任务 4.3 newSingleThreadExecutor
这是一个使用单个worker线程的Executor下面是它的源码
可以看到它的核心线程数和最大线程数都是1其它参数都和FixedThreadPool相同既然它是单线程那么就可以保证任务的顺序执行。
4.4 newCachedThreadPool
这是一个会根据需要创建新线程的线程池也就是大小可变的线程池下面是它的源码
可以看到核心线程数为0最大线程数为Integer的最大值可以理解为无界CachedThreadPool使用没有容量的SynchronousQueue上面有提到过作为线程池的工作队列但最大线程数是无界的这就意味着如果主线程提交任务的速度高于CachedThreadPool线程处理任务的速度时CachedThreadPool会不断地创建新线程。
ps极端情况下Cached会因为创建过多的线程而耗尽CPU和内存资源。
4.5 newScheduledThreadPool
它是可以执行定时任务的线程池用ScheduledThreadPoolExecutor的构造函数来创建而ScheduledThreadPoolExecutor继承自ThreadPoolExecutor源码如下
可以看到核心线程数是传进来的最大线程数为Integer的最大值工作队列用的是DelayWorkQueue延迟队列、无界。
ScheduledThreadPoolExecutor的实现 ScheduledThreadPoolExecutor会把待调度的任务ScheduledFutureTask下面都简化为task放到一个DelayQueue中。 task主要包含3个变量 long time表示这个任务将要被执行的具体时间。 long sequenceNumber表示这个任务被添加到ScheduledThreadPoolExecutor中的序号。 long period表示任务执行的间隔周期。 DelayQueue封装了一个PriorityQueue这个PriorityQueue会对队列中的task进行排序。排序时time小的排在前面时间早的任务先执行如果两个time相同就比较sequenceNumber小的排在前面。 代码Demo
public class ScheduleDemo {public static void main(String[] args) {//创建ScheduledExecutorService线程池ScheduledExecutorService executor Executors.newScheduledThreadPool(5);//延迟5秒执行任务executor.schedule(() - {//要执行的任务}, 5, TimeUnit.SECONDS);
//定时执行任务延迟5秒开始每隔10秒执行一次executor.schedule(() - {//要执行的任务}, 5, 10, TimeUnit.SECONDS);
// 定时执行任务延迟 5 秒开始每隔10秒执行一次上一次任务执行完成后再延迟 10 秒executor.scheduleWithFixedDelay(() - {// 要执行的任务}, 5, 10, TimeUnit.SECONDS);}
}
4.6 为什么不推荐通过Executors构建线程池
上述五种线程池都可以通过Executors工具类来创建出来使用起来也很方便但是在阿里巴巴开发手册中也明确指出不允许使用Executors创建线程池并且指出了存在的弊端
还有一点就是当你通过Executors创建线程池的时候它创建的线程池大多都是已经帮你设置好了的参数很多选项是没办法自定义的在特殊的业务场景下Executors工具类下的线程池也并不是“万能的”。
正确的做法就是通过ThreadPoolExecutor的构造函数来自己定义线程池。
除此之外还可以通过一些开源类库比如apache和guava等笔者推荐使用guava提供的ThreadFactoryBuilder来创建线程池
public class ThreadFactoryDemo {public static void main(String[] args) {//线程工厂ThreadFactory factory new ThreadFactoryBuilder().setNameFormat(demo-pool-%d).build();
ThreadPoolExecutor executor new ThreadPoolExecutor(5, 10, 0L,TimeUnit.SECONDS, new LinkedBlockingQueue(2048),factory, new ThreadPoolExecutor.AbortPolicy());
for (int i 0; i 100; i) {executor.execute(() - {//业务代码});}}
}
上述方式可以自定义线程名称更方便出错时的溯源。
五、总结 线程池是一种用于管理和复用线程的机制。通过线程池我们可以创建一组可用的线程并且可以根据需要执行任务避免频繁地创建和销毁线程提高系统的性能和稳定性。使用ThreadPoolExcutor类可以自定义线程池也是创建线程池的正确做法通过构造函数的7个核心参数我们可以找到自己合适的线程池。任务提交有两种方式一种是execute()适用于没有返回值的情况ThreadPoolExecutor自己的方法第二种是submit()适用于需要返回值的情况ThreadPoolExecutor父类的方法。ThreadPoolExecutor类的三个核心方法execute()、addWorker()和runWorker()了解其源码才更有助于我们使用和调优线程池。Java中自带的5种线程池通过Executors工具类创建虽说用起来很方便但并不推荐使用此方法来创建线程池。 End希望对大家有所帮助如果有纰漏或者更好的想法请您一定不要吝啬你的赐教。
