Java并发编程 (九) 线程调度-线程池

个人博客网：https://wushaopei.github.io/    (你想要这里多有)

声明：实际上，在开发中并不会普遍的使用Thread，因为它具有一些弊端，对并发性能的影响比较大，如下：

   new Thread 弊端：

         每次 new Thread 新建对象，性能差；

         线程缺乏统一管理，可能无限制的新建线程，相互竞争，有可能占用过多系统资源导致死机或OOM；

         缺少更多功能，如更多执行、定期执行、线程中断

一、 线程池-1

1、线程池的好处

  重用存在的线程，减少对象创建、消亡的开销，性能佳；

  可有效控制最大并发线程数，提高系统资源利用率，同时可以避免过多资源竞争，避免阻塞；

  提供定时执行、定期执行、单线程、并发数控制等功能

线程池相关的类：

2、第一种：线程池 - ThreadPoolExecutor

1）其主要参数有：

     corePoolSize :核心线程数量

     maximumPoolSize : 线程最大线程数

     workQueue: 阻塞队列，存储等待执行的任务，很重要，会对线程运行过程产生重大影响

2）具体说一下以上3个主要参数的关系：如果运行的线程数少于 corePooSize，直接创建新线程来处理任务，即使线程池中的其他线程是空闲的；如果线程池中的线程数量大于等于corePooSize，

且小于maximumPoolSize的时候，则只有当workQueue满了的时候，才会创建新的执行线程去处理任务。如果我们设置的corePooSize和maximumPoolSize的数量相同的话，那么由于线程池的大小是固定的，这时候如果有新任务提交，如果里面的workQueue还没满的时候，就会把请求放入到workQueue里面，等待有空闲的线程，去workQueue里面去取出来进行处理。

如果当前运行的线程数量大于等于maximumPoolSize时，如果此时workQueue也满了，那么就会通过拒绝策略 指定策略去处理任务。

所以，在任务提交时，它的顺序主要有三个，先判断是否小于corePooSize，如果小于它，就直接创建新线程来调用任务；然后再接着判断workQueue，最后再判断与maximumPoolSize的比较结果。

3）关于workQueue : 它是保存等待执行任务的阻塞队列。当提交一个新的任务到线程池以后，线程池会根据当前线程池中正在运行的线程数量来决定该任务的处理方式。

处理方式有三种：分别是直接切换，使用无限队列或使用有限队列。

1、直接切换这种方式的处理队列就是 Sync Queue；

2、使用无限队列，一般是使用基于链表的阻塞队列，如PriorityBlockingQueue这种方式，

线程池中能创建的最大线程数是corePooSize，而此时maximumPoolSize最大线程数就不会起作用了；当线程池中的所有核心线程状态都是运行状态的时候，这时一个新的线程提交后就会放入到等待队列里面去；

3、使用有限队列，workQueue 为有限队列时，一般使用的是ArrayBlockingQueue，这时候可以将线程池的最大值数量限制为maximumPoolSize，这样能够降低资源的消耗。但是，这种方式也使得线程池对线程的调度变得更困难。因为线程池核心线程和队列的容量都是有限的。

所以，要想使线程的处理效率和吞吐率达到一个相对合理的范围，使我们的线程调度相对简单，并且能够降低线程池对资源的消耗，就需要合理的设置这两个数量。

扩展连接地址：https://blog.csdn.net/xiaojin21cen/article/details/87363143

4）其他参数：

•    keepAliveTime : 线程没有任务执行时最多保持多久时间终止

           线程池维护线程所允许的空闲时间：当线程池中的线程数量大于corePooSize时，如果这时没有新的任务提交，核心线程外的线程不会立即销毁，而是会一直等待，直.到等待的时间超过了keepAliveTime.

•    unit : keepAliveTime的时间单位
•    threadFactory : 线程共产，用来创建线程

           线程池中默认会有一个默认的工厂用来创建线程；默认的工厂来创建线程时，会使新创建的线程会具有相同的优先级，并且是可以守护的线程。同时它也设置了线程的名称。

•   rejectHandler : 当拒绝处理任务时的策略

5）线程池对线程的具体处理策略：

如果线程池中的workQueue 满了，这时候还继续提交任务，我们就需要采取一种策略来处理这个任务。

线程池总共提供了四种策略：

       第一种是直接抛出异常，这也是默认的策略；默认是直接抛出异常；

       第二种是用调用者所在的线程来执行任务；

       第三种是丢弃队列中最靠前的任务并执行当前任务；

最后一种策略是直接丢弃这个任务。

6）线程池实例的创建分析：

常用的线程池创建如下：

ExecutorService exec = Executors.newCachedThreadPool();

7）该实例的底层实现是：

​

由源码可知，Executors创建线程池的底层实际是由 ThreadPoolExecutor的实例初始化返回的，再进入ThreadPoolExecutor底层看一看：

​

ThreadPoolExecutor的实例化包括但不限于上图中的构造方法的实现，当前图中的参数包含了所有可能需要用到的参数，这里对线程工厂和策略都做了参数传入。具体需要根据业务需求进行选择相应的构造器。

​

当我们初试化了一个线程池之后，它通常有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED集中状态。

RUNNING:能接收新提交的任务，并且也能阻塞队列中的任务；

SHUTDOWN:当一个线程处于shutdown状态时，不能接收、处理新提交的任务；但是可以处理阻塞队列中已经保存的任务；

在线程池处于Running状态时，调用shutdown（）方法就会进入到SHUTDOWN状态

STOP: stop 状态不接收新的任务，也不处理队列中的任务；它会中断正在处理任务的队列中的线程

线程池处理Running状态的任务时，如果调用了shutdownNow（）方法会使线程池进入到该状态；

TIDYING:如果所有任务都终止了，这时候的有效线程数为0，线程池就会进入到该状态；

TERMINATED:在TIDYING状态时调用terminated()方法就会进入到TERMINATED状态，默认的terminated()方法什么都不会做。只是会在调用terminated()方法让线程池进入到TERMINATED状态。

二、线程池-2

1、线程池常用的方法：

1） 提交任务的方法：

 execute () : 提交任务，交给线程池执行

 submit () : 提交任务，能够返回执行结果 execute + Future

2)  关闭线程池的方法：

shutdown ( ) : 关闭线程池，等待任务都执行完

shutdownNow () ： 关闭线程池，不等待任务执行完

3） 使用与监控的方法：

   getTaskCount ( ): 线程池已执行和未执行的任务总数

   get CompletedTaskCount ( ) :  已完成的任务数量

   getPoolSize ( ) : 线程池当前的线程数量

   getActiveCount ( ) : 当前线程池中正在执行任务的线程数量

4） ThreadPoolExecutor常见方法 总览：

​

2、线程池的类图：

​

Executors是根据一组执行策略执行调度调用异步任务的框架，目的是将任务提交与任务运行分离开来处理。

JUC里有三个Executor接口，分别是Executor、ExecutorService、ScheduledExecutorService

      Executor是一个运行新任务的简单接口；

      ExecutorService扩展了Executor接口，添加了用来管理执行器、生命周期和任务声明周期的方法；

      ScheduledExecutorService扩展了ExecutorService接口，支持Future和定期执行任务。

3、通过Executors提供的四种创建线程池的方法：

  第一种：   Executors.newCachedThreadPool

它可以创建一个可缓存的线程池。如果线程池的长度超过了处理的需要，可以灵活回收空闲线程，如果没有可回收的，就新建线程

第二种 : Executors.newFixedThreadPool

它创建的是一个定长的线程池，可以创建线程池的最大线程数，超出的任务会在队列中等待

第三种，Executors.newScheduledThreadPool

它创建的也是一个定长的线程池，它支持定时以及周期性的任务执行；

第四种，Executors.newSingleThreadExecutor

它是一个单线程化的线程池，它会用唯一的一个工作线程来执行任务，保证所有任务按照指定顺序去执行。该顺序可以按照指定先入先出、优先级等等来设定

三、线程池-3

1、分析四种线程池的创建的底层实现：

​

由上图中可知，四种线程其实都是有Executors来调用相应构造器实现创建的

​

由上图可知，newCachedThreadPool()创建可缓存的线程池底层实际还是调用ThreadPoolExecutor类的构造器返回的实例对象实现的。

​

newFixedThreadPool（）创建定长线程池的底层也是由ThreadPoolExecutor类的构造器返回的实例对象实现的。

​

​

由上述两个图可知，newScheduledThreadPool（）创建定长且可定时的线程池的底层是由ScheduledThreadPoolExecutor类的构造器创建实例实现的；而ScheduledThreadPoolExecutor又继承了ThreadPoolExecutor类

​

newSingleThreadExecutor（）创建单线程线程池的底层也是由ThreadPoolExecutor类的构造器返回的实例对象实现的。

2、代码实例演示线程池的实现：

1）newCachedThreadPool()实现可缓存线程池：

@Slf4j
public class ThreadPoolExample1 {

    public static void main(String[] args) {

        ExecutorService e = Executors.newCachedThreadPool();

        for (int i = 0 ; i < 10 ; i++){
            final int index = i;
            e.execute(new Runnable() {
                @Override
                public void run() {
                    log.info("task:{}" ,index);
                }
            });
        }
        e.shutdown();
    }
}

执行并打印结果：

22:59:26.711 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:1
22:59:26.711 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:2
22:59:26.711 [pool-1-thread-6] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:5
22:59:26.711 [pool-1-thread-9] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:8
22:59:26.711 [pool-1-thread-8] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:7
22:59:26.711 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:0
22:59:26.711 [pool-1-thread-7] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:6
22:59:26.711 [pool-1-thread-10] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:9
22:59:26.711 [pool-1-thread-4] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:3
22:59:26.711 [pool-1-thread-5] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample1 - task:4

Process finished with exit code 0

2）newFixedThreadPool()实现定长线程池：

@Slf4j
public class ThreadPoolExample2 {

    public static void main(String[] args) {

        ExecutorService e = Executors.newFixedThreadPool(3);

        for (int i = 0 ; i < 10 ; i++){
            final int index = i;
            e.execute(new Runnable() {
                @Override
                public void run() {
                    log.info("task:{}" ,index);
                }
            });
        }
        e.shutdown();
    }
}

执行并打印结果：

23:02:02.958 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:1
23:02:02.958 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:0
23:02:02.958 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:2
23:02:02.964 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:3
23:02:02.964 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:5
23:02:02.964 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:4
23:02:02.964 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:6
23:02:02.964 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:8
23:02:02.964 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:7
23:02:02.964 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample2 - task:9

Process finished with exit code 0

3）newSingleThreadExecutor（）创建单线程线程池：

ExecutorService e = Executors.newSingleThreadExecutor();

执行并打印结果：

23:04:33.663 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:0
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:1
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:2
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:3
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:4
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:5
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:6
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:7
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:8
23:04:33.670 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample3 - task:9

Process finished with exit code 0

由结果可知，单线程线程池中，线程任务的执行是有序的。

4）newScheduledThreadPool（）创建定长的线程池：

ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);

执行并打印结果：

23:07:51.452 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:1
23:07:51.452 [pool-1-thread-2] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:2
23:07:51.452 [pool-1-thread-4] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:4
23:07:51.452 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:0
23:07:51.459 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:6
23:07:51.459 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:7
23:07:51.459 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:8
23:07:51.459 [pool-1-thread-1] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:9
23:07:51.452 [pool-1-thread-5] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:3
23:07:51.459 [pool-1-thread-3] INFO com.mmall.concurrency.example.threadPool.ThreadPoolExample4 - task:5

5）newScheduledThreadPool（）创建定长且可定时的线程池：

单次执行定时任务：

executorService.schedule(new Runnable() {
    @Override
    public void run() {
        log.warn("schedule run");
    }
},3, TimeUnit.SECONDS);  //定时执行，3秒后执行任务

执行并打印结果：

23:15:13.526 [pool-1-thread-1] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run

延迟多次执行定时任务：

executorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        log.warn("schedule run");
    }
},1,3, TimeUnit.SECONDS);  // 以指定的延迟去执行任务，每隔3秒执行一次任务

执行并打印结果：

23:16:22.111 [pool-1-thread-1] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:25.110 [pool-1-thread-1] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:28.110 [pool-1-thread-2] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:31.109 [pool-1-thread-1] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:34.109 [pool-1-thread-3] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:37.109 [pool-1-thread-3] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run
23:16:40.109 [pool-1-thread-3] WARN com.mmall.concurrency.example.threadPool.ThreadPoolExample5 - schedule run

3、线程池 - 合理配置

•   CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 NCPU + 1
•  IO密集型任务，参考值可以设置为2*NCPU