04. 线程池

1. 线程池的含义

  1. /*
  2.    * 什么是线程池
  3.    *   线程池其实就是一种多线程处理像是,处理过程中可以将任务添加到队列中,然后再创建线程后自动启用这些任务.
  4.    * 这里的线程就是之前学过的线程,这里的任务就是实现了Runnable和callable接口的实例对象*/
  5.   /*
  6.    * 为什么要是使用线程池
  7.    *   使用线程池最大的原因就是可以根据系统的需求和硬件环境灵活的控制线程的数量,
  8.    * 且库对所有的线程进行统一的管理和控制,从而提高系统的运行效率,降低系统运行的压力,当然了,使用线程池的原因不仅仅只有这些,
  9.    * 我们可以从线程池自身的优点上来进一步了解下线程池的好处*/
  10.   /*
  11.    * 使用线程池的好处
  12.    *   1.线程和任务分开,提高线程的重用性
  13.    *   2.控制线程并发数量,降低服务器压力,统一管理所有线程
  14.    *   3.提升系统响应速度,假如创建线程用的时间为T1,执行任务用的时间为T2,
  15.    *     销毁线程用的时间为T3,那么使用线程就免去了T1和T3的时间*/
  16.   /*
  17.    * 线程池的应用场景
  18.    * 1.网购商品秒杀
  19.    * 2.云盘文件上传和下载
  20.    * 3.12306网上购票系统等
  21.    * 总之:只要有并发的地方,任务数量大或小,每个任务执行的时间长与短都可以使用线程池;
  22.    * 只不过在使用线程池的时候,注意一下设置合理的线程池大小即可 */
  23.   /*
  24.    * 通过观察java中内置的线程池参数讲解和线程池工作流程总结,我们不难发现,要设计一个号的线程池,
  25.    * 就必须合理的设置线程池的4个参数,那么到底该如何合理的设计4个参数的值
  26.    * 1.核心线程数(corePoolSize)
  27.    *   核心线程数的设计需要依据任务的处理时间和每秒产生的任务数量来确定,一般按照8020的原则设计
  28.    * ,即按照百分之80的情况设置核心线程数,声息的百分之20可以利用最大线程数处理
  29.    * 2.任务队列长度(workQueue)
  30.    *   任务队列长度一般设计为:(核心线程数/单个执行时间)*2
  31.    * 3.最大线程数(maximumPoolSize)
  32.    *   最大线程数的设计除了需要参照核心线程数的条件外,还需要参照每秒产生的最大任务数决定
  33.    * 例如:如果系统每秒产生的任务是1000个,那么最大线程数 = (最大任务数 - 任务队列长度) * 单个任务执行时间
  34.    * 即:最大线程数 = (1000 -200)*0.1 = 80个
  35.    * 4.最大空闲时间(keepAliveTime)
  36.    *  这个参数的设计完全参考系统运行环境和硬件压力设定,没有固定的参考值,开发者可以根据经验和系统产生任务的时间
  37.    * 间隔来合理设置一个值即可
  38.    * 注意:上面4个参数的设置只是一般的设计原则,并不是固定的,开发者可以根据实际情况灵活调整*/

2. java内置线程池-ExecutorService

2.1 newCachedThreadPool : 创建一个默认的线程池对象

  1.     作用是:创建一个默认的线程池对象,里面的线程可以重用,且在第一次使用时才创建
  1. public static void main(String[] args) {
  2.     /*
  3.      * java内置线程池-ExecutorService
  4.      * ExecutorService接口是java内置的线程池接口
  5.      * 常用方法:
  6.      * void     shutdown() 启用一次顺序关闭,执行以前提交的任务,但不接受新任务
  7.      * List<Runnable> shutdownNow() 停止所有正在执行的任务,暂停处理正在等待的任务,并返回等待执行的任务列表
  8.      * <T>Future<T> submit(Callable<T> task) 执行带返回值的任务,返回一个 Future对象
  9.      * */
  10.     //    fn1();
  11.     fn2();
  12.   }
  14.   public static void fn1() {
  15.     // 1.创建一个默认的线程池对象,里面的线程可以重用,且在第一次使用时才创建
  16.     ExecutorService es = Executors.newCachedThreadPool();
  17.     for (int i = 0; i < 20; i++) {
  18.       es.submit(new ThreadFn(i));
  19.     }
  20.     es.shutdown();
  21.   }
  23.   public static void fn2() {
  24.     // 1.使用工厂类获取线程池对象
  25.     ExecutorService es =
  26.         Executors.newCachedThreadPool(
  27.             new ThreadFactory() {
  28.               int n = 1;
  30.               @Override
  31.               public Thread newThread(Runnable r) {
  32.                 return new Thread(r, "自定义线程名称" + n++);
  33.               }
  34.             });
  35.     // 2.提交任务
  36.     for (int i = 0; i < 20; i++) {
  37.       es.submit(new ThreadFn(i));
  38.     }
  39.     es.shutdown();
  40.   }
  41. }

2.2 newFixedThreadPool:创建一个可重用固定线程数的线程池

  1. public class PoolTest2 {
  2.   public static void main(String[] args) {
  3.     // 创建一个可重用固定线程数的线程池
  4.     //    fn1();
  5.     fn2();
  6.   }
  8.   public static void fn1() {
  9.     ExecutorService es = Executors.newFixedThreadPool(5);
  10.     for (int i = 0; i < 30; i++) {
  11.       es.submit(new ThreadFn(i));
  12.     }
  13.     es.shutdown();
  14.   }
  16.   public static void fn2() {
  17.     ExecutorService es =
  18.         Executors.newFixedThreadPool(
  19.             5,
  20.             new ThreadFactory() {
  21.               int i = 0;
  23.               @Override
  24.               public Thread newThread(Runnable r) {
  25.                 return new Thread(r, "自定义名称" + i++);
  26.               }
  27.             });
  28.     for (int i = 0; i < 20; i++) {
  29.       es.submit(new ThreadFn(i));
  30.     }
  31.     es.shutdown();
  32.   }
  33. }

2.3 newSingleThreadExecutor: 创建一个使用单个worker线程的Executor

  1. public class PoolTest3 {
  2.   public static void main(String[] args) {
  3.     // 创建一个使用单个worker线程的Executor,以无界队列的方式来运行该程序
  4.     //    fn1();
  5.     fn2();
  6.   }
  8.   public static void fn1() {
  9.     ExecutorService es = Executors.newSingleThreadExecutor();
  10.     for (int i = 0; i < 10; i++) {
  11.       es.submit(new ThreadFn(i));
  12.     }
  13.     es.shutdown();
  14.   }
  16.   public static void fn2() {
  17.     ExecutorService es =
  18.         Executors.newSingleThreadExecutor(
  19.             new ThreadFactory() {
  20.               int i = 0;
  22.               @Override
  23.               public Thread newThread(Runnable r) {
  24.                 return new Thread(r, "自定义线程名" + i++);
  25.               }
  26.             });
  27.     for (int i = 0; i < 10; i++) {
  28.       es.submit(new ThreadFn(i));
  29.     }
  30.     //    es.shutdown();
  31.     // 测试shutdown和shutdownNow的区别
  32.     // shutdownNow会将未完成的任务用集合返回出来
  33.     List<Runnable> rs = es.shutdownNow();
  34.     System.out.println(rs);
  35.   }
  36. }

测试的Run方法

  1. public class ThreadFn implements Runnable {
  2.   private int id;
  4.   public ThreadFn(int id) {
  5.     this.id = id;
  6.   }
  8.   @Override
  9.   public void run() {
  10.     System.out.println(Thread.currentThread().getName() + "执行了任务" + id);
  11.   }

3. ScheduleExecutorService是ExecutorService的子接口

3.1 newScheduledThreadPool: 创建一个可重用固定线程数的线程池且允许延迟运行或定期执行任务

  1. public class PoolTest1 {
  2.   // ScheduleExecutorService是ExecutorService的子接口,具备了延迟运行或者定期执行任务的能力
  3.   // 创建一个可重用固定线程数的线程池且允许延迟运行或定期执行任务
  4.   public static void main(String[] args) {
  5.     //    fn1();
  6.     fn2();
  7.   }
  9.   public static void fn1() {
  10.     ScheduledExecutorService ss = Executors.newScheduledThreadPool(5);
  11.     for (int i = 0; i < 10; i++) {
  12.       // 给这个线程任务延时,定时2秒后执行,不是每个都延时了2秒,而是全部在2秒后开始运行
  13.       ss.schedule(new ThreadFn(i), 2, TimeUnit.SECONDS);
  14.     }
  15.     ss.shutdown();
  16.   }
  18.   public static void fn2() {
  19.     ScheduledExecutorService ss =
  20.         Executors.newScheduledThreadPool(
  21.             3,
  22.             new ThreadFactory() {
  23.               int i = 0;
  25.               @Override
  26.               public Thread newThread(Runnable r) {
  27.                 return new Thread(r, "自定义线程名" + i++);
  28.               }
  29.             });
  30.     for (int i = 0; i < 5; i++) {
  31.       // scheduleAtFixedRate,有4个参数 : 该方法意味着无限定期的执行
  32.       // 1.为线程任务 2.开始延时时间 3.多长时期执行一次 4.时间单位
  33.       ss.scheduleAtFixedRate(new ThreadFn(i), 2, 5, TimeUnit.SECONDS);
  34.     }
  35.     //    ss.shutdown();
  36.   }
  37. }

3.2 newSingleThreadScheduledExecutor : 创建一个单线程执行程序,它允许在给定的延迟后运行命令或者定期的执行

  1. public class PoolTest2 {
  2.   // 创建一个单线程执行程序,它允许在给定的延迟后运行命令或者定期的执行
  3.   // newSingleThreadScheduledExecutor()
  4.   public static void main(String[] args) {
  5.     //    fn1();
  6.     fn2();
  7.   }
  9.   public static void fn1() {
  10.     ScheduledExecutorService se = Executors.newSingleThreadScheduledExecutor();
  11.     for (int i = 0; i < 5; i++) {
  12.       se.schedule(new ThreadFn(i), 2, TimeUnit.SECONDS);
  13.     }
  14.     se.shutdown();
  15.   }
  17.   public static void fn2() {
  18.     ScheduledExecutorService se =
  19.         Executors.newSingleThreadScheduledExecutor(
  20.             new ThreadFactory() {
  21.               int i = 0;
  23.               @Override
  24.               public Thread newThread(Runnable r) {
  25.                 return new Thread(r, "自定义线程名" + i++);
  26.               }
  27.             });
  28.     for (int i = 0; i < 5; i++) {
  29.       se.scheduleAtFixedRate(new ThreadFn(i), 2, 5, TimeUnit.SECONDS);
  30.     }
  31.     //    se.shutdown();
  32.   }
  33. }

测试用的Run方法

  1. public class ThreadFn implements Runnable {
  2.   private int id;
  4.   public ThreadFn(int id) {
  5.     this.id = id;
  6.   }
  8.   @Override
  9.   public void run() {
  10.     System.out.println(Thread.currentThread().getName() + "执行了任务" + id);
  11.   }
  12. }

4. 线程池的拒绝策略

  1. public class Pool1 {
  2.   /*ThreadPoolExecutor 构造方法的最后一个参数指定了拒绝策略.当提交给线程池的任务量超过实际承载能力时,如何处理?
  3.   即线程池中 的线程已经用完了,等待队列也满了,无法为新提交的任务服务,
  4.   可以 通过拒绝策略来处理这个问题. JDK提供了四种拒绝策略:
  5.        AbortPolicy 策略,会抛出异常
  6.        CallerRunsPolicy 策略,只要线程池没关闭,会在调用者线程中运行当前被丢弃的任务
  7.        DiscardOldestPolicy 将任务队列中最老的任务丢弃,尝试再次提交新任务
  8.        DiscardPolicy 直接丢弃这个无法处理的任务
  9.        Executors 工具类提供的静态方法返回的线程池默认的拒绝策略是AbortPolicy 抛出异常,
  10.        如果内置的拒绝策略无法满足实际需求,可以扩展RejectedExecutionHandler 接口*/
  11.   public static void main(String[] args) {
  12.     NewPool n = new NewPool(5, 5, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10));
  13.     ThreadPoolExecutor newPool = n.getNewPool();
  14.     for (int i = 0; i < 100; i++) {
  15.       newPool.submit(new TestRun());
  16.     }
  17.     newPool.shutdown();
  18.   }
  19. }

自定义线程池

  1. public class NewPool {
  2.   private ThreadPoolExecutor te;
  4.   public NewPool(
  5.       int corePoolSize,
  6.       int maximumPoolSize,
  7.       long keepAliveTime,
  8.       TimeUnit unit,
  9.       BlockingQueue<Runnable> workQueue) {
  10.     this.te =
  11.         new ThreadPoolExecutor(
  12.             corePoolSize,
  13.             maximumPoolSize,
  14.             keepAliveTime,
  15.             unit,
  16.             workQueue,
  17.             Executors.defaultThreadFactory(),
  18.             new RejectedExecutionHandler() {
  19.               @Override
  20.               public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
  21.                 // r就是请求的任务, executor就是当前线程池
  22.                 System.out.println(r + "执行拒绝策略");
  23.               }
  24.             });
  25.   }
  27.   public ThreadPoolExecutor getNewPool() {
  28.     return te;
  29.   }
  30. }

测试Run方法

  1. public class TestRun implements Runnable {
  2.   // 定义任务
  3.   @Override
  4.   public void run() {
  5.     int num = new Random().nextInt(5);
  6.     System.out.println(
  7.         Thread.currentThread().getName() + "--" + System.currentTimeMillis() + "开始睡眠" + num + "秒");
  8.     try {
  9.       TimeUnit.SECONDS.sleep(num);
  10.     } catch (InterruptedException e) {
  11.       e.printStackTrace();
  12.     }
  13.   }
  14. }