java线程(四) 线程池原理

心已赠人 2023-06-10 10:27 22阅读 0赞

## 线程池 ##

作用：限制线程数，管理线程、避免频繁创建和销毁线程造成性能损耗

--------------------

## ThreadPoolExecutor ##

### 作用 ###

封装线程池的一系列逻辑，通过该类可创建线程池。

### 构造参数 ###

*  corePoolSize 核心线程数，有任务时才会创建线程，一般情况核心线程不会过期，一直占用，除非调`allowCoreThreadTimeOut`方法设置允许过期，核心线程才会因为空闲超时而过期
 *  keepAliveTime 线程保持存活时间。线程空闲超过指定时间后会超时，默认是超过`核心线程数`创建的线程才会超时终止。如果调用了allowCoreThreadTimeOut方法设置为true,则核心线程池超过空闲时间也会被回收。
 *  unit 指定超时时间的时间单位，TimeUnit有7个时间单位选择
 *  workqueue 工作队列，当线程数超过核心线程数时，请求会进入队列中等待分配线程。不同的队列进队出队有些区别。
 *  maximumPoolSize 最大线程数，表示线程池最大允许创建的线程数，当阻塞队列满了之后才会尝试继续创建线程，如果超过最大线程数，则会触发拒绝策略。
 *  RejectedExecutionHandler 拒绝处理器，当阻塞队列满了、并且当前线程数大于最大线程数时，请求获取线程会被拒绝，不同的拒绝策略有不同的执行方式。

### execute方法分析 ###

works变量: 工作线程集合，存储处于运行状态的线程。

**执行流程如下**

1.  首先获取当前处于工作状态的线程数，如果小于`核心线程数corePoolSize`，则将当前执行runnale封装成`worker`类，通过worker类(本身也是实现runnale)创建线程，加入处于工作状态的集合，并调用线程的`start`执行。
2.  `start方法会执行worker类的run方法，该方法会先执行当前任务`，当前任务执行结束后，去`阻塞队列workqueue`中取出其他任务，取到任务则继续执行。如果核心线程允许超时或者线程大于核心线程数，则会用阻塞队列的poll方法指定`keepAliveTime`等待时间，在这个时间内如果没有新的任务进入队列，则不再等待，停止执行。
3.  如果处于工作状态的线程数小于核心线程数，则`cas修改工作线程数`加入工作线程集合（cas失败是因为并发都去加入队列,要重新再判断再cas）。如果工作状态的线程数大于核心线程数，则加入`阻塞队列workqueue`。
4.  如果加入队列失败，则会判断是否处于工作状态的线程是否大于`最大线程数maximumPoolSize`，如果没有则尝试加入工作线程集合，封装worker创建线程，调用start方法执行。
5.  如果大于`最大线程数maximumPoolSize`则调用拒绝执行处理器进行拒绝处理。

### 工作队列 ###

阻塞队列均实现于BlockingQueue接口，列举常用的阻塞队列。队列的指定时间拉取通过Condition进行阻塞，offer后通过唤醒。  
condition : ReentrantLock::newCondition获取的condition

**ArrayBlockingQueue**

*  有界队列、规定大小的的阻塞队列。进队和出队为FIFO，进队和出队都用同一个ReentrantLock加锁。
 *  底层通过数组实现，需要指定容量的大小。

**LinkedBlockingQueue**

*  有界队列，但是可以不指定大小，默认为integer的最大值。进队和出队为FIFO，进出队用同个锁。
 *  底层通过双向链表实现。可以指定大小，通过维护一个count值计数当前的容量，同样进队和出队用同一个锁。相比ArrayBlockingQueue不用初始化时就占用指定大小的空间，而且也可以限制队列大小。

**PriorityBlockingQueue**

*  优先队列。需要构造方法指定Comparator用于排序进入队列的实现了Runnable的类，或者实现了Runnable的类要实现Comparable接口，用于进队出队后的排序，也就是执行方法。
 *  进出队用同个锁。
 *  底层通过`最小堆`实现，数组存储元素。无界，数组大小不足时50%扩展。小的优先级高。

**SynchronousQueue**

*  同步队列。每个进队都要等待另一个线程调用移除操作，否则一直阻塞。线程池调用的offer方法则不会去等待，直接存放后就返回。该队列底层使用`链表`，通过cas和生产者消费者匹配模式避免锁操作，节点分为生产者和消费者两种.假设来了一个生产者，如果头部当前为生产者则入队，如果为消费者则通过cas与队列中的消费者cas匹配，`无锁操作`吞吐量高。
 *  `支持公平队列`TransferQueue和`非公平队列`TransferStack，默认`非公平`。公平队列通过维护头尾节点实现，从头部开始匹配，发现是同个类型则把当前节点从`尾节点插入`。非公从头部开始匹配，发现是同个类型则`头节点插入`。
 *  poll的时候，如果与头部相同类型，把当前线程封装成`SNode`加入队列,通过`LockSupport::park`进入阻塞。当有`offer`进来匹配到时，通过SNode包含的Thread进行unpark唤醒这个线程。新进来的节点与头节点匹配时，把头节点的`SNode`的match设置成当前节点,被唤醒的。
 *  带有效时间阻塞，通过park阻塞指定时间，醒来时会判断是否返回的匹配节点与阻塞时的节点是同一个，同一个是代表没有匹配成功。
 *  按照这个队列的实现，链表上除了完成匹配的节点，要么为空，要么都是生产者节点，要么都是消费者节点。
 *  队列分为三种状态`REQUEST`拉去数据、`DATA`添加数据、`FULFILLING`匹配成功。poll方法对应REQUEST，offer方法对应DATA。进队时判断队列头与当前状态一样则将队列头替换为当前数据，旧的队列头为当前数据的next。通过将队列头cas为FULFILLING状态进行匹配，cas成功后则将当前队列头出队。

**DelayQueue**

*  延时队列。无界队列。依赖PriorityQueue优先级队列，队列的元素按照优先级存储，出队按照优先级出队。存放元素要实现Delayed接口的getDelay方法，通过这个方法获取当前元素还需要延迟多久，如果返回0则表示不用延迟，大于0则表示还需要延迟，则返回空。

### 拒绝策略 ###

**线程池达到最大线程后，新加入的任务会被拒绝执行RejectedExecutionHandler**

**AbortPolicy** 默认的拒绝策略，直接抛出异常

**DiscardPolicy** 该策略不做任何处理，也不抛出异常，相当于抛弃了任务

**DiscardOldestPolicy** 把阻塞队列中最老的任务删除，调用poll方法出队，即队头出列。如果是优先级队列可能就不是最老，而是最优先的。重新执行execute()

**CallerRunsPolicy** 如果执行失败则由主线程执行。

### 停止线程池执行 ###

shutdown 停止线程池，执行此方法后会暂停接受新任务，并等线程池队列中的任务都执行后，再停止。  
shutdownNow 停止线程池，暂停接受任务，抛弃队列中处于等待状态还未执行的任务，调用interrupt方法尝试中断线程。

--------------------

### ForkJoinPool ###

jdk7以上引入的新的线程池，`抽象类`。ThreadPoolExecutor线程池在执行线程任务时，如果有个任务特别耗时会一直占用某个线程，而其他线程又处于空闲状态。为了充分利用多核cpu资源，ForkJoinPool使用`分治法`的思想，将执行的任务进行分割成多个子任务，当任务被切分到设定的界限后，由各个线程各自执行。可以指定线程数，如果不指定则会获取当前处理器cpu核数作为线程数。

**ForkJoinTask**  
线程池执行任务的submit方法，会把任务封装成`ForkJoinTask`类，如果任务本身就是ForkJoinTask类或其子类则不会重复包装。

**RecursiveAction**  
ForkJoinTask子类（抽象类），ForkJoinTask类的子类，用于封装`没有返回值的任务`。需要继承该类，并重写`compute方法`进行任务和切分和执行，该方法`没有返回值`。ForkJoinPool在执行任务时，会通过调用该方法去执行任务。`compute方法需要实现任务的切分和任务的执行`，该方法可以实现当任务还没切分到设定的界限大小，则切分成多个子任务。分成多个任务后，再执行这些子任务，执行方式分为`fork、join、invokeAll`。

**RecursiveTask**  
ForkJoinTask子类（抽象类），类似RecursiveAction，要重写`compute方法`,compute`有返回值`。

**ForkJoinTask 执行子任务**  
invokeAll方法.将多个子任务添加到队列，不等待任务执行结果，这些任务会被该线程后续拉取或者被其他线程拉取执行。  
join方法.同步执行任务，阻塞直到获取到任务执行结果，调用该方法类似Thread的join方法。  
fork方法.将该子任务推入队列，等待被线程拉取。

**workQueues\[\]**  
阻塞队列数组，该线程池与ThreadPoolExecutor的各个线程共享阻塞队列不一样，ForkJoinPool线程池`每个线程都有一个单独的阻塞队列`。线程池执行任务时，会进行切分任务，通过获取当前线程的一个固定随机数，通过随机数计算命中阻塞队列数组的某个阻塞队列。切分的任务会放到放到这个队列中.每个队列`workQueue内部封装一个任务数组，即通过数组实现队列`。队列的push方法只能由该线程调用，而队列的pop方法则其他线程也能调用，当线程本身的队列没有任务则会去拉别人队列中的任务，且为避免竞争，线程拉取任务从队列头部拉取，窃取的线程从尾部拉取，即双端队列。这个算法被称为`工作窃取算法（work-stealing）`。

**执行任务的三个方法**  
execute 异步执行  
submit 异步执行,获取ForkJoinTask，需要获取结果可以调用task的get方法  
invoke 同步等待获取结果

**submit**  
第一次执行submit方法，先把任务封装成ForkJoinTask类，初始化workQueues数组（数组大小为`当前核心线程数的最接近2的幂次方的数的两倍`，比如核心线程数为10，则为16\*2），初始化workQueue，根据当前线程获取一个随机数计算到某个数组位置，将workQueue放入数组，将任务放入队列。然后会启动一个线程去队列中拉任务，拉取到任务后会执行`compute`方法（任务切分或任务执行），任务执行后继续去该线程对应的队列中拉取任务，队列没有数据后会去拉取别的线程的队列。

--------------------

### Executors ###

Executors提供了几种线程池，可以便捷的创建线程池。大部分通过ThreadPoolExecutor创建，只是提供一些默认的参数。

**newCachedThreadPool**  
缓存线程池。核心线程数为0，最大线程数为integer最大值，超时时间60秒，队列使用`SynchronousQueue`同步队列。由于核心线程数为0，相当于每次来一个任务都会进入队列。使用`非公平`队列。

1.  第一个线程进入队列时，因为线程池调用的队列的不指定超时时间offer方法，此方法当队列头没数据时是直接返回的，由于最大线程数是Integer的最大值，所以会直接创建一个线程执行任务，执行结束后会带`超时时间keepAliveTime`调用poll方法获取任务，此时会进队等待匹配。
2.  此时来第二个任务时，该任务与该线程匹配到，此时会将队列头替换成下一个元素（又变成空），返回该任务，该线程执行该任务。
3.  此时如果第二任务线程还没执行，那么第三个任务调用poll方法又是返回空，所以会再去创建线程。
4.  如果两个线程都执行完，都带超时时间进队，那么都会进队，变成head=线程1，线程.next=线程2。

`综上所诉，newCachedThreadPool线程池每来个任务，就判断有没有空闲线程，没有就会创建线程，有就复用线程，最高的线程数是integer最大值。`

**newFixedThreadPool**  
固定线程池。创建要指定线程数，且**最大线程数和核心线程数相等**，队列使用LinkedBlockingQueue，默认线程不会超时，线程池一直占用线程。`最大队列数为int最大值，线程高于核心线程数就直接进队列,队列满了直接拒绝`。

**newSingleThreadExecutor**  
单一线程池。与固定线程池机制类似，`且线程数固定为一个线程`。

**newScheduledThreadPool**  
支持定时、延迟执行任务的线程池。通过创建ScheduledThreadPoolExecutor类实现该线程池，该类通过继承ThreadPoolExecutor实现。

*  构造指定`核心线程数`，最大线程数设置为integer最大值,`keepAliveTime`设置为0，`该线程池的执行方法只有线程数没超过核心线程池才会创建线程去队列中拉任务`，否则不会创建线程，因此最大线程数和keepAliveTime没有作用，`最大线程数不会超过核心线程数`。`线程创建后不过期，一直去拉取任务.`
 *  队列使用该类内部封装的延迟队列`DelayedWorkQueue`。该队列通过数组实现，数组容量不足时1.5倍扩容（小堆）。ScheduledThreadPoolExecutor的schedule方法，执行延迟任务时，会将任务封装成`ScheduledFutureTask`类(包含了任务和开始执行任务的时间，即延时时间加上当前时间)进队,进队列的时候会`根据执行任务的时间排序`，越小的排越前面，`相同时间的早进队的排在前面`。进队后判断如果当前的工作线程数没有超过核心线程数，则创建一个线程去队列中拉任务，从队列头开始拉。
 *  任务会被封装到`ScheduledFutureTask`去执行，如果是固定速率的，会有一个标示，并记录任务间隔，每次执行完判断是固定速率，会把间隔加上上个任务开始时间作为下个任务开始时间加入队列，然后继续拉取任务。

**newWorkStealingPool**  
jdk8新增，创建一个forkJoinPool，可以指定线程数，不指定则为cpu核数。