【多线程】线程池(上)

线程池基本概念

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

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

池化技术不仅在线程池中体现,还在数据库连接池,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{@Overridepublic 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{@Overridepublic 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{@Overridepublic 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{@Overridepublic 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被抛弃
运行结果: