12.常用线程池与线程拒绝策略汇总学习

骑猪看日落 2023-10-08 17:36 46阅读 0赞

## 常用线程池 ##

### 1.FixedThreadPool固定线程池 ###

FixedThreadPool，它的核心线程数和最大线程数是一样的，所以可以把它看作是固定线程数的线程池，它的特点是线程池中的线程数除了初始阶段需要从 0 开始增加外，之后的线程数量就是固定的，就算任务数超过线程数，线程池也不会再创建更多的线程来处理任务，而是会把超出线程处理能力的任务放到任务队列中进行等待。而且就算任务队列满了，到了本该继续增加线程数的时候，由于它的最大线程数和核心线程数是一样的，所以也无法再增加新的线程了。

ExecutorService service = Executors.newFixedThreadPool(1);

### 2.CachedThreadPool缓存线程池 ###

CachedThreadPool，可以称作可缓存线程池，它的特点在于线程数是几乎可以无限增加的（实际最大可以达到 Integer.MAX\_VALUE，为 2^31-1=2 147 483 647，这个数非常大，所以基本不可能达到），而当线程闲置时还可以对线程进行回收。

也就是说该线程池的线程数量不是固定不变的，当然它也有一个用于存储提交任务的队列，但这个队列是 SynchronousQueue，队列的容量为0，实际不存储任何任务，它只负责对任务进行中转和传递，所以效率比较高。

当我们提交一个任务后，线程池会判断已创建的线程中是否有空闲线程，如果有空闲线程则将任务直接指派给空闲线程，如果没有空闲线程，则新建线程去执行任务，这样就做到了动态地新增线程。

ExecutorService service = Executors.newCachedThreadPool();

### 3.ScheduledThreadPool定时或周期性线程池 ###

ScheduledThreadPool，它支持定时或周期性执行任务。比如每隔 10 秒钟执行一次任务。

ScheduledExecutorService service = Executors.newScheduledThreadPool(10);
    //表示延迟指定时间后执行一次任务，如果代码中设置参数为 10 秒，也就是 10 秒后执行一次任务后就结束。
    service.schedule(new Task(), 10, TimeUnit.SECONDS);
    
    //scheduleAtFixedRate 表示以固定的频率执行任务，它的第二个参数 initialDelay 表示第一次延时时间，第三个参数 period 表示周期，也就是第一次延时后每次延时多长时间执行一次任务。10秒延迟后 每10秒这个周期执行一次任务
    service.scheduleAtFixedRate(new Task(), 10, 10, TimeUnit.SECONDS);
    
    // scheduleWithFixedDelay 与第二种方法类似，也是周期执行任务，区别在于对周期的定义，之前的 scheduleAtFixedRate 是以任务开始的时间为时间起点开始计时，时间到就开始执行第二次任务，而不管任务需要花多久执行；而 scheduleWithFixedDelay 方法以任务结束的时间为下一次循环的时间起点开始计时。
    service.scheduleWithFixedDelay(new Task(), 10, 10, TimeUnit.SECONDS);

第三个可能有点迷糊，举个例子比如程序员需要多喝水，第二种方式假如设置了1小时，喝水需要10分钟，那每个整点都需要喝一次水，时长10分钟。

那第三种呢就是 我1小时+10分钟喝水时长，作为下一次的开始计时。

### 4.SingleThreadExecutor唯一线程线程池 ###

SingleThreadExecutor，它会使用唯一的线程去执行任务，原理和 FixedThreadPool 是一样的，只不过这里线程只有一个，如果线程在执行任务的过程中发生异常，线程池也会重新创建一个线程来执行后续的任务。

这种线程池由于只有一个线程，所以非常适合用于所有任务都需要按被提交的顺序依次执行的场景，而前几种线程池不一定能够保障任务的执行顺序等于被提交的顺序，因为它们是多线程并行执行的。

ExecutorService service = Executors.newSingleThreadExecutor();

### 5.SingleThreadScheduledExecutor定时唯一线程池 ###

SingleThreadScheduledExecutor，它实际和第三种 ScheduledThreadPool 线程池非常相似，它只是 ScheduledThreadPool 的一个特例，内部只有一个线程。

ExecutorService service = Executors.newSingleThreadScheduledExecutor();

源码：

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        
            return new DelegatedScheduledExecutorService
                (new ScheduledThreadPoolExecutor(1));
        }

### 6.ForkJoinPool拆分（子任务）-汇总线程池 ###

ForkJoinPool，这个线程池是在 JDK 7 加入的，它的名字 ForkJoin 也描述了它的执行机制，主要用法和之前的线程池是相同的，也是把任务交给线程池去执行，线程池中也有任务队列来存放任务。场景应用也比较广泛。

#### 适合执行可以产生子任务的任务 ####

我们有一个 Task，这个 Task 可以产生三个子任务，三个子任务并行执行完毕后将结果汇总给 Result，比如说主任务需要执行非常繁重的计算任务，我们就可以把计算拆分成三个部分，这三个部分是互不影响相互独立的，这样就可以利用 CPU 的多核优势，并行计算，然后将结果进行汇总。

这里面主要涉及两个步骤，第一步是拆分也就是 Fork，第二步是汇总也就是 Join。

**案例：波那切数列**

import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * @author cf
     * @description: ForkJoinPool 案例
     * 菲波那切数列，这个数列的特点就是后一项的结果等于前两项的和，第 0 项是 0，第 1 项是 1，那么第 2 项就是 0+1=1，以此类推。
     * @date 2023/2/13下午 2:21
     */
    public class ForkJoinPoolDemo {
        
        public static void main(String[] args) throws ExecutionException, InterruptedException {
        
            ForkJoinPool  forkJoinPool = new ForkJoinPool();
            for (int i = 0; i < 10; i++) {
        
                ForkJoinTask task = forkJoinPool.submit(new ForkJoinPoolTest(i));
                System.out.println(task.get());
            }
        }
    }
    
    /**
     * 继承了 RecursiveTask，RecursiveTask 类是对ForkJoinTask 的一个简单的包装
     */
    class ForkJoinPoolTest extends RecursiveTask<Integer> {
        
    
        int n;
    
        public ForkJoinPoolTest(int n) {
        
            this.n = n;
        }
    
    
        /**
         * 重写 compute() 方法，当 n<=1 时直接返回，当 n>1 就创建递归任务，也就是 f1 和 f2，
         * 然后我们用 fork() 方法分裂任务并分别执行，最后在 return 的时候，使用 join() 方法把结果汇总
         * @return
         */
        @Override
        public Integer compute() {
        
            if (n <= 1) {
        
                return n;
            }
            ForkJoinPoolTest f1 = new ForkJoinPoolTest(n - 1);
            f1.fork();
            ForkJoinPoolTest f2 = new ForkJoinPoolTest(n - 2);
            f2.fork();
            return f1.join() + f2.join();
        }
    }

#### 内部结构不同 ####

之前的线程池所有的线程共用一个队列，但 ForkJoinPool 线程池中每个线程都有自己独立的任务队列。

ForkJoinPool 线程池内部除了有一个共用的任务队列之外，每个线程还有一个对应的双端队列 deque，这时一旦线程中的任务被 Fork 分裂了，分裂出来的子任务放入线程自己的 deque 里，而不是放入公共的任务队列中。

如果此时有三个子任务放入线程 t1 的 deque 队列中，对于线程 t1 而言获取任务的成本就降低了，可以直接在自己的任务队列中获取而不必去公共队列中争抢也不会发生阻塞（除了后面会讲到的 steal 情况外），减少了线程间的竞争和切换，是非常高效的。

#### （扩展）work-stealing 偷工作 ####

双端队列 deque 中，线程 t1 获取任务的逻辑是后进先出，也就是LIFO（Last In Frist Out），而线程 t0 在“steal”偷线程 t1 的 deque 中的任务的逻辑是先进先出，也就是FIFO（Fast In Frist Out）。

使用 “work-stealing” 算法和双端队列很好地平衡了各线程的负载。

所以ForkJoinPool 非常适合用于递归的场景，例如树的遍历、最优路径搜索等场景。

## 线程拒绝策略 ##

### 为什么要使用拒绝策略 ###

因为线程池没有能力继续处理新提交的任务，也就是工作已经非常饱和的时候，还要一直塞任务进去，会造成任务积压，也就是由于工作饱和导致的拒绝，这个时候再提交新任务时就会被拒绝。

### 1.AbortPolicy（中止策略） ###

这种拒绝策略在拒绝任务时，会直接抛出一个类型为 RejectedExecutionException 的 RuntimeException，让你感知到任务被拒绝了，于是你便可以根据业务逻辑选择重试或者放弃提交等策略。

### 2.DiscardPolicy（直接被丢弃） ###

这种拒绝策略正如它的名字所描述的一样，当新任务被提交后直接被丢弃掉，也不会给你任何的通知，相对而言存在一定的风险，因为我们提交的时候根本不知道这个任务会被丢弃，可能造成数据丢失。

### 3.DiscardOldestPolicy（丢弃任务队列中的头结点） ###

如果线程池没被关闭且没有能力执行，则会丢弃任务队列中的头结点，通常是存活时间最长的任务，这种策略与第二种不同之处在于它丢弃的不是最新提交的，而是队列中存活时间最长的，这样就可以腾出空间给新提交的任务，但同理它也存在一定的数据丢失风险。

### 4.CallerRunsPolicy（丢回） ###

相对而言它就比较完善了，当有新任务提交后，如果线程池没被关闭且没有能力执行，则把这个任务交于提交任务的线程执行，也就是谁提交任务，谁就负责执行任务【丢回去了任务】。这样做主要有两点好处。

*  第一点新提交的任务不会被丢弃，这样也就不会造成业务损失。
 *  第二点好处是，由于谁提交任务谁就要负责执行任务，这样提交任务的线程就得负责执行任务，而执行任务又是比较耗时的，在这段期间，提交任务的线程被占用，也就不会再提交新的任务，减缓了任务提交的速度，相当于是一个负反馈。在此期间，线程池中的线程也可以充分利用这段时间来执行掉一部分任务，腾出一定的空间，相当于是给了线程池一定的缓冲期。