Java多线程与并发编程(8)_ThreadPoolExecutor

深碍√TFBOYSˉ_ 2023-02-19 12:21 41阅读 0赞

# Java多线程与并发编程(8)\_ThreadPoolExecutor #

ThreadPoolExecutor是我们使用Java线程池的核心，它所处的体系如下：

### 文章目录 ###

*  Java多线程与并发编程(8)\_ThreadPoolExecutor
 *   *  Executor体系结构
     *  线程池状态
     *  线程池构造参数
     *  线程池工作方式
     *  创建各种线程池
     *  FixedThreadPool
     *  CachedThreadPool
     *  SingleThreadExecutor
     *  ScheduledThreadPool
     *  启动线程池
     *  关闭线程池
     *  异常处理

## Executor体系结构 ##

![format_png][]

## 线程池状态 ##

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量

![format_png 1][]

*  从数字上比较线程池的终结程度，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING
 *  这些信息存储在一个原子变量 ctl 中，目的是将**线程池状态与线程个数**合二为一，这样就可以用一次 cas 原子操作进行赋值
    
     *      // c 为旧值， ctlOf 返回结果为新值
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
            // rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
            private static int ctlOf(int rs, int wc) {
                     return rs | wc; //合并}

## 线程池构造参数 ##

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

通过构造方法，我们可以分析Java线程池的几个核心要素

*  corePoolSize 核心线程数目 (最多保留的线程数)
 *  maximumPoolSize 最大线程数目 包含救急线程
 *  keepAliveTime 生存时间 - 针对救急线程
 *  unit 时间单位
 *  workQueue 工作队列/阻塞队列
 *  threadFactory 线程工厂 - 可以为线程创建时起个好名字
 *  handler 拒绝策略

## 线程池工作方式 ##

线程池可以抽象成如下模型

![format_png 2][]

*  线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
 *  当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue排队，直到有空闲的线程。
 *  **如果队列选择了有界队列，那么任务超过了队列大小时**，会创建 maximumPoolSize - corePoolSize(图中m-c) 数目的线程来救急（图中救急线程）。即workQueue的优先级大于救急线程
 *  如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现
    
     *  AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
     *  CallerRunsPolicy 让调用者运行任务
     *  DiscardPolicy 放弃本次任务
     *  DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
     *  Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
     *  Netty 的实现，是创建一个新线程来执行任务
     *  ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
     *  PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
 *  当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由

keepAliveTime 和 unit 来控制。

Java中拒绝策略的实现类如下：

统一继承自RejectExecutionHandler接口

![format_png 3][]

## 创建各种线程池 ##

由于ThreadPoolExecutor的参数过多，JDK Executors 类中提供了众多**工厂方法**来创建各种用途的线程池

其中拒绝策略又通过泛函接口实现，这是一个策略模式，不同场景下用不同算法

![format_png 4][]

## FixedThreadPool ##

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>(),
                                          threadFactory);
        }

特点：

*  第一个参数为指定核心容量，第二个参数可以自定义命名方式
 *  核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
 *  任务队列是无界的（`LinkedBlockingQueue`），可以放任意数量的任务
 *  **适用于任务量已知，相对耗时的任务**

## CachedThreadPool ##

public static ExecutorService newCachedThreadPool() {
        
         return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
         60L, TimeUnit.SECONDS,
         new SynchronousQueue<Runnable>());
    }

特点：

*  核心线程数是 0， 最大线程数是 Integer.MAX\_VALUE，救急线程的空闲生存时间是 60s，意味着
    
     *  全部都是救急线程（60s 后可以回收）
     *  救急线程可以无限创建
 *  队列采用了 SynchronousQueue 实现，它没有容量，没有线程来取是放不进去的（一手交钱、一手交
    
    货）

> **整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。 适合任务数比较密集，但每个任务执行时间较短的情况**

## SingleThreadExecutor ##

public static ExecutorService newSingleThreadExecutor() {
        
         return new FinalizableDelegatedExecutorService
         (new ThreadPoolExecutor(1, 1,
         0L, TimeUnit.MILLISECONDS,
         new LinkedBlockingQueue<Runnable>()));
    }

使用场景：

希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放。

**与自己创建一个线程去执行任务的区别**：

自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一

个线程，保证池的正常工作

## ScheduledThreadPool ##

在『任务调度线程池』功能加入之前，可以使用 java.util.Timer 来实现定时功能，Timer 的优点在于简单易用，但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务

`ScheduledExecutorService`基本使用：

System.out.println("主线程开始运行   "+new Date());
    		//线程池两个线程处理两个任务
    		ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
            // 添加两个任务，希望它们都在 1s 后执行
            executor.schedule(() -> {
        
                System.out.println("任务1，执行时间：" + new Date());
                  Thread.sleep(2000); 
            }, 1000, TimeUnit.MILLISECONDS);
            executor.schedule(() -> {
        
                System.out.println("任务2，执行时间：" + new Date());
            }, 1000, TimeUnit.MILLISECONDS);

> 主线程开始运行 Sat Jun 20 14:22:38 CST 2020  
> 任务2，执行时间：Sat Jun 20 14:22:40 CST 2020  
> 任务1，执行时间：Sat Jun 20 14:22:40 CST 2020

`scheduleAtFixedRate()`:以一定频率反复执行一个任务

ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
            System.out.println("start..."+new Date());
    
            /*第一个1 - initialDelay:初始延迟时间*/
            /*第二个1 - period:间隔时间*/
            pool.scheduleAtFixedRate(() -> {
        
                System.out.println("running"+new Date());
            }, 1, 1, TimeUnit.SECONDS);

> start…Sat Jun 20 14:27:00 CST 2020  
> runningSat Jun 20 14:27:01 CST 2020  
> runningSat Jun 20 14:27:02 CST 2020  
> runningSat Jun 20 14:27:03 CST 2020  
> runningSat Jun 20 14:27:04 CST 2020  
> runningSat Jun 20 14:27:05 CST 2020

如果任务执行时间超过了间隔时间呢？比如这样

ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
            System.out.println("start..."+new Date());
            pool.scheduleAtFixedRate(() -> {
        
                System.out.println("running..."+new Date());
                //模拟单个线程执行时间超过
                try {
        
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }, 1, 1, TimeUnit.SECONDS);
            /*第一个1 - initialDelay:初始延迟时间*/
            /*第二个1 - period:间隔时间*/

> 结果：
> 
> start…Sat Jun 20 14:32:15 CST 2020  
> running…Sat Jun 20 14:32:16 CST 2020  
> running…Sat Jun 20 14:32:18 CST 2020  
> running…Sat Jun 20 14:32:20 CST 2020  
> running…Sat Jun 20 14:32:22 CST 2020  
> running…Sat Jun 20 14:32:24 CST 2020

分析可知，构造时指定的间隔“一”将失效，**以线程具体运行时间为真实间隔单位**

**解决方案-`scheduleWithFixedDelay`**

ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
            System.out.println("start..."+new Date());
            pool.scheduleWithFixedDelay(()-> {
        
                System.out.println("running..."+new Date());
                    Thread.sleep(2000);
            }, 1, 1, TimeUnit.SECONDS);
            /*第一个1 - initialDelay:初始延迟时间*/
            /*第二个1 - period:间隔时间*/

> 结果：
> 
> start…Sat Jun 20 14:35:43 CST 2020  
> running…Sat Jun 20 14:35:44 CST 2020  
> running…Sat Jun 20 14:35:47 CST 2020  
> running…Sat Jun 20 14:35:50 CST 2020  
> running…Sat Jun 20 14:35:53 CST 2020

这回真实间隔时间就是构造指定的延迟“1”+线程执行时间“2” = 3s

## 启动线程池 ##

// 执行任务
    void execute(Runnable command);
    
    // 提交任务 task，用返回值 Future 获得任务执行结果 带返回值
    <T> Future<T> submit(Callable<T> task);

## 关闭线程池 ##

`void shutdown();`

/*
    线程池状态变为 SHUTDOWN
    - 不会接收新任务
    - 但已提交任务会执行完
    - 此方法不会阻塞调用线程的执行
    */
    void shutdown();

**源码分析**：

public void shutdown() {
        
         final ReentrantLock mainLock = this.mainLock;
         mainLock.lock();
         try {
        
         		checkShutdownAccess();
         // 修改线程池状态
         		advanceRunState(SHUTDOWN);
         // 仅会打断空闲线程，还是会继续处理阻塞队列中的任务
         		interruptIdleWorkers();
        		 onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
         } finally {
        
         		  mainLock.unlock();
         }
         // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等，让线程自己去运行结束)
        		 tryTerminate();
    }

`List<Runnable> shutdownNow();`

/*
    线程池状态变为 STOP
    - 不会接收新任务
    - 会将队列中的任务返回
    - 并用 interrupt 的方式中断正在执行的任务
    */
    List<Runnable> shutdownNow();

public List<Runnable> shutdownNow() {
        
    List<Runnable> tasks;
         final ReentrantLock mainLock = this.mainLock;
         mainLock.lock();
         try {
        
         checkShutdownAccess();
         // 修改线程池状态
         advanceRunState(STOP);
         // 打断所有线程
         interruptWorkers();
         // 获取队列中剩余任务
         tasks = drainQueue();
         } finally {
        
         mainLock.unlock();
         }
         // 尝试终结
         tryTerminate();
         return tasks;
    }

**其他方法**：

// 不在 RUNNING 状态的线程池，此方法就返回 true
    boolean isShutdown();
    // 线程池状态是否是 TERMINATED
    boolean isTerminated();
    // 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事
    情，可以利用此方法等待
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

## 异常处理 ##

在线程池中，单个线程出现异常，不会抛出，我们可以采取：

*  主动捕获异常
    
        ExecutorService pool = Executors.newFixedThreadPool(1);
                pool.submit(() -> {
              
                    try {
              
                        System.out.println("task1");
                        int i = 1 / 0;
                    } catch (Exception e) {
              
                        System.out.println("error:"+e);
                    }
                });
    
    > 结果：
    > 
    > task1  
    > error:java.lang.ArithmeticException: / by zero
 *  使用带有返回值的任务Callable与Future
    
        ExecutorService pool = Executors.newFixedThreadPool(1);
                Future<Boolean> f = pool.submit(() -> {
              
                    System.out.println("task1");
                    int i = 1 / 0;
                    return true;
                });
                System.out.println("result:"+f.get());
    
    > 结果：
    > 
    > task1  
    > Exception in thread “main” java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zero  
    > at java.util.concurrent.FutureTask.report(FutureTask.java:122)  
    > at java.util.concurrent.FutureTask.get(FutureTask.java:192)  
    > at com.Thread.TestThreadPool.main(TestThreadPool.java:19)  
    > Caused by: java.lang.ArithmeticException: / by zero  
    > at com.Thread.TestThreadPool.lambda$main 0 ( T e s t T h r e a d P o o l . j a v a : 16 ) a t j a v a . u t i l . c o n c u r r e n t . F u t u r e T a s k . r u n ( F u t u r e T a s k . j a v a : 266 ) a t j a v a . u t i l . c o n c u r r e n t . T h r e a d P o o l E x e c u t o r . r u n W o r k e r ( T h r e a d P o o l E x e c u t o r . j a v a : 1142 ) a t j a v a . u t i l . c o n c u r r e n t . T h r e a d P o o l E x e c u t o r 0(TestThreadPool.java:16) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor 0(TestThreadPool.java:16)atjava.util.concurrent.FutureTask.run(FutureTask.java:266)atjava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)atjava.util.concurrent.ThreadPoolExecutorWorker.run(ThreadPoolExecutor.java:617)  
    > at java.lang.Thread.run(Thread.java:745)

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwNDI1MTAucG5n?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwNDI3MDAucG5n?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwNDQ4MTAucG5n?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwNDQ2MzYucG5n?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vanVuc2lyMS9NeVBpY0Nsb3VkU3RvcmUvcmF3L21hc3Rlci9pbWcvMjAyMDA2MjAwNTAwNTIucG5n?x-oss-process=image/format,png