ThreadPoolExecutor 线程池

2024-04-11 23:23:01

线程池的好处

实现了任务提交和任务执行的解耦, 用户只需提供Runnable对象, 而任务的执行和调度都由线程池负责

降低资源消耗
提升系统响应速度
提高线程的可管理性

工作流程

执行过程图

首先判断线程池是否已满, 未满->创建新的核心线程执行任务
已满->判断阻塞队列是否已满, 未满->加入阻塞队列
已满->线程池是否已满(自旋?), 未满->创建新线程执行任务
已满->按照饱和策略 handler处理

线程的创建参数讲解

ThreadPoolExecutor 的构造方法

ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler){}

corePoolSize：表示核心线程池的大小
如果当前线程池中线程个数小于它, 新任务到来时就会创建新的线程(即使当前核心线程池有空闲的线程)
maximumPoolSize：表示线程池能创建线程的最大个数
如果当阻塞队列已满时，假如线程池线程个数没有超过 maximumPoolSize 的话，就会创建新的线程来执行任务
keepAliveTime：空闲线程存活时间

超过corePoolSize的空闲线程, 空闲时间如果超过该值就会被销毁
unit: 时间单位
workQueue：阻塞队列, 用于保存任务的阻塞队列
可以使用ArrayBlockingQueue, LinkedBlockingQueue, SynchronousQueue, PriorityBlockingQueue
threadFactory：创建线程的工厂类
handler: 饱和策略, 如果线程和线程池阻塞队列都已饱和, 就需要一种策略来处理新来的线程
- AbortPolicy：直接拒绝所提交的任务，并抛出RejectedExecution-Exception异常；
- CallerRunsPolicy：直接用调用者所在的线程来执行任务；
  也就是说, 让处理请求的线程来执行任务, 这会降低新任务的提交速度,影响整体性能
  对于可伸缩的应用, 建议使用该策略, 因为最大池填满之后, 这个策略会提供一个可伸缩队列, 保存任务
- DiscardPolicy：不处理直接丢弃掉任务；
- DiscardOldestPolicy：丢弃掉阻塞队列中存放时间最久的任务，添加当前任务

线程池参数 - 如何设计线程池

主要从任务的性质角度考虑: CPU 密集型任务，IO 密集型任务和混合型任务

任务执行的时间: 长、中、短
任务的优先级:
任务的依赖性:

不同类型的任务

CPU 密集型任务配置尽可能少的线程数量, 可以配置NCPU+1个线程
IO 密集型任务则由于需要等待 IO 操作，线程并不是一直在执行任务，则配置尽可能多的线程，如2xNCPU
混合型的任务，可以将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务

只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐率要高于串行执行的吞吐率
如果这两个任务执行时间相差太大，则没必要进行分解
优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理。它可以让优先级高的任务先得到执行
如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行, 饿死
执行时间不同的任务可以交给不同规模的线程池来处理，或者也可以使用优先级队列，让执行时间短的任务先执行
依赖数据库连接池的任务，因为线程提交 SQL 后需要等待数据库返回结果，等待的时间越长 CPU 空闲时间就越长
所以数据连接池, 线程数应该设置尽量大，才能更好的利用 CPU
阻塞队列最好是使用有界队列，如果采用*队列的话，一旦任务积压在阻塞队列中的话就会占用过多的内存资源，甚至会使得系统崩溃

ThreadPoolExecutor 的数据结构

使用整型状态数 ctl - 线程池的状态

通过位切分同时表示线程池状态和活动线程数量, 这样也更容易保证线程安全, 状态位在高3位，活动线程数在低29位, 正常的增减操作就可以控制线程数量

COUNT_BITS 为 29

int RUNNING    = -1 << COUNT_BITS;
int SHUTDOWN   =  0 << COUNT_BITS;
int STOP       =  1 << COUNT_BITS;
int TIDYING    =  2 << COUNT_BITS;
int TERMINATED =  3 << COUNT_BITS;

running - 111
能够接受新任务
shutdown - shutdown() - 000
不接受新任务, 可以处理已加入的任务
stop - shutdownNow() - 001
不接受新任务, 不处理已添加的任务, 中断正在执行的任务
如何中断? 在安全点中断
tidying - 010
所有的任务都已终止, ctl记录为0
可以定义钩子函数 terminated()
terminated - 011
线程池彻底终止后转为的彻底终止状态

worker类 - 可以说是一把封装了线程和任务的锁

继承了AQS, mainLock&termination 非公平锁, 成员 thread 通过 getThreadFactory().newThread(this) 实例化

成员 firstTask 在实例化时由线程池提供
成员 completedTasks 表明完成任务的数量
HashSet:workers 是线程池中所拥有的线程集合

如何实现线程复用

这个worker有点特殊, 通过工厂产生的线程并不是任务, 而是自身this
那这个worker将如何工作呢? 它将以线程的形式工作, 也就是 run() 方法, run方法会调用外部的方法 runWorker(),
这才是任务被执行的核心, 在runWorker()中, while (task != null || (task = getTask()) != null) 就调用 task.run() 方法来执行任务.

那么它为什么是个锁? 为什么需要加锁?

worker 继承了 AbstractQueuedSynchronizer, 根据其重写的 tryAcquire 和 tryRelease, 可以看出这是一把非公平、独占的非重入锁.
worker内置锁在初始化的时候 state为(-1), 只有在worker.start()之后, 才执行worker.unlock()将state++, 将锁置于正常状态, 因为获取锁时都是执行 CAS(0,1) 操作
如果发现state>=0, 说明该worker已经被start(), 已经开始执行任务
其次worker有一个 interruptIfStarted(), state为-1时, 是无法进行中断的, 这应该是出于保护firsttask的目的
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted())
线程位于sleep中，无论何时都会响应 InterruptedException, 那么对于处于sleep的Worker, 执行shtdownNow()会直接抛出异常, 结束执行

execute 方法 - 不能保证顺序执行

1. 比较线程池中线程的数量和`core pool size`的大小关系

假如小于核心线程池大小，则调用 addWorker()，创建一个线程执行任务

addWorker 会实例化一个 worker 对象加入到 workers 中
还要执行一些检查和cas同步操作, 修改workingCount等
如何实现线程复用? 初次执行是来自于实例化时, 之后则来自于阻塞队列,
默认是无限定时间的, 如果自己是核心线程, 就会一直从阻塞队列中take任务, 一直阻塞一直存活一直工作
有限定时间的话, 任务线程从任务队列取任务（出队）时poll()是有最大等待时间的, 如果超时, 就对外返回null, 让当前线程结束
Excutors类提供了DefaultThreadFactory
ThreadFactory中包含ThreadGroup:group、
Thread t = new Thread(group, r,namePrefix + threadNumber.getAndIncrement(),0);

2. 无法直接创建就要进入阻塞队列

入队成功，我们也要二次检查能否创建一个线程，因为这个过程中可能有线程死亡，还要再检查一次线程池是否要关闭

如果线程池将要关闭，就 remove(command)&reject(command)，如果所有的线程都执行任务完毕(workingCount==0)，还要创建新线程

3. 两种尝试都失败

如果入队也失败，尝试MAX_POOL_SIZE模式创建线程也失败，就就会 reject(command);

线程池的关闭

关闭线程池，可以通过shutdown和shutdownNow这两个方法

shutdownNow

首先将线程池的状态设置为STOP,然后尝试停止所有的正在执行和未执行任务的线程，并返回等待执行任务的列表；

将运行状态设置为STOP - 修改ctl
执行interruptWorkers() 方法
for (Worker w : workers) w.interruptIfStarted();
就像名字一样, 如果worker已经start了, 调用每一个worker线程的 interrupt()
- 如果线程池状态为STOP，要保证当前线程是中断状态；
  如果不是的话，则要保证当前线程不是中断状态；
  如果是STOP, 需要保证worker在本次任务完成后自行退出
  如果不为STOP, 那么需要保证外界的interrupt不影响内部线程的执行
- 触发了STOP的interrupted-worker, 会在getTask的时候抛出异常进而执行processWorkerExit()
  为了表明和线程池为空的区别
然后执行tryTerminated()方法
主要是interruptIdleWorkers()
和ctl的更新
- interruptIdleWorkers() 目标是为了终止未在执行任务的worker, 这些worker会阻塞在workingQueue处, 能够成功获取锁, 然后interrupt
- 更新ctl, 过渡至tidying状态

shutdown

shutdown只是将线程池的状态设置为SHUTDOWN状态，然后中断所有没有正在执行任务的线程, 会等待线程执行完

区别在于先将状态改为SHUTDOWN
如果workingQueue不为空就不再采取措施, 等待线程池将任务执行完毕之后自行终止
当所有的线程都关闭成功，才表示线程池成功关闭，这时调用isTerminated方法才会返回 true

线程的死亡

外部扰动, 调用了 interruptIfStarted() 方法, 在执行阻塞队列的take()方法时异常, 离开循环
外部扰动, 将运行状态改为 STOP, 变为interrupted, 转情况1
外部扰动, 线程阻塞在workingQueue的take()方法处, 外部通过interruptIdleWorkers()将线程变为interrupted, 转情况1
内部消化 (wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty()), 任务为null, 自行终止
task.run() 方法执行过程中出现运行时错误, 异常在循环外被处理, 离开循环

码农公寓