Java线程池ThreadPoolExcutor

2020-12-29 本文已影响0人 NengLee

线程池的存在

Java中的线程池是对线程开销资源的管理，以及解决线程并发的一套框架。在开发过程中可以带来以下好处：

降低资源销毁，避免重复的创建线程销毁线程，带来的内存抖动
提高响应速度，线程池里面有维护的空闲线程，当收到任务直接run，从而提高效率
方便线程的管理，线程是稀缺资源，用一套框架进行统一分配，调优、监控

ThreadPoolExcutor

ThreadPoolExcutor结构派生图

Executor：是一个接口，将业务逻辑提交与任务执行进行分离
ExecutorService：接口继承了Execute，做了shutdowm()等业务的扩展，可以是说真正的线程池接口
AbstractExecutirService：抽象类实现了ExecutorService接口的大部分方法
ThreadPoolExecutor：线程池的核心实现类，用来执行被提交的任务

构造函数

/**
 *
 * @param corePoolSize  核心线程数
 *        
 * @param maximumPoolSize   最大线程数
 *
 * @param keepAliveTime  空闲线程存活时间
 *       
 * @param unit  @keepAliveTime 时间的单位
 *
 * @param workQueue  BlockingQueue<接口>阻塞队列
 *      
 * @param threadFactory   Thread的初始化设置
 *        
 * @param handler 4种拒绝策略
 *        
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

corePoolSize

线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize；如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；

maximumPoolSize

线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前线程数小于maximumPoolSize

keepAliveTime

线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间。默认情况下，该参数只在线程数大于corePoolSize时才有用

TimeUnit

keepAliveTime的时间单位

workQueue

必须是BlockingQueue阻塞队列。当线程池中的线程数超过它的corePoolSize的时候，线程会进入阻塞队列进行阻塞等待。通过workQueue，线程池实现了阻塞功能。

threadFactory

创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名，当然还可以更加自由的对线程做更多的设置，比如设置所有的线程为守护线程。

RejectedExecutionHandler

线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略。

AbortPolicy：直接抛出异常，默认策略
CallerRunsPolicy：用调用者所在的线程来执行任务
DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务
DiscardPolicy：直接丢弃任务

所以在构造函数线程池的时候，基本也就配置了该线程池的操作机制！

工作机制

假设几个重要参数：

corePoolSize=3，maximumPoolSize=8，ArrayBlockingQueue.capacity=10

通过execute(run) 或者是 submit(Callable)把任务给线程池执行，线程池一下子收到20条任务
会通过corePoolSize=3先执行前3条任务
然后从第4~13条任务存到消息队列中wokeQueue
这里假设核心线程数的还未执行完毕，还剩下5个线程Thread，分别执行14~19条任务。
剩下的任务，线程池已经无法工作，处于饱和无能为力状态，就会执行动态配置的饱和策略

提交

execute(Runnable run) ：提交一个抽象任务
submit(Callable call)：用于提交需要返回值的任务，线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法获取返回值，当然get()方法会阻塞当期线程直到任务完成，而使用get(long timeout,TimeUnit unit)方法则会阻塞当前线程一段时间后立即返回，这个时候可能任务没有执行完毕

关闭

shutdown 或者是 shutdownNow 方法来关闭线程池

关闭原理：遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止，但是它们存在一定的区别。

shutdown ：会中断没有正在执行的任务线程
shutdownNow ：则是将状态设置stop，然后尝试停止或暂停所有正在执行的线程，并返回等待执行的列表

队列

队列Queue数据集合，遵守先进先出的原则，在此基础也有优先级队列，可以通过time -- MeesageQueue优先级对列。关于队列介绍可以参考之前的文章：队列&栈

阻塞队列

支持阻塞的插入方法，当队列满的时候，队列会阻塞插入元素的线程，直到队列不满
支持阻塞的移除方法，当队列为空的时候，获取元素的线程会等待队列变为非空

在并发编程中，使用生产者和消费者模式能够解决大多数并发问题，该模式通过平衡生产线程和消费线程的工作能力，提高程序的整体处理数据的数速度。

生产者和消费者模式是通过一个容器来解决它们之间的耦合问题，生产者和消费者彼此之间不直接通讯，而是通过阻塞队列进行通讯，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

阻塞队列集合

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列。
LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列。
PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。
DelayQueue：一个使用优先级队列实现的无界阻塞队列。
SynchronousQueue：一个不存储元素的阻塞队列。
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

以上队列都是对BlockingQueue接口的实现，也都属于线程

格式：结构 + 有界/无界

有界指的是：该集合有需要有具体的长度以及大小

无界指的是：该集合可以放无法的对象而不会因为队列的长度限制被阻塞，当然也是属于系统空间载体限制。不然会导致内存超限，OOM的因素。