Java多线程 - 创建线程池的方法 - ThreadPoolExecutor和Executors

以你之姓@ 2023-09-23 13:53 73阅读 0赞

#### 文章目录 ####

*   *  线程池(重点)
     *   *  线程池介绍
         *  实现线程池的方式
         *   *  方式一: 实现类ThreadPoolExecutor
             *   *  ThreadPoolExecutor构造器的参数
                 *  线程池处理Runnable任务
                 *  线程池处理Callable任务
             *  方式二: Executors工具类创建线程池

### 线程池(重点) ###

#### 线程池介绍 ####

**什么是线程池**?

> 线程池就是一个可以复用线程的技术。

**不使用线程池的问题**:

> 如果用户每发起一个请求，后台就创建一个新线程来处理，下次新任务来了又要创建新线程，而创建新线程的开销是很大的，这样会严重影响系统的性能。

**线程池工作原理**:

> 例如线程池中最多可以允许创建三个工作线程, 也叫核心线程, 前面三个任务来的时候会给前面三个任务单独创建三个线程; 但是后面任务再来的时候, 因为创建的工作线程已达到最大数, 那么后面的任务就会进入任务队列中排队等待; 等前面的任务执行完成, 有空闲的线程的时候使用空闲的线程依次执行任务队列中的任务

![在这里插入图片描述][c0842db7a2ce4926841ff7a668ed4a43.png_pic_center]

#### 实现线程池的方式 ####

**谁代表线程池**?

> JDK 5.0起提供了`代表线程池的接口：ExecutorService`

**如何得到线程池对象**?

*  方式一：使用ExecutorService的实现类ThreadPoolExecutor自创建一个线程池对象
 *  方式二：使用Executors（线程池的工具类）调用方法返回不同特点的线程池对象

##### 方式一: 实现类ThreadPoolExecutor #####

###### ThreadPoolExecutor构造器的参数 ######

> ThreadPoolExecutor的构造器有如下参数

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

> 参数介绍:
> 
>  *  参数一：指定线程池的线程数量(核心线程): corePoolSize ----> 不能小于0
>  *  参数二：指定线程池可支持的最大线程数： maximumPoolSize ----> 最大数量 >= 核心线程数量
>  *  参数三：指定临时线程的最大存活时间： keepAliveTime ----> 不能小于0
>  *  参数四：指定存活时间的单位(秒、分、时、天): unit ----> 时间单位
>  *  参数五：指定任务队列： workQueue ----> 不能为null
>  *  参数六：指定用哪个线程工厂创建线程： threadFactory ----> 不能为null
>  *  参数七：指定线程忙，任务满的时候，新任务拒绝策略： handler ----> 不能为null

**新任务拒绝策略**:

<table> 
 <thead> 
  <tr> 
   <th>策略</th> 
   <th>详解</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ThreadPoolExecutor.AbortPolicy</td> 
   <td>丢弃任务并抛出RejectedExecutionException异常。是默认的策略</td> 
  </tr> 
  <tr> 
   <td>ThreadPoolExecutor.DiscardPolicy</td> 
   <td>丢弃任务，但是不抛出异常 这是不推荐的做法</td> 
  </tr> 
  <tr> 
   <td>ThreadPoolExecutor.DiscardOldestPolicy</td> 
   <td>抛弃队列中等待最久的任务 然后把当前任务加入队列中</td> 
  </tr> 
  <tr> 
   <td>ThreadPoolExecutor.CallerRunsPolicy</td> 
   <td>由主线程负责调用任务的run()方法从而绕过线程池直接执行</td> 
  </tr> 
 </tbody> 
</table>

**ThreadPoolExecutor创建线程池对象示例**:

ExecutorService pools = new ThreadPoolExecutor(3, 
                                                   5, 
                                                   8, 
                                                   TimeUnit.SECONDS, 
                                                   new ArrayBlockingQueue<>(6), 
                                                   Executors.defaultThreadFactory(), 
                                                   new ThreadPoolExecutor.AbortPolicy());

**思考**:

> **临时线程什么时候创建啊**?
> 
>  *  新任务提交时发现核心线程都在忙，任务队列也满了，并且还可以创建临时线程，此时才会创建临时线程。
> 
> **什么时候会开始拒绝任务**?
> 
>  *  核心线程和临时线程都在忙，任务队列也满了，新的任务过来的时候才会开始任务拒绝。

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

###### 线程池处理Runnable任务 ######

**ExecutorService的常用方法**:

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>execute(Runnable command)</code></td> 
   <td>执行任务/命令，没有返回值，一般用来执行 Runnable 任务</td> 
  </tr> 
  <tr> 
   <td>shutdown()</td> 
   <td>等任务执行完毕后关闭线程池(一般不会关闭线程池)</td> 
  </tr> 
  <tr> 
   <td>shutdownNow()</td> 
   <td>立刻关闭，停止正在执行的任务，并返回队列中未执行的任务(一般不会关闭线程池)</td> 
  </tr> 
 </tbody> 
</table>

**演示代码**:

> 创建一个Runnable任务线程类

public class MyRunnable implements Runnable{
        
        @Override
        public void run() {
        
            for (int i = 0; i < 5; i++) {
        
                System.out.println(Thread.currentThread().getName() + "线程执行输出: " + i);
            }
            // 为了测试, 我们让每个线程睡眠3秒
            try {
        
                System.out.println(Thread.currentThread().getName() + "线程进入休眠");
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName() + "线程执行完成");
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    }

> 在主类中, 创建一个线程池对象并创建Runnable任务交给线程池处理

*  如果只有三个任务, 那么会被三个核心线程同时执行

*  打印结果如下

pool-1-thread-1线程执行输出: 0
    pool-1-thread-2线程执行输出: 0
    pool-1-thread-3线程执行输出: 0
    pool-1-thread-2线程执行输出: 1
    pool-1-thread-2线程执行输出: 2
    pool-1-thread-2线程执行输出: 3
    pool-1-thread-1线程执行输出: 1
    pool-1-thread-2线程执行输出: 4
    pool-1-thread-3线程执行输出: 1
    pool-1-thread-3线程执行输出: 2
    pool-1-thread-3线程执行输出: 3
    pool-1-thread-1线程执行输出: 2
    pool-1-thread-3线程执行输出: 4
    pool-1-thread-1线程执行输出: 3
    pool-1-thread-1线程执行输出: 4
    pool-1-thread-2线程进入休眠
    pool-1-thread-3线程进入休眠
    pool-1-thread-1线程进入休眠
    pool-1-thread-3线程执行完成
    pool-1-thread-1线程执行完成
    pool-1-thread-2线程执行完成

*  如果核心线程全部被占用, 那么后面的任务会进入任务队列中排队等待有空闲的核心线程; 由于我们设置的任务队列数是5, 所以进入任务队列的任务数量小于等于5时, 不会创建临时线程

public static void main(String[] args) {
        
        // 1. 创建一个线程池对象 核心线程为5, 最大线程数5, 临时线程存活6秒, 任务队列最大为5
        ExecutorService es = new ThreadPoolExecutor(3, 5, 6,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
        // 2. Runnable任务交给任务线程池处理
        Runnable target = new MyRunnable();
        // 三个Runnable任务会被核心线程执行
        es.execute(target);
        es.execute(target);
        es.execute(target);
    
        // 三个核心线程被占满, 会进入任务队列排队等待有空闲的核心线程
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
    }

*  由于核心线程都在忙, 任务队列也满了, 这时候我们再继续添加任务时, 就会创建临时线程; 因为我们设置的核心线程数是3个, 最大线程数是5个, 所以临时线程最多只会创建两个

public static void main(String[] args) {
        
        // 1. 创建一个线程池对象 核心线程为5, 最大线程数5, 临时线程存活6秒, 任务队列最大为5
        ExecutorService es = new ThreadPoolExecutor(3, 5, 6,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
        // 2. Runnable任务交给任务线程池处理
        Runnable target = new MyRunnable();
        // 三个Runnable任务会被核心线程执行
        es.execute(target);
        es.execute(target);
        es.execute(target);
    
        // 三个核心线程被占满, 会进入任务队列排队等待有空闲的核心线程(我们设置的任务队列最大允许排队5个任务)
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
    
        // 三个核心线程都在忙, 任务队列满, 创建临时线程
        es.execute(target);
        es.execute(target);
    }

*  核心线程都忙, 任务队列满, 临时线程忙, 此时再继续添加任务就会触发拒绝任务策略

public static void main(String[] args) {
        
        // 1. 创建一个线程池对象 核心线程为5, 最大线程数5, 临时线程存活6秒, 任务队列最大为5
        ExecutorService es = new ThreadPoolExecutor(3, 5, 6,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
        // 2. Runnable任务交给任务线程池处理
        Runnable target = new MyRunnable();
        // 三个Runnable任务会被核心线程执行
        es.execute(target);
        es.execute(target);
        es.execute(target);
    
        // 三个核心线程被占满, 会进入任务队列排队等待有空闲的核心线程(我们设置的任务队列最大允许排队5个任务)
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
        es.execute(target);
    
        // 三个核心线程都在忙, 任务队列满, 创建临时线程
        es.execute(target);
        es.execute(target);
    
        // 后续任务拒绝任务会被拒绝
        es.execute(target);
    }

###### 线程池处理Callable任务 ######

**ExecutorService常用方法**:

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>submit(Callable&lt;T&gt; task) </code></td> 
   <td>执行Callable任务，返回未来任务对象(FutureTask)获取线程结果</td> 
  </tr> 
  <tr> 
   <td>shutdown()</td> 
   <td>等任务执行完毕后关闭线程池</td> 
  </tr> 
  <tr> 
   <td>shutdownNow()</td> 
   <td>立刻关闭，停止正在执行的任务，并返回队列中未执行的任务</td> 
  </tr> 
 </tbody> 
</table>

> **注意: submit方法返回的是Future对象, Future对象是FutureTask对象继承的父类**

**演示代码**:

> 定义一个Callable任务类, 用于计算1到n的和返回

package com.chenyq.d4_threadpool2;
    
    import java.util.concurrent.Callable;
    
    public class MyCallable implements Callable<String> {
        
        private int n;
        public MyCallable(int n) {
        
            this.n = n;
        }
    
        @Override
        public String call() throws Exception {
        
            int sum = 0;
            for (int i = 1; i <= n ; i++) {
        
                sum += i;
            }
            // 为了测试, 让每一个线程任务睡眠3秒
            System.out.println(Thread.currentThread().getName() + "线程进入睡眠");
            Thread.sleep(3000);
            System.out.println(Thread.currentThread().getName() + "线程执行完成");
            return Thread.currentThread().getName() + "执行1-" + n + "结果是: " + sum;
        }
    }

> 线程池处理Callable任务时的细节和处理Callable的一样, 这里不再一一赘述, 代码如下

public static void main(String[] args) {
        
        // 1. 创建一个线程池对象 核心线程为5, 最大线程数5, 临时线程存活6秒, 任务队列最大为5
        ExecutorService pool = new ThreadPoolExecutor(3, 5, 6,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
        // 2. 创建Callable任务, 调用submit方法, 交给任务线程池处理, 返回未来线程对象
        // 三个Runnable任务会被核心线程执行
        Future<String> f1 = pool.submit(new MyCallable(10));
        Future<String> f2 = pool.submit(new MyCallable(20));
        Future<String> f3 = pool.submit(new MyCallable(30));
    
        // 三个核心线程被占满, 会进入任务队列排队等待有空闲的核心线程(我们设置的任务队列最大允许排队5个任务)
        Future<String> f4 = pool.submit(new MyCallable(30));
        Future<String> f5 = pool.submit(new MyCallable(40));
        Future<String> f6 = pool.submit(new MyCallable(50));
        Future<String> f7 = pool.submit(new MyCallable(40));
        Future<String> f8 = pool.submit(new MyCallable(50));
    
        // 三个核心线程都在忙, 任务队列满, 创建临时线程
        Future<String> f9 = pool.submit(new MyCallable(80));
        Future<String> f10 = pool.submit(new MyCallable(60));
    
        // 后续任务拒绝任务会被拒绝
        Future<String> f11 = pool.submit(new MyCallable(30));
    
        // 3. 互获取未来线程的返回结果
        try {
         // 正常的结果
            System.out.println(f1.get());
            System.out.println(f2.get());
            System.out.println(f3.get());
            System.out.println(f4.get());
            System.out.println(f5.get());
            System.out.println(f6.get());
            System.out.println(f7.get());
            System.out.println(f8.get());
            System.out.println(f9.get());
            System.out.println(f10.get());
            System.out.println(f11.get());
        } catch (Exception e) {
         // 异常的结果
            e.printStackTrace();
        }
    }

##### 方式二: Executors工具类创建线程池 #####

**Executors得到线程池对象的常用方法如下(*工具类的方法基本都是静态方法*)**:

> Executors：线程池的工具类通过调用方法返回不同类型的线程池对象。

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ExecutorService newCachedThreadPool()</td> 
   <td>线程数量随着任务增加而增加，如果线程任务执行完毕且空闲了一段时间则会被回收掉。</td> 
  </tr> 
  <tr> 
   <td><code>ExecutorService newFixedThreadPool(int nThreads)</code></td> 
   <td>创建固定线程数量的线程池，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程替代它(<em>保证线程存活数量</em>)。</td> 
  </tr> 
  <tr> 
   <td>ExecutorService newSingleThreadExecutor ()</td> 
   <td>创建只有一个线程的线程池对象，如果该线程出现异常而结束，那么线程池会补充一个新线程。</td> 
  </tr> 
  <tr> 
   <td>static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)</td> 
   <td>创建一个线程池，可以实现在给定的延迟后运行任务，或者定期执行任务。</td> 
  </tr> 
 </tbody> 
</table>

> **注意：Executors的底层其实也是基于线程池的实现类ThreadPoolExecutor创建线程池对象的。**
> 
> 例如下面代码当我们通过newFixedThreadPool方法创建线程池对象时
> 
>     ExecutorService pool = Executors.newFixedThreadPool(3);
> 
> 它的源码是创建了一个核心线程3, 最大线程3, 意味着没有临时线程, 所以临时线程存活时间也为0, 源码如下
> 
>     public static ExecutorService newFixedThreadPool(int nThreads) {
>          
>         return new ThreadPoolExecutor(nThreads, nThreads,
>                                       0L, TimeUnit.MILLISECONDS,
>                                       new LinkedBlockingQueue<Runnable>());
>     }

**演示代码**:

> 我们演示使用`newFixedThreadPool(int nThreads)`方法创建线程池.
> 
> 在主类中创建Runnable任务提交到线程池

public static void main(String[] args) {
        
        // 1. 创建线程池对象
        ExecutorService pool = Executors.newFixedThreadPool(3);
        // 2. 提交Runnable任务到线程池
        pool.execute(new MyRunnable());
        pool.execute(new MyRunnable());
        pool.execute(new MyRunnable());
        // 提交第四个任务就已经没有多余线程了, 会被拒绝
        pool.execute(new MyRunnable());
    }

> Executors执行Callable任务

public static void main(String[] args) throws Exception {
        
        // 1. 创建线程池对象
        ExecutorService pool = Executors.newFixedThreadPool(3);
    
        // 2. 提交Callable任务
        Future f1 = pool.submit(new MyCallable(10));
        Future f2 = pool.submit(new MyCallable(20));
        Future f3 = pool.submit(new MyCallable(30));
    
        // 获取任务返回结果
        System.out.println(f1.get());
        System.out.println(f2.get());
        System.out.println(f3.get());
    }

**Executors使用可能存在的陷阱**

> 大型并发系统环境中使用Executors如果不注意可能会出现系统风险。

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>存在问题</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ExecutorService newFixedThreadPool(int nThreads)</td> 
   <td>允许创建的线程是固定的, 但是线程允许请求的任务队列长度是无穷大的(Integer.MAX_VALUE)，可能出现OOM内存溢出错误（java.lang.OutOfMemoryError ）</td> 
  </tr> 
  <tr> 
   <td>ExecutorService newSingleThreadExecutor()</td> 
   <td>允许创建的线程是固定的, 允许请求的任务队列长度是无穷大的(Integer.MAX_VALUE)，可能出现OOM内存溢出错误（java.lang.OutOfMemoryError ）</td> 
  </tr> 
  <tr> 
   <td>ExecutorService newCachedThreadPool()</td> 
   <td>创建的线程数量最大上限是无穷大(Integer.MAX_VALUE)， 线程数可能会随着任务1:1增长，也可能出现OOM内存溢出错误 ( java.lang.OutOfMemoryError ）</td> 
  </tr> 
  <tr> 
   <td>ScheduledExecutorService newScheduledThreadPool(int corePoolSize)</td> 
   <td>创建的线程数量最大上限是无穷大(Integer.MAX_VALUE)， 线程数可能会随着任务1:1增长，也可能出现OOM内存溢出错误 ( java.lang.OutOfMemoryError ）</td> 
  </tr> 
 </tbody> 
</table>

> 阿里巴巴开发手册中明确说明, 不允许Executors创建线程池

![在这里插入图片描述][c16eafaa6efd46399c592ff57f7874fa.png_pic_center]

[c0842db7a2ce4926841ff7a668ed4a43.png_pic_center]: https://img-blog.csdnimg.cn/c0842db7a2ce4926841ff7a668ed4a43.png#pic_center
[c16eafaa6efd46399c592ff57f7874fa.png_pic_center]: https://img-blog.csdnimg.cn/c16eafaa6efd46399c592ff57f7874fa.png#pic_center