[Java EE] 多线程(六):线程池与定时器

1. 线程池

1.1 什么是线程池

我们前面提到,线程的创建要比进程开销小,但是如果线程的创建/销毁比较频繁,开销也会比较大.所以我们便引入了线程池,线程池的作用就是提前把线程都创建好,放到用户态代码中写的数据结构中,后面就可以随用随取.
线程池最大的好处就是减少每次启动,销毁线程的开销.线程的创建和销毁,需要通过用户态+内核态来配合完成,但是线程池只需要通过用户态即可完成,不需要内核态的配合.
在这里插入图片描述

举例说明:渣女小故事
有一位小姐姐是个渣女,创建和销毁线程就像一位小姐姐和他的男朋友分手了,再和下一位男朋友一点一点培养感情,而线程池就相当于,小姐姐有一个备胎池,他同时和好几个小哥哥培养感情,如果和其中一个分手了,就可以从备胎池中调用其他小哥哥,无缝衔接.

1.2 线程池的构造方法参数解释(面试常考)

线程池的类名是TreadPoolExecutor,其中他的构造方法中提供了丰富的参数类型,我们下面来一一解释一下这些参数:

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

corePoolSize:线程池中的核心线程数.
maximumPoolSize:线程池中的最大线程数.等于核心线程数+非核心线程数,其中非核心线程数是根据当前任务的多少来动态管理的.(至于最大线程要设置为多少,这个没有一个固定的值,一般认为,和CPU的逻辑核心数有关,可能是n+1,2n等)
keepAliveTime: 允许非核心线程数的最大空闲时间.超出最大空闲时间,就会被回收.
unit:等待时间的时间单位,时,分,秒.
workQueue:用于存放任务的阻塞队列.后续线程池内部的工作线程就会消费这个队列,从而完成任务的执行.
threadFactory:线程创建工厂,参与具体的线程创建工作.通过不同的工厂创建出的不同线程相当于对一些属性进行了不同的初始化设置.
针对线程工厂,我们又可以引出另一个话题:工厂设计模式.
这种设计模式主要是针对解决构造方法的不足而创建出的一种设计模式.
比如,我们给出了一个点(Point)类:
我们知道,一个点想要表示出来,一种是通过笛卡尔坐标系表示,一种是通过极坐标的方式来表示,但是我们会发现,笛卡尔坐标系的参数和极坐标系的参数个数相同,而且它们的参数类型也相同,这就使得我们不可以用构造方法来构造这个点.

public class Point {public double x = 0;public double y = 0;public double r = 0;public double a = 0;public Point(double x, double y) {this.x = x;this.y = y;}public Point(double r,double a){this.a = a;this.r = r;}
}

编译报错:
在这里插入图片描述
所以我们便引入了工厂设计模式来解决这个问题:
我们引入PointBuilder这个类,通过PointBuilder来构造这个点,并在Point这个点中设置set方法.

class Point {private double x = 0;private double y = 0;private double r = 0;private double a = 0;public void setX(double x) {this.x = x;}public void setY(double y) {this.y = y;}public void setR(double r) {this.r = r;}public void setA(double a) {this.a = a;}
}
class PointBuilder{public static Point pointXY(double x,double y){Point p = new Point();p.setX(x);p.setY(y);return p;}public static Point pointRA(double r,double a){Point p = new Point();p.setA(a);p.setR(r);return p;}
}

handler:任务超出线程池的承受范围的拒绝策略
- AbortPolicy():超过负荷,直接抛出异常.
- CallerRunsPolicy():调用者负责处理多出来的任务.
- DiscardOldestPolicy():丢弃队列中最老的任务.
- DiscardPolicy():丢弃新来的任务.

举例说明:年会不能停
有请老朋友:马杰克,潘妮,杰弗瑞,胡建林
corePoolSize:相当于公司中的正式员工,比如马杰克,杰弗瑞,胡建林.
maximumPoolSize:相当于公司中的所有员工,包括正式员工+外包.
keepAliveTime:允许外包员工的最大空闲时间,比如允许潘妮的最大空闲时间,一旦超过最大空闲时间,杰弗瑞就会把潘妮开除掉.
unit:允许潘妮的空闲时间单位,时,分,秒.
workQueue:每次员工们领取工作任务的账号.
threadFactory:公司不同的HR,通过HR招人(创建线程)
handler:员工们任务太多时候的拒绝策略
比如有一天杰弗瑞需要和胡建林要去干一场直播,但是胡建林由于工作太多,它需要拒绝杰弗瑞,其中胡建林有多种拒绝他的策略.第一种方法就是:胡建林直接罢工,工作和直播都不干了,就是直接罢工了.(AbortPolicy()).第二种方法就是:胡建林让杰弗瑞自己去直播(CallerRunsPolicy()).第三种方法就是:胡建林可以先放下手头的工作,和杰弗瑞去直播(DiscardOldestPolicy()).第四种方法就是:胡建林可以先完成自己的工作,之后在去和杰弗瑞直播(DiscardPolicy())

1.3 标准库中的线程池

Java源码的编写者也知道,上面这个方法用起来特别难,因为有许多参数.所以Java的标准库中提供了创建线程的一个工厂类Executors(本质上是Executor的封装),其中有一些创建线程的方法.这些方法的返回值可以直接当做线程池来使用,返回值的类型为ExecutorService.

线程池中的核心方法为submit,它是为线程池的阻塞队列中添加Runnable对象.
创建线程的工厂类Executor中右许多创建线程的方法,不同方法返回的线程属性各不相同.
- newFixedTreadPool(int nTread):创建固定线程数的线程池
- newCachedTreadPool():创建线程数目动态变化的线程池.(随任务数量变化)
- newSingleTreadExectuor():创建单线程的线程池.
- newScheduleTreadPool():设定延迟时间后执行指令,或者定期执行指令,是进阶版的Timer.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Demo24 {public static void main(String[] args) {ExecutorService service = Executors.newFixedThreadPool(10);//固定创建10个线程,注意返回类型for (int i = 0; i < 100; i++) {int finalI = i;service.submit(new Runnable() {@Overridepublic void run() {System.out.println("tread"+ finalI + Thread.currentThread().getName());}//只有固定的10个线程在执行这100个任务});}ExecutorService service1 = Executors.newCachedThreadPool();//随着任务的增多,线程创建增多for (int i = 0; i < 100; i++) {int finalI = i;service1.submit(new Runnable() {@Overridepublic void run() {System.out.println("tread"+ finalI + Thread.currentThread().getName());}});}}
}

创建固定线程线程池的运行结果:
在这里插入图片描述
我们看到,submit方法提供了100个任务,但是执行这100个任务的只有10个线程在执行.

下面是创建一个现场数量动态变化的线程池:
在这里插入图片描述
我们看到,线程池中被创建出了很多线程,100个任务由很多线程来执行.

1.4 实现线程池

首先我们要实现核心方法submit,将任务添加到线程池的队列中.
使用Worker类描述一个工作线程.使用Runnable描述一个任务.
使用BlockingQueue组织所有的任务.
每个Worker线程需要做的事情就是不断从BlockingQueue中获取任务并执行.
指定线程池的最大线程数.
submit方法时生产者,不断往队列中添加Runnable元素,而构造方法是消费者,不停地从队列中获取Runnable元素并执行.

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;/*** 创建固定线程的线程池*/
public class MyTreadPool {public BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();//通过阻塞队列来保存可运行对象//向阻塞队列提交任务,生产者public void submit(Runnable runnable) throws InterruptedException {queue.put(runnable);}//构造方法创建线程来执行队列中的任务.消费者public MyTreadPool(int nTread) {for (int i = 0; i < nTread; i++) {Thread thread = new Thread(()->{while (true){//每个线程不停地从队列中取出元素try {queue.take().run();//获取任务清单} catch (InterruptedException e) {throw new RuntimeException(e);}}});thread.start();//创建完线程之后立即启动}}public static void main(String[] args) throws InterruptedException {MyTreadPool myTreadPool = new MyTreadPool(10);for (int i = 0; i < 1000; i++) {int finalI = i;myTreadPool.submit(new Runnable() {@Overridepublic void run() {System.out.println("Tread"+ finalI + Thread.currentThread().getName());}});}}
}

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

2. 定时器

2.1 概念

指定一个任务(Runnable),并且指定一个时间,此时这个任务不会立即执行,而是在时间到达之后再执行.
在这里插入图片描述
定时器在我们日常的开发中也非常常见.比如在双11,0点开始定时抢购,再比如如果网络超过5000ms无响应的时候,就会尝试重新连接等.

2.2 标准库中的定时器

标准库中提供一个Timer类.Timer的核心方法为schedule,为Timer的队列中添加元素.
schedule包含两个参数,第一个参数指定将要执行的任务代码,第二个参数指定多长时间执行.

import java.util.Timer;
import java.util.TimerTask;public class Demo25 {public static void main(String[] args) {Timer timer = new Timer();timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello");}},3000);//延迟3s后再执行}
}

2.3 定时器的模拟实现

队列中使用优先级队列实现,把时间作为优先级队列的比较规则,delay时间短的放在堆顶.(注意不要使用PriorityBlockingQueue,容易死锁)
队列中的每一个元素是一个Task对象.
Task中带有一个时间属性.
同时有一个Worker线程一直扫描队首元素,通过与当前系统的时间戳去比较,看队首元素是否到达执行时间.
其次,由于我们使用的是优先级队列,其中一定存在线程安全问题,我们需要使用synchronized对其进行加锁.

import java.util.PriorityQueue;/*** 描述一个任务类*/
class TimerTask implements Comparable<TimerTask>{public Runnable runnable;public long time;public TimerTask(Runnable runnable, int delay) {this.runnable = runnable;this.time = System.currentTimeMillis()+delay;}@Overridepublic int compareTo(TimerTask o) {return (int)(this.time-o.time);//这里谁减谁不要记,试一试就知道了}
}
/*** 定时器*/
public class MyTimer {public PriorityQueue<TimerTask> priorityQueue = new PriorityQueue<>();/*** 该方法为优先级队列提供任务,生产者* @param runnable 可运行对象* @param delay 延时时间*/public void schedule(Runnable runnable,int delay){synchronized (this){//添加元素时加锁priorityQueue.offer(new TimerTask(runnable,delay));this.notify();//唤醒构造方法的wait}}/*** 创建线程执行队列中的任务,消费者*/public MyTimer() {Thread thread = new Thread(()->{while (true){//循环扫描堆顶元素,判断是否要执行synchronized (this){//由于优先级队列的不安全性,所以加锁if (priorityQueue.isEmpty()){try {this.wait();} catch (InterruptedException e) {throw new RuntimeException(e);}}long cur = System.currentTimeMillis();//注意保存当前时间,不可以写在判断时间中//否则每次都在改变TimerTask task = priorityQueue.peek();if (cur >= task.time){priorityQueue.poll().runnable.run();}else {try {this.wait(task.time-cur);//注意等待时间是任务时间减去当前是时间} catch (InterruptedException e) {throw new RuntimeException(e);}}}}});thread.start();//注意启动线程}public static void main(String[] args) {MyTimer myTimer = new MyTimer();myTimer.schedule(new Runnable() {@Overridepublic void run() {System.out.println("hello");}},2000);}
}