多线程与高并发五:等待队列及Executor和线程池详解(重点)

1:等待队列

1.1:队列介绍

1. 关于创建线程池的好处这里就不介绍了,直接上干货.首先,要想弄明白线程池,先要认识一些队列,这是线程池的基础.
2. 多线程容器,以后多考虑Queue,少考虑List，Set.
3. 这里来一道面试题:Queue相比较list比较好的地方? 有对多线程比较友好的接口,有以下方法:
   offer:----相当于add，会有返回值，成功了返回ture
   poll:取数据并且remove掉
   peek：取数据不remove掉
4. BlockingQueue在线程池里是经常出现的队列：它是阻塞队列.多了下面的阻塞方法.
   put()，往里面装，满了会阻塞
   take(),往外面取,空了会阻塞.

   1.2:队列的分类

BlocKingQueue有以下几种:

A: LinkedBlockingQueue

链表实现,等待队列长度最大为 Integer.Max_Value**

B:ArrayBlockingQueue

可以设置初始队列大小**

C:DelayQueue

按照等待的时间排序，按时间进行任务调 度,本质上是PriorityQueue .实现类要实现Delay接口，传入等待时间，隔多长时间运行

D: PriorityQueue

继承AbstractQueue,二叉树的模型，对添加的元素排序，默认是升序

E: SynchronizedQueue

容量为0，不可以往里面扔东西add，前面等着拿东西的时候才可以装put,传递东西**

F:TransferQueue

里面有transfer方法，装完东西，阻塞，等着东西被取走，才离开

1.2：常见的拒绝策略

A:new ThreadPoolExecutor.AbortPolicy()
AbortPolicy 总是抛出异常,无特殊使用场景，默认就是这个拒绝策略。对于一些比较重要的业务，可以使用该拒 绝策略，方便出错的时候即时发现错误原因
B:new ThreadPoolExecutor.CallerRunsPolicy()
CallerRunsPolicy 将任务丢给启动线程池的线程去执行。适用于不太重要的业务场景，不抛出错误，比如博客的阅读量这种的
C:new ThreadPoolExecutor.DiscardOldestPolicy()
DiscardOldestPolicy 让最早进入阻塞队列的离开，然后自己进去排队。将最早进入阻塞队列的丢弃，典型的喜新厌旧，看你是不是对于老的任务需要。
D:new ThreadPoolExecutor.DiscardPolicy()
DiscardPolicy 直接丢弃任务。将任务丢给线程池本身的线程去运行，一般在不允许失败的、对性能要求不高、并发量较小的场景下使用。不然的话，容易降低性能

2:Executor

2.1:Runnable接⼝和Callable接口

6. 我们知道,线程里面是要开启一个任务的,让线程去工作.一般实现一个任务有两种方式:实现Runnable接⼝和Callable接口.
   来一道面试题:实现Runnable接口和Callable接口的区别?
   Runnable 接口在Java 1.0以来一直存在，但Callable 仅在Java 1.5中引入,目的就是为了来处理Runnable没有返回值的现象或抛出检查异常.这个返回值一般用Future对象来接收,Future.get();阻塞方法，直到有结果才会返回,停止阻塞

2.2:FutureTask

7. 除了Future,还有FutureTask:相当于Future+Runnable,既是一个任务，又是一个Future，执行完的结果自己也可以接收。同时，又是一个Runnable，可以new一个线程或者线程池来执行。

2.3关于Executor接口:

. 1：Executor接口 : 执行者,一个任务的定义和运行可以分开了 .只有一个方法executr()，让线程去执行。
2：ExecutorService接口:继承Executor接口，完善了任务执行器的生命周期.还有submit(Callable或者Runnable)方法，异步的提交任务，返回值是一个Future
3：CompletableFuture:运行一个任务，会有返回值，类型是double，自己可以接收。allOf():对一堆任务进行管理，当所有的任务都结束的时候，才会继续往下执行。管理多个Future的结果

3:线程池(ThreadPoolExecutor )

3.1: 介绍

ThreadPoolExecutor 继承于ExecutorService 继承Executor,线程池维护两个集合：一个是线程的集合（HashSet），一个是任务的集合

3.2. 自定义线程池:七个参数(面试重点)

//自定义线程,但是还不够具体,没有自定义拒绝策略
ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4,
        60, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(4),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.CallerRunsPolicy());

七个参数:
A:corePoolSize:核心线程数，线程池中一开始存在的线程数,核心线程永远活着(可以通过参数控制是否关闭核心线程,默认不关闭)
B:maximumPoolSize:最大线程数
C:keepAliveTime:线程空闲的时间，超过这个时间，线程资源归还给操作系统
D:TimeUnit:生存时间的单位
E:workQueue:线程队列，blockIngQueue–> LinkedBlockingQueue,ArrayBlockingQueue等
F:ThreadFactory:线程工厂,创建新线程，可以按照自己的方式去指定线程名称，守护线程
G:Handler:拒绝策略(饱和策略):线程池忙，等待队列满了，默认是4种，可以自定义,一般都是自定义,可以设置线程名程,防止出错的时候进行跟踪,另外就是可以在自定义中保存线程的一些内容和状态等.

3.3:线程池的默认实现

通过Executors(线程池的工厂),可以实现四种线程池,但是《阿里巴巴Java开发手册》中强制线程池不允许使用 Executors 去创建,下面结合源码去分析为什么.

A:SingleThreadPool

只有一个线程的线程池：可以保证扔进去的任务顺序执行**

//创建
 ExecutorService service = Executors.newSingleThreadExecutor();

下面是源码:

public static ExecutorService newSingleThreadExecutor() {
    
      return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
    //这里等待队列为LinkedBlockingQueue,会造成Integer.Max_Value
    //线程在等待,堆积大量的请求,这样可能会造成资源耗尽,从而导致OOM
                             new LinkedBlockingQueue<Runnable>()));
       }

B:CachedThreadPool

线程数目不定的线程池

//创建
ExecutorService service = Executors.newCachedThreadPool();

下面是源码:

  public static ExecutorService newCachedThreadPool() {
    
  // //最大线程数为Integer.MAX_VALUE,可能会创建大量线程，从而导致OOM
   return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                              60L, TimeUnit.SECONDS,
                            new SynchronousQueue<Runnable>());
}

C:FixedThreadPool

固定数量的线程池,线程并行

//创建
ExecutorService service = Executors.newFixedThreadPool(10)；

下面是源码:

public static ExecutorService newFixedThreadPool(int nThreads) {
    
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
        //这里等待队列为LinkedBlockingQueue,会造成Integer.Max_Value
    //线程在等待,堆积大量的请求,这样可能会造成资源耗尽,从而导致OOM
                                  new LinkedBlockingQueue<Runnable>());
  } 

D:ScheduledThreadPool：

定时任务线程池 有三个参数
下面是源码:

public ScheduledThreadPoolExecutor(int corePoolSize) {
    
最大线程数为Integer.MAX_VALUE,可能会创建大量线程，从而导致OOM
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

4:其他线程池

4.1:WorkStealingPool

线程池中每一个线程有自己单独的任务队列,一个线程执行完自己的任务之后,会去其他的线程拿任务运行.本质上是一个ForkJoinPool,只不过提供了更方便使用的接口

4.2 :ForkJoinPool

把大任务分割成很多小任务运行的线程池.还可以把小任务切分成更小的任务.如果需要把任务进行汇总,子任务汇总到父任务,父任务汇总到跟任务. 需要定义成能分叉的任务 ForkJoinTask,一般使用RecursiveTask(有返回值的任务),RecursiveAction(没有返回值的任务),二者都继承于ForkJoinTask**

4.3:面试题

这一篇就写到这里,学习了多线程之后,来一个面试题目:假如提供了一个闹钟服务,订阅这个服务的人特别多,10亿人,该怎么优化?
思路:把订阅任务分发到其他的边缘服务器上,在每一台服务器上用线程池+服务队列

注意：本文仅代表菜鸟博主的个人观点，如果哪里不对或者路过技术大大有更好的想法，欢迎留言告知，分享和交流使我们进步，谢谢。