【多线程】线程池(上)

文章目录

线程池基本概念
- 线程池的优点
- 线程池的特点
创建线程池
- 自定义线程池
- 线程池的工作原理
- 线程池源码分析
- 内置线程池
- - `newFixedThreadPool`
  - `SingleThreadExecutor`
  - `newCachedThreadPool`
  - `ScheduledThreadPool`
线程池的核心线程是否会被回收?
拒绝策略
- ThreadPoolExecutor.AbortPolicy
- ThreadPoolExecutor.CallerRunsPolicy
- ThreadPoolEcecutor.DiscardPolicy
- ThreadPoolExecutor.DiscardOldestPolicy

线程池基本概念

在讲解线程池之前,我们应该先来了解池化技术

池化技术:
池化技术是一种用于管理和复用线程的技术，旨在提高系统的性能和资源利用率。线程池通过预先创建一组线程来处理任务，从而避免频繁地创建和销毁线程带来的开销。

池化技术不仅在线程池中体现,还在数据库连接池,HTTP连接池,字符串常量池中均有体现.

如果不使用池化技术,我们创建一个线程的步骤:

手动创建线程对象
执行任务
执行完毕,释放线程对象
此处,每一次创建线程资源和释放线程资源,均会消耗系统资源,而如果我们采用线程池的方式,由线程池统一创建和管理线程资源,就可以降低系统资源的消耗,提高对资源的利用率.

所以线程池,提供一种限制和管理资源的方式.

线程池的优点

降低资源消耗:通过重复利用已经创建的线程降低线程创建和销毁造成的资源消耗.
提高程序的响应速度:当任务到达时,任务可以不需要等待线程的创建就可以立即执行
便于统一管理线程对象:线程池可以统一进行线程分配
可以控制最大的并发数(线程池会限制最大的线程对象的个数,这样可以限制并发量,避免无限制的创建线程资源,造成系统资源的损耗)

线程池的特点

线程复用
控制最大并发数
管理线程

创建线程池

自定义线程池

对于线程池的管理和使用,我们使用在java.util.concurrent下的ThreadPoolExecutor线程池工具类
来进行线程池的创建和管理

public class ThreadPoolExecutor extends AbstractExecutorService{
    //成员属性:
    private volatile int corePoolSize;//核心线程数
    private volatile int maximumPoolSize;//最大线程数
    private volatile long keepAliveTime;//非核心线程的最大存活时间
    private final BlockingQueue<Runnable> workQueue;//工作队列
    private volatile ThreadFactory threadFactory;//线程工厂
    private volatile RejectedExecutionHandler handler;//拒绝策略
   //...
    
    //线程池的构造方法:
    public ThreadPoolExecutor(int corePoolSize, 
                              int maximumPoolSize, 
                              long keepAliveTime, 
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler
                              )
   
   //...                           
  }

参数:

corePoolSize:任务队列未达到队列最大容量时,最大可以同时运行的线程数量
maximumPoolSize:任务队列中存放的任务达到队列容量的时候,当前可以同时运行的线程数量变为最大线程数
workQueue:新任务来的时候,会先判断当前运行的线程数量是否达到核心线程数,如果达到的话,新任务就会被存放在队列中
keepAliveTime:当线程池中的线程数量大于corePoolSize,即有非核心线程(线程池中核心线程以外的线程)时,这些非核心线程空闲后不会立即被销毁,而是会等待,直到等待的时间超过了keepAliveTime才会被回收销毁
Unit:keepAliveTime参数的时间单位
threadFactory:executor创建新线程时会用到
headler:拒绝策略(后续详解))

接下来我们采用图解的方式,来系统讲解一下这里各种参数具体使用,并了解线程池的工作原理.

线程池的工作原理

我们下述线程池原理的讲解,以一个corePoolSize(核心线程数)为9,maximumPoolSize(最大线程数)为13,workQueue(工作队列)大小为5,keepAliveTime(非核心线程存活时间)为10s来举例.

当此时的线程池中已被创建的线程<核心线程数,线程池会创建新的线程执行任务
此时的线程池中已创建的线程池数==核心线程数
此时新的任务进入线程池中,会先进入阻塞队列中进行等待,核心线程中如果存在空闲的线程,就会去阻塞队列中取任务进行执行
如果工作阻塞队列已满,核心线程也在执行任务,但最大线程数>已创建线程数,那么进入的任务后,线程池会创建新的线程,直到最大线程数==已创建线程数.
在线程池中工作阻塞队列已满,且已创建线程数==最大线程数时,此时再来任务时,线程池就会触发拒绝策略.
当线程池中的额外创建的线程空闲下来了,是否马上就会销毁呢?
并不是,线程池会根据设定的keepAliveTime时间来判断什么时候销毁这些线程.
那么是不是一定要销毁原本创建的额外创建的线程呢?
并不是,线程池只需要维护核心线程数和最大线程数的数量即可,所以线程池中的线程并没有一个明确的身份

其中的几个线程空闲时间超过了keepAliveTime时间后:

以上就是线程池的工作原理和参数的理解.

线程池源码分析

我们从execute入手,了解线程池的源码中是如何实现上述过程的.
```
public void execute(Runnable command) {
     if (command == null)
         throw new NullPointerException();
             int c = ctl.get();
     if (workerCountOf(c) < corePoolSize) {
         if (addWorker(command, true))
             return;
         c = ctl.get();
     }
     if (isRunning(c) && workQueue.offer(command)) {
         int recheck = ctl.get();
         if (! isRunning(recheck) && remove(command))
             reject(command);
         else if (workerCountOf(recheck) == 0)
             addWorker(null, false);
     }
     else if (!addWorker(command, false))
         reject(command);
 }
```
接下来我们来观察一下addWorker方法里是如何创建一个新的线程的

private boolean addWorker(Runnable firstTask, boolean core) {
//......
 Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;

在这里插入图片描述
我们发现,最终是调用了ThreadFactory参数来创建线程的.

这里还需要注意Worker类是实现了Runnable接口的类,所以Worker本身也是一个Runnable,所以创建线程是,调用的this.thread=getThreadFactory().newThread(this)传入的this对象就是Worker,所以线程执行的run()方法,是Worker内部的run()方法,而调用内部的run()方法后,在调用runWorker(this)方法,我们来看一看runWorker()方法中的内容:

我们看到,这里传入了参数this对象,其中this中的成员属性firstTask就是我们传入的FirstTask

在这里我们看到了task.run()的身影,说明此处执行了我们传入的任务,并且在执行任务之前和执行任务之后还可以调用相关的方法

接下来我们来观察一下是如何将任务加入到队列中的.
在这里插入图片描述
底层的工作队列是阻塞队列,任务会通过offer方法加入到队列中,此时如果队列中没有任务,线程来队列中获取任务时,就会阻塞在这个位置,直到有任务加入到队列中.所以使用阻塞队列,也实现了核心线程的保活

内置线程池

除了使用JUC包下的ThreadPoolExecutor来创建线程,我们还可以使用JUC包下的工具类Executor来创建具有一定固定功能的线程池
建立一个任务(后面的几个实验也是使用此任务执行)

      class MyRunnable implements Runnable{
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName()+"执行任务");
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

`newFixedThreadPool`

newFixedThreadPool:创建固定线程数量的线程池

使用newFixedThreadPool创建一个线程数为4的线程池

public class test1 {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService threadPool= Executors.newFixedThreadPool(4);
        for(int i=0;i<10;i++){
            threadPool.execute(new MyRunnable());
        }
        TimeUnit.SECONDS.sleep(15);

        threadPool.shutdown();

        System.out.println(threadPool.isShutdown());
    }
}

运行结果:
在这里插入图片描述
我们发现,这里会重复利用4个线程,不会创建多于4的线程.

`SingleThreadExecutor`

SingleThreadExecutor:只有一个线程的线程池

public class test2 {
    public static void main(String[] args) throws InterruptedException {

        ExecutorService threadPool= Executors.newSingleThreadExecutor();
        for(int i=0;i<10;i++){
            threadPool.execute(new MyRunnable());
        }
        TimeUnit.SECONDS.sleep(15);
        threadPool.shutdown();

    }
}

运行结果:
在这里插入图片描述

`newCachedThreadPool`

newCachedThreadPool:可以根据实际情况动态调整线程数量的线程池

public class test3 {
    public static void main(String[] args) {
        ExecutorService threadPool= Executors.newCachedThreadPool();
        for(int i =0;i<10;i++){
            threadPool.execute(new MyRunnable());
        }

        threadPool.shutdown();
    }
}

运行结果:
在这里插入图片描述
我们发现,此时线程池选择创建十个线程来执行这十个任务

`ScheduledThreadPool`

ScheduledThreadPool:给定的延迟后运行任务或者定期执行任务的线程池

线程池的核心线程是否会被回收?

ThreadPoolExecutor默认不会回收核心线程,即使他们已经空闲了.这是为了减少创建线程的开销,因为核心线程通常是要长期保持活跃的.
但如果线程池是用于周期性使用的场景(也就是线程池忙碌-空闲存在周期性),可以使用allowCoreThreadTimeOut(boolean value)方法的参数设置为true,这样就会回收空闲的核心线程了

拒绝策略

我们在阅读源码的过程中,已经关注到了拒绝策略这个问题.也就是说:当线程池已经达到最大并发量的时候,我们需要对新来的任务进行"拒绝".那么线程池中的拒绝策略有哪些呢?

ThreadPoolExecutor.AbortPolicy

抛出RejectedExecutionException来直接拒绝新任务的处理(这里线程池默认的拒绝策略)

public class test5 {
    public static void main(String[] args) {

        BlockingQueue workQueue=new ArrayBlockingQueue(1);

        ExecutorService threadPool= new ThreadPoolExecutor(1,2,1L,TimeUnit.SECONDS,workQueue);

        for(int i=0;i<10;i++){
            threadPool.execute(new MyRunnable());
        }
        threadPool.shutdown();

    }
}

分析代码:
此处的场景中,线程池的最大并发量是3,但我们需要执行十个任务,那么一定会触发默认的拒绝策略.
运行结果:

我们看到,这里执行了三个任务,就在主线程中抛出了RejectedExecutionException

ThreadPoolExecutor.CallerRunsPolicy

这个是使用调用execute方法的线程来运行被拒绝的任务,如果执行execute方法的线程已关闭,则会丢弃该任务.

public static void main(String[] args) {                                                                                                                                             
                                                                                                                                                                                     
    BlockingQueue workQueue=new ArrayBlockingQueue(1);                                                                                                                               
                                                                                                                                                                                     
    ThreadPoolExecutor threadPool= new ThreadPoolExecutor(1,2,1L,TimeUnit.SECONDS,workQueue,new ThreadPoolExecutor.CallerRunsPolicy());                                              
                                                                                                                                                                                     
    for(int i=0;i<10;i++){                                                                                                                                                           
        threadPool.execute(new MyRunnable());                                                                                                                                        
    }                                                                                                                                                                                
    threadPool.shutdown();                                                                                                                                                           
                                                                                                                                                                                     
}                                                                                                                                                                                    
}

分析代码:
这里我们创建了一个最大并发数为3的线程池,并且将拒绝策略设定为ThreadPoolExecutor.CallerRunsPolicy,此时我们在运行过程中,肯定会存在线程达到了最大并发数,触发拒绝策略的情况.而ThreadPoolExecutor.CallerRunsPolicy会将需要执行的任务退回给调用者(即调用execute方法的线程,在此处是main线程),由调用者线程来执行该任务,这样就能够保证所有任务都不会被抛弃,而是被执行

运行结果:
在这里插入图片描述
我们看到这里mian方法执行了任务

ThreadPoolEcecutor.DiscardPolicy

不处理新任务,直接丢掉

public class test6 {
    public static void main(String[] args) {

        BlockingQueue workQueue=new ArrayBlockingQueue(1);

        ThreadPoolExecutor threadPool= new ThreadPoolExecutor(1,2,1L, TimeUnit.SECONDS,workQueue,new ThreadPoolExecutor.DiscardPolicy());

        for(int i=0;i<10;i++){
            threadPool.execute(new MyRunnable());
        }
        threadPool.shutdown();

    }
}

运行结果:
在这里插入图片描述
这里我们发现,线程池只执行了三个任务,就结束了,并没有抛出异常或者将任务退回给execute的调用者.

ThreadPoolExecutor.DiscardOldestPolicy

此策略将丢弃掉阻塞队列中最早的任务.
我们来进行一个实验:

public class test7 {
    private  static int i=1;
    static class MyRunnable_1 implements Runnable{
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"执行任务1");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    static class MyRunnable_2 implements Runnable{
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"执行任务2");
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    static class MyRunnable_3 implements Runnable{
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"执行任务3");
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue workQueue=new ArrayBlockingQueue(1);

        ThreadPoolExecutor threadPool= new ThreadPoolExecutor(1,1,1L, TimeUnit.SECONDS,workQueue,new ThreadPoolExecutor.DiscardOldestPolicy());

            threadPool.execute(new MyRunnable_1());
            threadPool.execute(new MyRunnable_2());
            threadPool.execute(new MyRunnable_3());

        threadPool.shutdown();
    }
}

分析代码:
这里我们创建了三个任务,分别是MyRunnable_1,MyRunnable_2,MyRunnable_3,我们设定线程池的最大并发数为2,即最多线程数为1,阻塞队列为1,这样三个任务执行的时候,首先会执行第一个任务,而第一个任务需要10s,第二个任务进入阻塞队列中进行等待,第三个任务进入触发拒绝策略,所以会将第二个任务抛弃掉.所以代码最终呈现的结果应该是:任务1,3被执行,任务2被抛弃
运行结果: