devbox
小丑西瓜9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

线程池ThreadPoolExecutor详解（整理详细）_threadpooltaskexecutor

作者：小丑西瓜9 | 2024-02-18 04:24:29

赞

踩

threadpooltaskexecutor

1.什么是线程池？
（首先要理解什么是线程）
线程池，thread pool，是一种线程使用模式，线程池维护着多个线程，等待着监督管理者分配可并发执行的任务。

  • 通俗来说，就是为了优化线程的内存开销。

2.为什么使用线程池？
为了减少创建和销毁线程的次数，让每个线程都可以多次的使用，可以根据系统情况调整线程的数量，防止消耗过多内存。在实际使用中，服务器在创建和销毁线程上花费的时间和消耗的系统资源都相当大，使用线程池就可以优化。
线程池主要用来解决线程生命周期开销问题和资源不足问题。通过对多个任务重复使用线程，线程创建的开销就被分摊到了多个任务上了，而且由于在请求到达时线程已经存在，所以消除了线程创建所带来的延迟。这样，就可以立即为请求服务，使用应用程序响应更快。另外，通过适当的调整线程中的线程数目可以防止出现资源不足的情况。

通俗来说，就是为了优化线程的内存开销。
3.线程池的核心参数

  1. public ThreadPoolExecutor(int corePoolSize,//核心线程数
  2.                           int maximumPoolSize,//最大线程数
  3.                           long keepAliveTime,//线程空闲时间
  4.                           TimeUnit unit,//时间单位
  5.                           BlockingQueue<Runnable> workQueue,//任务队列
  6.                           ThreadFactory threadFactory,//线程工厂
  7.                           RejectedExecutionHandler handler//拒绝策略) 
  8. {
  9.     ...
  10. }

4.线程池的执行顺序
线程池按以下行为执行任务

  • 当线程数小于核心线程数时，创建线程。
  • 当线程数大于等于核心线程数，且任务队列未满时，将任务放入任务队列。
  • 当线程数大于等于核心线程数，且任务队列已满，若线程数小于最大线程数，创建线程。
  • 若线程数等于最大线程数，则执行拒绝策略

5.线程池的参数详解

  • corePoolSize

核心线程数，默认为1。
设置规则：
CPU密集型(CPU密集型也叫计算密集型，指的是运算较多，cpu占用高，读/写I/O(硬盘/内存)较少)：corePoolSize = CPU核数 + 1
IO密集型（与cpu密集型相反，系统运作，大部分的状况是CPU在等I/O (硬盘/内存) 的读/写操作，此时CPU Loading并不高。）：corePoolSize = CPU核数 * 2

  • maximumPoolSize

最大线程数，默认为Integer.MAX_VALUE
一般设置为和核心线程数一样

  • keepAliveTime

线程空闲时间，默认为60s，一般设置为默认60s

  • unit

时间单位，默认为秒

  • workQueue

队列，当线程数目超过核心线程数时用于保存任务的队列。（BlockingQueue workQueue）此队列仅保存实现Runnable接口的任务。（因为线程池的底层BlockingQueue的泛型为Runnable）
无界队列
队列大小无限制，常用的为无界的LinkedBlockingQueue，使用该队列作为阻塞队列时要尤其当心，当任务耗时较长时可能会导致大量新任务在队列中堆积最终导致OOM。阅读代码发现，Executors.newFixedThreadPool 采用就是 LinkedBlockingQueue，而博主踩到的就是这个坑，当QPS很高，发送数据很大，大量的任务被添加到这个无界LinkedBlockingQueue 中，导致cpu和内存飙升服务器挂掉。
当然这种队列，maximumPoolSize 的值也就无效了。当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
有界队列
当使用有限的 maximumPoolSizes 时，有界队列有助于防止资源耗尽，但是可能较难调整和控制。常用的有两类，一类是遵循FIFO原则的队列如ArrayBlockingQueue，另一类是优先级队列如PriorityBlockingQueue。PriorityBlockingQueue中的优先级由任务的Comparator决定。
使用有界队列时队列大小需和线程池大小互相配合，线程池较小有界队列较大时可减少内存消耗，降低cpu使用率和上下文切换，但是可能会限制系统吞吐量。
同步移交队列
如果不希望任务在队列中等待而是希望将任务直接移交给工作线程，可使用SynchronousQueue作为等待队列。SynchronousQueue不是一个真正的队列，而是一种线程之间移交的机制。要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接收这个元素。只有在使用无界线程池或者有饱和策略时才建议使用该队列。
原文：https://blog.csdn.net/riemann_/article/details/104704197

  • threadFactory

线程工厂，用来创建线程。
为了统一在创建线程时设置一些参数，如是否守护线程，线程一些特性等，如优先级。通过这个TreadFactory创建出来的线程能保证有相同的特性。
它是一个接口类，而且方法只有一个，就是创建一个线程。
如果没有另外说明，则在同一个ThreadGroup 中一律使用Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的NORM_PRIORITY 优先级和非守护进程状态。
通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。
如果从newThread 返回 null 时ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。

  • handler

拒绝策略，默认是AbortPolicy，会抛出异常。
当线程数已经达到maxPoolSize，且队列已满，会拒绝新任务。
当线程池被调用shutdown()后，会等待线程池里的任务执行完毕再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝新任务。
AbortPolicy 丢弃任务，抛运行时异常。
CallerRunsPolicy 由当前调用的任务线程执行任务。
DiscardPolicy 忽视，什么都不会发生。
DiscardOldestPolicy 从队列中踢出最先进入队列（最后一个执行）的任务。
6.ThreadPoolExecutor和spring封装的ThreadPoolTaskExecutor案例
ThreadPoolExecutor是Java的线程池
ThreadPoolTaskExecutor是spring封装的线程池

  1. package com.thgy.bc.common.config;
  2.  
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  7.  
  8. import java.util.concurrent.*;
  9.  
  10. @Slf4j
  11. @Configuration
  12. public class ThreadPoolConfig {
  13.     @Bean
  14.     public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
  15.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  16.         int i = Runtime.getRuntime().availableProcessors();
  17.         //核心线程数目
  18.         executor.setCorePoolSize(i * 2);
  19.         //指定最大线程数
  20.         executor.setMaxPoolSize(i * 2);
  21.         //队列中最大的数目
  22.         executor.setQueueCapacity(i * 2 * 10);
  23.         //线程名称前缀
  24.         executor.setThreadNamePrefix("ThreadPoolTaskExecutor-");
  25.         //rejection-policy：当pool已经达到max size的时候，如何处理新任务
  26.         //CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
  27.         //对拒绝task的处理策略
  28.         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  29.         //当调度器shutdown被调用时等待当前被调度的任务完成
  30.         executor.setWaitForTasksToCompleteOnShutdown(true);
  31.         //线程空闲后的最大存活时间
  32.         executor.setKeepAliveSeconds(60);
  33.         //加载
  34.         executor.initialize();
  35.         log.info("初始化线程池成功");
  36.         return executor;
  37.     }
  38.  
  39.     @Bean
  40.     public ThreadPoolExecutor threadPoolExecutor() {
  41.         //获取cpu核心数
  42.         int i = Runtime.getRuntime().availableProcessors();
  43.         //核心线程数
  44.         int corePoolSize = i * 2;
  45.         //最大线程数
  46.         int maximumPoolSize = i * 2;
  47.         //线程无引用存活时间
  48.         long keepAliveTime = 60;
  49.         //时间单位
  50.         TimeUnit unit = TimeUnit.SECONDS;
  51.         //任务队列，接收一个整型的参数，这个整型参数指的是队列的长度，
  52.         //ArrayBlockingQueue(int,boolean)，boolean类型的参数是作为可重入锁的参数进行初始化，默认false，另外初始化了notEmpty、notFull两个信号量。
  53.         BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue(i * 2 * 10);
  54.         //1. 同步阻塞队列 (put,take)，直接提交。直接提交策略表示线程池不对任务进行缓存。新进任务直接提交给线程池，当线程池中没有空闲线程时，创建一个新的线程处理此任务。
  55.         // 这种策略需要线程池具有无限增长的可能性。实现为：SynchronousQueue
  56.         //2. 有界队列。当线程池中线程达到corePoolSize时，新进任务被放在队列里排队等待处理。有界队列（如ArrayBlockingQueue）有助于防止资源耗尽，
  57.         // 但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，
  58.         // 但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，
  59.         // CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。
  60.         //3. 无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。
  61.         // 这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，
  62.         // 适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
  63.  
  64.         //线程工厂
  65.         //defaultThreadFactory()
  66.         //返回用于创建新线程的默认线程工厂。
  67.         //privilegedThreadFactory()
  68.         //返回一个用于创建与当前线程具有相同权限的新线程的线程工厂。
  69.         ThreadFactory threadFactory =Executors.defaultThreadFactory();
  70.         //拒绝执行处理器
  71.         RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
  72.         //创建线程池
  73.         ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
  74.         return threadPoolExecutor;
  75.     }
  76. }

7.jdk自带的四种线程池创建方式

  1. // 第一种线程池:固定个数的线程池,可以为每个CPU核绑定一定数量的线程数
  2. ExecutorService fixedThreadPool = Executors.newFixedThreadPool(processors * 2);
  3. // 缓存线程池，无上限
  4. ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
  5. // 单一线程池,永远会维护存在一条线程
  6. ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();
  7. // 固定个数的线程池，可以执行延时任务，也可以执行带有返回值的任务。
  8. ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);

8.队列的详解
参考：BlockingQueue（阻塞队列）详解 - aspirant - 博客园

9.调用线程池的线程案例

  1. public List<AccountRecordVO> requestTest() throws ExecutionException, InterruptedException {
  2.         List<String> ids = Lists.newArrayList();
  3.         ids.add("1");
  4.         ids.add("2");
  5.         ids.add("3");
  6.         ids.add("4");
  7.         //有返回值的情况，定义接收返回值
  8.         List<AccountRecordVO> futureList2 = Lists.newArrayList();
  9.         //分布式计数器，若业务不需要则可以不定义
  10.         CountDownLatch countDownLatch = new CountDownLatch(ids.size());
  11.         for (String id : ids) {
  12.             //调用线程池的线程执行任务
  13.             threadPoolTaskExecutor.submit(new Runnable() {
  14.                 @Override
  15.                 public void run() {
  16.                     test(Lists.newArrayList(id),futureList2);
  17.                     //计数器-1
  18.                     countDownLatch.countDown();
  19.                 }
  20.             });
  21.         }
  22.         //await阻塞，直到计数器为0
  23.         countDownLatch.await();
  24.         System.out.println("主线程"+"====");
  25.         return futureList2;
  26.     }
  27.  
  28.     public List<AccountRecordVO> test(List<String> ids, List<AccountRecordVO> list2){
  29.         //随便写的业务逻辑代码，无实际意义，仅作演示
  30.         System.out.println("线程体" + "====");
  31.         List<AccountRecordVO> accountRecordVOS = Lists.newArrayList();
  32.         int i = 0;
  33.         AccountRecordVO accountRecordVO = new AccountRecordVO();
  34.         accountRecordVO.setUserId("123");
  35.         accountRecordVO.setAmount(12333);
  36.         for (String id : ids){
  37.             accountRecordVOS.add(accountRecordVO);
  38.             list2.add(accountRecordVO);
  39.         }
  40.         try{
  41.             Thread.sleep(Long.valueOf("1000"));
  42.         }catch (Exception e){
  43.             log.error(e.getMessage());
  44.         }
  45.         System.out.println("线程体结束" + "====");
  46.         return accountRecordVOS;
  47.     }

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/小丑西瓜9/article/detail/104691
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号