【多线程】线程池详解

缺乏、安全感 2023-03-03 11:21 41阅读 0赞

**目录**

一、线程池介绍

1.1 概念

1.2 为什么要使用线程池

1.3 执行机制 Executor

1.3.1 Executors 支持的四种线程池

1.3.2 为什么不建议使用 Executors?

二、ThreadPoolExecutor

2.1 ThreadPoolExecutor 概念

2.2 工作流程

2.3 使用示例

三、 ThreadPoolExecutor 基本参数

3.1 关键参数

3.2 参数详解

3.3 workQueue 的三种类型

3.3.1 有界队列

3.3.2 无界队列

3.3.3 直接握手队列

3.4 rejected tasks 饱和策略

3.5 其他方法

3.5.1 Hook methods 钩子方法

3.5.2 Queue maintenance 维护队列

3.5.3 Finalization 关闭

3.6 线程池状态

3.7 关闭线程池

3.7.1 原理

3.7.2 关闭方式

3.8 线程池的两种提交任务方式

3.8.1 execute 提交任务

3.8.2 submit 提交任务

3.8.3 execute 和 submit 比较

四、ThreadPoolExecutor 使用样例

4.1 无回调的任务处理

4.2 有回调的任务处理

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

# 一、线程池介绍 #

## 1.1 概念 ##

线程池介绍提前创建好多个线程放入池中，使用时直接获取，使用完放回池中。同时支持根据业务压力，根据配置实现临时工作线程的创建和管理

## 1.2 为什么要使用线程池 ##

由于于创建线程的代价和维护十分昂贵

*  JVM 中默认一个线程需要使用256k~1M的内存
 *  加重 GC 的回收压力

**使用线程池的好处：**

*  提高响应速度(减少了创建新线程的时间)
 *  降低资源消耗
 *  便于线程管理

## 1.3 执行机制 Executor ##

在 JDK 5 开始，将工作单元(单个线程) 和执行机制(线程运行和管理) 分离开，工作单元如 Runnable 和 Callable，而执行机制则是依赖 Executor

### 1.3.1 Executors 支持的四种线程池 ###

**● newFixedThreadPool **

![20200802123813872.png][]

**父类：**ThreadPoolExecutor

**描述：**创建一个固定核心线程的线程池，池内的所有的线程空闲时都不会消亡

**问题：**由于是固定线程池，所以设置的最大线程数，以及空闲存活时间参数无效。更大的问题是使用了无界队列，所以永远不会走饱和策略。

● **newSingleThreadExecutor**

![20200802124005607.png][]

**父类：**ThreadPoolExecutor

**描述：**初始化的线程池中只有一个线程，当线程异常结束时，会重新创建一个新的线程继续执行剩下的任务

**问题：**唯一线程可以保证提交任务的顺序性，但由于使用了无界队列，会导致饱和策略失效

● **newCachedThreadPool**

![20200802124214195.png][]

**父类：**ThreadPoolExecutor

**描述：**创建一个无限制线程数(integer.max\_value = 2^31 -1)的线程池，当有新的任务进来时，若没有空闲线程，会创建新的线程执行任务。若线程空闲时间超过 60s，则会被销毁

**问题：**长时间空闲的 CachedThreadPool 不会持有任何线程资源，但由于没有控制线程的数量，有oom(内存耗尽) 或 cpu 占满的风险

● **newScheduledThreadPool**

![20200802124214195.png][]

**父类：**ScheduledThreadPoolExecutor

**描述：**创建能够延迟执行，并定期循环执行的线程池，用于替代 timer 类的定时执行

### 1.3.2 为什么不建议使用 Executors? ###

若使用 Executors 的方法创建，有几个缺点：

*  封装了内部的参数，难以让使用者根据实际需求修改
 *  对线程池最大线程数给了 int 的最大值，可能导致内存溢出，或是 cpu 占满

# 二、ThreadPoolExecutor #

## 2.1 ThreadPoolExecutor 概念 ##

ThreadPoolExecutor 是 java 提供的创建线程池的对象，通过构造 ThreadPoolExecutor 可以快速地创建一个用有缓存队列、任务拒绝策略以及线程控制的线程池。

## 2.2 工作流程 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70][]

## 2.3 使用示例 ##

public ThreadPoolExecutor(
                int corePoolSize, // 
                int maximumPoolSize,  // 2
                long keepAliveTime,  // 3
                TimeUnit unit,  // 4
                BlockingQueue<Runnable> workQueue, // 5
                ThreadFactory threadFactory,  // 6
                RejectedExecutionHandler handler ) { //7
            if (corePoolSize < 0 ||
                    maximumPoolSize <= 0 ||
                    maximumPoolSize < corePoolSize ||
                    keepAliveTime < 0)
                throw new IllegalArgumentException();
            if (workQueue == null || threadFactory == null || handler == null)
                throw new NullPointerException();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

# 三、 ThreadPoolExecutor 基本参数 #

## 3.1 关键参数 ##

<table> 
 <tbody> 
  <tr> 
   <td style="width:170px;"><strong>参数名称</strong></td> 
   <td style="width:227px;"><strong>类型</strong></td> 
   <td style="width:498px;"><strong>描述</strong></td> 
  </tr> 
  <tr> 
   <td style="width:170px;">corePoolSize</td> 
   <td style="width:227px;">int</td> 
   <td style="width:498px;">核心线程池大小，核心线程即被创建后就不会销毁，一直等待任务的线程</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">maximumPoolSize</td> 
   <td style="width:227px;">int</td> 
   <td style="width:498px;">线程池最大线程数，当线程数量达到这个数值就不会再增长。</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">keepAliveTime</td> 
   <td style="width:227px;">int</td> 
   <td style="width:498px;">空闲线程存活时间</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">unit</td> 
   <td style="width:227px;">TimeUnit</td> 
   <td style="width:498px;">时间单位</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">workQueue</td> 
   <td style="width:227px;">BlockingQueue&lt;Runnable&gt;</td> 
   <td style="width:498px;">线程等待队列，用于当新的任务到来却没有空闲线程能够处理时，先缓存这些未处理的任务，等待空闲线程处理</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">threadFactory</td> 
   <td style="width:227px;">ThreadFactory</td> 
   <td style="width:498px;">线程创建工厂，若没有指定，则会使用默认的defaultThreadFactory工厂创建。在这里你可以对创建的线程设置名称、优先级，守护进程状态等</td> 
  </tr> 
  <tr> 
   <td style="width:170px;">handler</td> 
   <td style="width:227px;">RejectedExecutionHandler</td> 
   <td style="width:498px;">拒绝策略，当线程被关闭，或是任务队列已满时触发</td> 
  </tr> 
 </tbody> 
</table>

## 3.2 参数详解 ##

**● corePoolSize 核心线程数**

核心线程即在线程池日常维护，不让其因为超时而消亡的线程。保持一定的核心线程数目可以让少量的请求进入时立刻得到处理，提高线程池处理速度。同时，由于每个线程都需要消耗资源(内存)，故核心线程数目应该适当

**● maximumPoolSize 线程池最大线程数**

最大线程数 = 核心线程数 + 临时线程数，当任务进入时，会先判断当前存活的线程数是否满足核心线程数，若不满足则创建核心线程，并处理任务。若满足，则会尝试创建工作线程。空闲的工作线程会在超出存活时间后被销毁

**● workQueue 任务等待队列/缓冲区**

线程等待队列，用于当新的任务到来却没有空闲线程能够处理时，先缓存这些未处理的任务，等待空闲线程处理。

## **3.3 workQueue 的三种类型** ##

### **3.3.1 有界队列** ###

常用的有界队列为 ArrayBlockingQueue，也可能是限定了范围的其他队列，创建方式如下：

BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(50000);
    
    # 也可能是限定了范围的其他队列，如
    BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(50000);

**优点：**

设置有界队列的最大数量，能够防止资源耗尽。在生产环境中，需要合理调整有界队列的最大数量以及 maximunPoolSizes，达到任务处理和任务缓存的平衡，降低任务被拒绝的风险。

**两种情况：**

<table> 
 <tbody> 
  <tr> 
   <td style="width:232px;"><strong>场景</strong></td> 
   <td style="width:656px;"><strong>结果</strong></td> 
  </tr> 
  <tr> 
   <td style="width:232px;">大队列，小 maximumPoolSizes</td> 
   <td style="width:656px;"> <p><strong>现象：</strong>较大的缓存队列，以及较少的并发线程数目</p> <p><strong>好处：</strong>可以最大限度地减少cpu的使用率，降低占用的系统资源，以及上下文切换开销。</p> <p><strong>坏处：</strong>吞吐量降低</p> </td> 
  </tr> 
  <tr> 
   <td style="width:232px;">小队列，大&nbsp;maximumPoolSizes</td> 
   <td style="width:656px;"> <p><strong>现象：</strong>较小的缓存队列，以及较多的并发线程数目</p> <p><strong>好处：</strong>提高线程池的吞吐量，处理效率较高</p> <p><strong>坏处：</strong>系统开销增大，cpu 繁忙，若遇到耗时较多的任务，会导致拒绝的情况出现</p> </td> 
  </tr> 
 </tbody> 
</table>

**结论：**

应根据实际情况合理调整缓冲队列大小，已经最大并发线程数

### **3.3.2 无界队列** ###

常用的无界队列为没有预定容量的 LinkedBlockingQueue，创建方式如下：

BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();

**优点：**

由于没有边界限制，缓冲队列能够存储更多的任务，可以平滑顺时大量请求。

**劣势：**

由于没有限制队列大小，其所占用的内存空间不可预知，有内存溢出的风险

同时当任务平均提交速度大于平均处理速度时，有队列无限增长的风险

由于队列无限增长，不会触发拒绝策略

**结论：**

在处理大量并发时，无界队列效果优于有界队列，但要注意不要发生OOM

### **3.3.3 直接握手队列** ###

Direct handoffs 的一个很好的默认选择是 SynchronousQueue，它将任务交给线程而不需要保留。这里，如果没有线程立即可用来运行它，那么排队任务的尝试将失败，因此将构建新的线程。  
此策略在处理可能具有内部依赖关系的请求集时避免锁定。Direct handoffs 通常需要无限制的maximumPoolSizes来避免拒绝新提交的任务。 **但得注意，当任务持续以平均提交速度大余平均处理速度时，会导致线程数量会无限增长问题。**

## 3.4 **rejected tasks 饱和策略** ##

饱和策略有两种情况：线程池已经被关闭，或是任务队列已满且 maximunPoolSizes 已满，无论哪种情况，都会调用到 rejectedExecution 方法。其内部预定义了4种处理策略，同时也支持我们自定义策略

**● 自定义策略**

通过继承 RejectedExecutionHandler，重写 rejectedExecution 方法，实现自定义拒绝策略

public static class MyIgnorePolicy implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            doLog(r, e);
        }
    
        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            System.err.println( r.toString() + " rejected");
        }
    }

**● AbortPolicy 异常抛出策略**

触发 reject 后会直接抛出异常，会导致队列处理终止

![20200731164532656.png][]

**● CallerRunsPolicy 反馈控制策略**

可以减慢提交新任务的速度，触发 reject 后会减慢新任务提交的速度。被拒绝的任务会被重新提交

**● DiscardPolicy 丢弃策略**

直接丢弃新提交的任务

**● DiscardOldestPolicy 丢弃旧任务策略**

如果执行器没有关闭，队列头的任务将会被丢弃，然后执行器重新尝试执行任务(如果失败，则重复这一过程)

## 3.5 其他方法 ##

### 3.5.1 **Hook methods 钩子方法** ###

ThreadPoolExecutor为提供了每个任务执行前后提供了钩子方法，重写`beforeExecute(Thread，Runnable)`和`afterExecute(Runnable，Throwable)`方法来操纵执行环境； 例如，重新初始化ThreadLocals，收集统计信息或记录日志等。此外，`terminated()`在Executor完全终止后需要完成后会被调用，可以重写此方法，以执行任殊处理。  
**注意：**如果hook或回调方法抛出异常，内部的任务线程将会失败并结束。

### 3.5.2 **Queue maintenance 维护队列** ###

`getQueue()`方法可以访问任务队列，一般用于监控和调试。绝不建议将这个方法用于其他目的。当在大量的队列任务被取消时，`remove()`和`purge()`方法可用于回收空间。

### 3.5.3 **Finalization 关闭** ###

如果程序中不在持有线程池的引用，并且线程池中没有线程时，线程池将会自动关闭。如果您希望确保即使用户忘记调用 `shutdown()`方法也可以回收未引用的线程池，使未使用线程最终死亡。那么必须通过设置适当的 keep-alive times 并设置allowCoreThreadTimeOut(boolean) 或者 使 corePoolSize下限为0 。  
`一般情况下，线程池启动后建议手动调用shutdown()关闭。`

## 3.6 线程池状态 ##

线程池使用一个 AtomicInteger 的 ctl 变量将 workerCount（工作线程数量）和 runState（运行状态）两个字段压缩在一起。ThreadPoolExecutor用3个比特位表示runState(2^3=8 > 5 种运行状态)， 29个比特位表示workerCount

整个ctl的状态，会在线程池的不同运行阶段进行CAS转换

<table> 
 <tbody> 
  <tr> 
   <td style="width:181px;"><strong>状态</strong></td> 
   <td style="width:639px;"><strong>解释</strong></td> 
  </tr> 
  <tr> 
   <td style="width:181px;"><strong>RUNNING</strong></td> 
   <td style="width:639px;">运行态，可以处理新任务，并执行队列中的任务</td> 
  </tr> 
  <tr> 
   <td style="width:181px;"><strong>SHUTDOWN</strong></td> 
   <td style="width:639px;">关闭态，不接受新任务，但可以处理队列中的任务</td> 
  </tr> 
  <tr> 
   <td style="width:181px;"><strong>STOP</strong></td> 
   <td style="width:639px;">停止态，不接受新任务，也不处理队列中的任务，且打断运行中的任务</td> 
  </tr> 
  <tr> 
   <td style="width:181px;"><strong>TIDYING</strong></td> 
   <td style="width:639px;">整理态，所有任务都已经结束，workerCount = 0，将执行 terminated() 方法</td> 
  </tr> 
  <tr> 
   <td style="width:181px;"><strong>TERMINATED</strong></td> 
   <td style="width:639px;">结束态，terminated() 方法已完成</td> 
  </tr> 
 </tbody> 
</table>

![这里写图片描述][SouthEast]

## 3.7 关闭线程池 ##

### 3.7.1 原理 ###

关闭线程池即便遍历线程池中的所有线程，逐个调用线程的 interrupt() 方法来中断线程

### 3.7.2 关闭方式 ###

● shutdown

将线程池里的线程状态设置成 shutdown 状态，但不会中断正在执行任务的线程，而是等待执行完毕

● shutdownNow

将线程池里的线程状态设置成 stop状态，同时会将所有正在执行或暂停任务的线程停止

只要这两个方法任意一个被调用，isShutDown() 会返回 true

当所有的任务线程都关闭，isTerminated() 会返回 true

## 3.8 线程池的两种提交任务方式 ##

线程池提交任务有两种方式 execute() 和 submit()

### 3.8.1 execute 提交任务 ###

![20200802152825944.png][]

可以看出,execute 只能使用继承了 Runnable 接口的任务，且没有返回值

### 3.8.2 submit 提交任务 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70 1][]

submit 支持 Runnable 方式和 Callable 方式提交任务，支持回调方式处理。且会返回一个 Future 响应结果。

### 3.8.3 execute 和 submit 比较 ###

*  **支持类型和返回值不同：**execute 只能提交 Runnable 类型的任务，无返回值。但 submit 同时支持 Runnable 和 Callable 类型任务，并返回 Future。当任务类型为 Runnable 时，返回值为 null
 *  **抛出异常时机不同：**execute 在执行任务时，如果遇到异常会直接抛出，而 submit 不会直接抛出，而是通过 Future 的 get 方法获取返回值时，才会抛出异常

# 四、ThreadPoolExecutor 使用样例 #

## 4.1 无回调的任务处理 ##

public class ThreadPoolTest {
    
        public static void main(String[] args) throws IOException {
            // 核心线程数
            int corePoolSize = 20;
            // 最大线程数
            int maximumPoolSize = 20;
            // 空闲线程存活时间
            long keepAliveTime = 10;
            TimeUnit unit = TimeUnit.SECONDS;
    
            BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(50000);
            ThreadFactory threadFactory = new NameTreadFactory();
            RejectedExecutionHandler handler = new MyIgnorePolicy();
    
            ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                    workQueue, threadFactory, handler);
    
            // 预热，会启动所有核心线程，若没有预热核心线程会在任务到达时创建
            executor.prestartAllCoreThreads();
    
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            for (int i = 0; i < 50000; i++) {
                MyTask myTask = new MyTask(String.valueOf(i));
                executor.execute(myTask);
            }
    
            System.in.read();
        }
    
        static class NameTreadFactory implements ThreadFactory {
    
            private final AtomicInteger mThreadNum = new AtomicInteger(1);
    
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
                System.out.println(t.getName() + " has been created");
                return t;
            }
        }
    
        public static class MyIgnorePolicy implements RejectedExecutionHandler {
    
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                doLog(r, e);
            }
    
            private void doLog(Runnable r, ThreadPoolExecutor e) {
                // 可做日志记录等
                System.err.println( r.toString() + " rejected");
            }
        }
    
        static class MyTask implements Runnable {
            private String name;
    
            public MyTask(String name) {
                this.name = name;
            }
    
            public void setName(String name) {
                this.name = name;
            }
    
            @Override
            public String toString() {
                return "MyTask{" +
                        "name='" + name + '\'' +
                        '}';
            }
    
            @Override
            public void run() {
                System.out.println(this.toString() + "is running!");
            }
        }
    }

## 4.2 有回调的任务处理 ##

package com.tom;
    
    import java.io.IOException;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class ThreadPoolCallTest {
    
        public static void main(String[] args) throws IOException {
            // 核心线程数
            int corePoolSize = 20;
            // 最大线程数
            int maximumPoolSize = 20;
            // 空闲线程存活时间
            long keepAliveTime = 10;
            TimeUnit unit = TimeUnit.SECONDS;
    
            BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(50000);
            ThreadFactory threadFactory = new NameTreadFactory();
            RejectedExecutionHandler handler = new MyIgnorePolicy();
    
            ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                    workQueue, threadFactory, handler);
    
            // 预热，会启动所有核心线程，若没有预热核心线程会在任务到达时创建
            executor.prestartAllCoreThreads();
    
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            for (int i = 0; i < 5000; i++) {
                MyCallTask myTask = new MyCallTask(String.valueOf(i));
                Future<?> f = executor.submit(myTask);
                    try {
                        System.out.println(f.get());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
            }
    
            System.in.read();
        }
    
        static class NameTreadFactory implements ThreadFactory {
    
            private final AtomicInteger mThreadNum = new AtomicInteger(1);
    
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
    //            System.out.println(t.getName() + " has been created");
                return t;
            }
        }
    
        public static class MyIgnorePolicy implements RejectedExecutionHandler {
    
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                doLog(r, e);
            }
    
            private void doLog(Runnable r, ThreadPoolExecutor e) {
                // 可做日志记录等
                System.err.println( r.toString() + " rejected");
            }
        }
    
        static class MyCallTask implements Callable<String> {
            private String name;
    
            public MyCallTask(String name) {
                this.name = name;
            }
    
            public void setName(String name) {
                this.name = name;
            }
    
            @Override
            public String call() throws Exception {
                return "threadName: "+ Thread.currentThread().getName() + "  call: " + name;
            }
        }
    }

[20200802123813872.png]: /images/20230208/68477c2d45eb4cc0bb611684df946a49.png
[20200802124005607.png]: /images/20230208/84e4ec35f82c49c7ad8224bcbcfb1765.png
[20200802124214195.png]: /images/20230208/ca2a742a330b4538b34fbb3763683266.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70]: /images/20230208/7f5a248750af4437a744dfdf87739866.png
[20200731164532656.png]: /images/20230208/100bc7cbd68640d9a1436150ba47a277.png
[SouthEast]: /images/20230208/f2685e07ec684135b501cad015cb9eeb.png
[20200802152825944.png]: /images/20230208/eb9cfabde14f4ded8d3a27d177f40294.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70 1]: /images/20230208/8f034040e6f946bba04aa815694df65a.png