当前位置：首页 > news >正文

多线程（4）——单例模式、阻塞队列、线程池、定时器

news 来源：原创 2024/9/20 14:41:12

1. 多线程案例

1.1 单例模式

单例模式能保证某个类在程序中只存在唯一一份实例，不会创建出多个实例（这一点在很多场景上都需要，比如 JDBC 中的 DataSource 实例就只需要一个

tip：设计模式就是编写代码过程中的 “软性约束”，不是强制的；框架就是编写代码过程中的 “硬性约束”，针对一些特定的场景问题，基本的代码逻辑是固定的

单例模式具体的实现方式有很多，最常见的是 “饿汉” 和 “懒汉” 两种

1.1.2 饿汉模式

//创建一个单例的类，饿汉方式实现
//饿 的意思是 “迫切”
//在类被加载的时候，就会创建出这个单例的实例
class Singleton {private static Singleton instance = new Singleton();public static Singleton getInstance() {return instance;}//单例模式的最关键部分private Singleton() { }
}
public class Demo24 {public static void main(String[] args) {Singleton s1 = Singleton.getInstance();Singleton s2 = Singleton.getInstance();System.out.println(s1 == s2);}
}

其中的构造方法用 private 修饰，意味着在类的外面就无法调用构造方法，也就无法创建实例了

只要不在其他代码中 new 这个类，每次使用都通过 getInstance 来获取实例，这个类就是单例的了

单例主要解决的问题就是防止别人不小心 new 了对象

tip：

单例模式只能避免失误，不能应对 “故意攻击”，如：使用反射或序列化反序列化能打破上述单例模式

单例模式的前提是 “一个进程中”，如果有多个 Java 进程，自然是每个进程中都可以有一个实例了

1.1.3 懒汉模式

//懒汉模式实现的单例模式
class SingletonLazy {//此处先把这个实例的引用设为 null，不着急创建实例private static SingletonLazy instance = null;public static SingletonLazy getInstance() {if (instance == null) {instance = new SingletonLazy();}return instance;}private SingletonLazy() { }
}
public class Demo25 {public static void main(String[] args) {SingletonLazy s1 = SingletonLazy.getInstance();SingletonLazy s2 = SingletonLazy.getInstance();System.out.println(s1 == s2);}
}

计算机中谈到的 “懒” 是褒义词，意思是，效率会更高，懒汉模式推迟了创建实例的时机，第一次使用的时候才会创建实例

上述代码中，当首次调用 getInstance，由于此时引用为 null，就会进入 if 分支创建实例，后续再重复调用 getInstance 就不会创建实例了，直接返回

1.1.4 饿汉模式和懒汉模式的线程安全问题

观察饿汉模式的代码，发现其只有读操作，不涉及到修改操作，所以没有线程安全问题

而懒汉模式中有 if 判定和其中的修改操作，这种代码模式是典型的线程不安全代码，因为判定和修改之间可能涉及到线程的切换，如下图：

上述例子中就创建了两个实例，虽然第二次创建覆盖了第一次的值，使得第一次创建的实例没有引用指向，很快就会被垃圾回收机制给消除掉，但是仍然认为上述代码是存在 bug 的

tip：在实际场景中，构造方法内部可能会执行很多逻辑，假设现有 100G 数据，加载到内存中需要 10 分钟，若是上述代码构造实例来管理加载数据到内存中，耗时就会翻倍成 20 分钟

1.1.5 通过加锁来解决懒汉模式中的线程安全问题

问题一：

上面加锁之后确实解决了线程安全问题，但是当已经 new 完对象后，if 分支就再也进不去了，后续的代码就应该是单纯的读操作，此时 getInstance 不加锁也是线程安全的

问题二：

但是当前代码的写法只要调用 getInstance 都会触发加锁操作，虽然没有线程安全问题了，但是会因为加锁产生阻塞，影响到性能

通过加一个条件判断，改进该问题：

tip：

创建的局部变量，处于 JVM 内存的 “栈” 区域中；new 出来的对象，处于 JVM 内存的 “堆” 区域中

对于整个 JVM 进程来说，堆是只有一份，线程之间公用的；栈则是每个线程有自己独立的（这是 Java 语法的限制，t1 无法访问 t2 栈上的变量，C++、系统原生 API 中不存在这样的限制，任何一个变量都是可以给其他线程用的）

正是因为变量的共享是常态，所以就容易触发多个线程修改同一个变量，从而引起线程安全问题

问题三：

该代码还可能会因为指令重排序，引起线程安全问题（指令重排序也是一种编译器的优化方式）

编译器可能会按照 1 2 3 的顺序来执行，也可能按照 1 3 2 的顺序来执行，对于单线程来说，先执行 2 和先执行 3 本质上是一样的，但是在多线程的环境下，按照 1 3 2 的顺序来执行可能会出现问题：

为解决该问题，引入关键字 volatile，编译器就发现 instance 是易失的，围绕这个变量的优化就会非常克制，不仅在读取变量的优化上克制，也会在修改变量的优化上克制，上述的 1 2 3 操作不会再成为 1 3 2 了

tip：Java 中的 voiatile 两个功能

1) 保证内存可见性

2) 禁止指令重排序（针对赋值操作）

1.2 阻塞队列

1.2.1 概念

阻塞队列是一种特殊的队列，也遵循 “先进先出” 原则，是一种线程安全的数据结构（标准库中原有的队列 Queue 和其子类默认都是线程不安全的），并且具有以下特性：

当队列满的时候，继续入队列就会阻塞，直到有其他线程从队列中取走元素
当队列空的时候，继续出队列就会阻塞，直到有其他线程往队列中插入元素

阻塞队列的一个典型应用场景就是 “生产者消费者模型”，是一种非常典型的开发模型

1.2.2 生产者消费者模型

生产者消费者模型就是通过一个容器来解决生产者和消费者的强耦合问题

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列中取

优点一：阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力（削峰填谷）

比如：在“双十一秒杀”的场景下，服务器同一时刻可能会收到大量的支付请求，如果直接处理这些支付请求，服务器可能扛不住（每个支付请求的处理都需要比较复杂的流程），这时就可以把这些请求放到一个阻塞队列中，然后再由消费者线程慢慢处理每个支付请求，有效进行“削峰”，防止服务器被突然到来的一波请求直接冲垮

如上图，A 中请求突然激增，若是没有阻塞队列，B 很可能就挂了；若有阻塞队列在，A 往队列中写入数据变快了，但是 B 仍然可以按照原有速度来消费数据

优点二：阻塞队列能使生产者和消费者之间解耦

比如：过年一家人一起包饺子，一般都是有明确分工，一人负责擀皮，其他人负责包，擀饺子皮的人就是“生产者”，包饺子的人就是“消费者”，擀饺子皮的人不关心包饺子的是谁（能包就行，无论是手工包，还是机器包），包饺子的人也不关心擀饺子皮的人是谁（有饺子皮就行，无论是用擀面杖擀的，还是从超市买的）

如上图，“直接调用”的关系，编写 A 代码中会出现很多 B 服务器相关的代码；编写 B 代码中也会出现很多 A 服务器相关的代码，若 B 服务器挂了，可能 A 服务器也会直接受到影响，若后续想继续增加一个 C 服务器，对 A 的改动就很大

如上图，A 只和队列通信，B 也只和队列通信，A 不知道 B 的存在，代码中更没有 B 的影子，B 同理

这样看起来 AB 之间是解耦合了，但 A 和队列，B 和队列之间是否引入了新的耦合呢？

我们之所以害怕耦合，是因为耦合的代码，在后续变更的过程中，比较复杂，容易出现 bug

消息队列，是成熟稳定的产品，代码不会频繁修改，A 和队列，B 和队列之间的交互逻辑基本写一次就固定下来了

tip：

1. 消息队列

通常谈到的 “阻塞队列” 是代码中的一个数据结构，但是由于其实用性很强，就把这个数据结构单独封装成一个服务器程序，并且在单独的服务器上进行部署，称其为 “消息队列”（Message Queue，MQ）

2. 为什么一个服务器收到请求变多可能会挂（崩溃）？

一台服务器就是一台“电脑”，上面提供了一些硬件资源（包括不限于 CPU、内存、硬盘、网络带宽...），就算机器配置再好，硬件资源也是有限的，而服务器每次收到一个请求，处理这个请求的过程需要执行一系列的代码，在执行这些代码的过程中，需要消耗一定的硬件资源（CPU、内存、硬盘、网络带宽...），这些请求消耗的总的硬件资源量超出了机器能提供的上限，机器就会出现问题（卡死、程序崩溃等...）

3. 在请求激增的时候 A 为什么不会挂？队列为什么不会挂？

A 的角色是一个“网关服务器”，收到客户端请求，再把请求转发给其他的服务器，这样的服务器中的代码做的工作比较简单（单纯的数据转发），消耗的硬件资源通常更少，处理一个请求消耗的资源更少，同配置下，就能支持更多的请求处理；同理，队列也是如此。

像 MySQL 这样的数据库，处理每个请求的时候，做的工作就比较多，消耗的硬件资源也是比较多的，因此 MySQL 也是后端系统中容易挂的部分

像 Redis 这样的内存数据库，处理请求做的工作远远少于 MySQL 做的工作，消耗的资源更少，Redis 就不容易挂

生产者消费者模型的缺点：

1) 需要更多的机器来部署这样的消息队列

2) A 和 B 之间通信的延时会边长（如果对于 A 和 B 之间的调用要求响应时间比较短，就不适合了；若是“转账”这样的场景，宁愿慢点，也要稳）

1.2.3 Java 标准库中的阻塞队列

BlockingQueue 是一个接口，真正实现的类是 LinkedBlockingQueue
put 方法用于阻塞式的入队列，take 方法用于阻塞式的出队列，其可以被 Interrupt 方法所唤醒（是线程安全的）
因为 BlockingQueue 继承于 Queue ，所以其也有 offer、poll、peek 等方法，但是这些方法不具有阻塞特性

示例代码：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;public class Demo27 {public static void main(String[] args) {BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(1000);//生产者线程Thread t1 = new Thread(() -> {int i = 1;while (true) {try {queue.put(i);System.out.println("生产元素 " + i);i++;//给生产操作加上 sleep，生产慢点，消费快点//Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});//消费者线程Thread t2 = new Thread(() -> {while (true) {try {Integer i = queue.take();System.out.println("消费元素 " + i);//给消费操作加上 sleep，生产快点，消费慢点Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}
}

当给消费者操作加上 sleep 时运行结果：

当给生产者操作加上 sleep 时运行结果：

模拟实现基于数组的阻塞队列

//此处不考虑泛型参数，只是基于 String 进行存储
class MyBlockingQueue {private String[] data = null;private volatile int head = 0;//标记头节点private volatile int tail = 0;//标记尾节点private volatile int size = 0;//记录有效数据个数public MyBlockingQueue(int capacity) {data = new String[capacity];}public void put(String s) throws InterruptedException {//加锁的对象，可以单独定义一个 locker，也可以直接使用 thissynchronized (this) {if (size == data.length) { //此处使用 while 是最稳妥的//队列满了，此处的 wait 由 take 操作this.wait();}data[tail] = s;tail++;if (tail >= data.length) {tail = 0;}size++;this.notify();}}public String take() throws InterruptedException {String ret = "";synchronized (this) {if (size == 0) { //此处使用 while 是最稳妥的//队列为空，此处的 wait 由 put 操作this.wait();}ret = data[head];head++;if (head > data.length) {head = 0;}size--;this.notify();}return ret;}
}

tip1：上述两种写法优劣分析：（针对写法方面，功能上是一样的）

1) 方法一的可读性更高，更直观，更简单；方法二求余运算，需要程序员非常熟悉求余是什么样的效果（无形中提高了要求）

2) 方法一的效率也高于方法二，因为对于计算机来说，% 就是算除法，除非是针对 2 的 N 次方进行乘除运算（会被编译器优化成移位运算），速度会非常快，否则 CPU 计算乘除法是一个比较慢的操作（特别是除法）；而方法一的写法是 “判定”，往往是一个非常简单快速的 cmp 指令（比乘除法快很多）

tip2：

1.3 线程池

最初引入线程，就是因为进程太重了，频繁创建销毁进程开销比较大（大/小是相对的）

而随着业务上对于性能要求越来越高，线程创建/销毁的频次越来越多，这时线程创建销毁的开销变得比较明显，无法忽视了

线程池就是解决上述问题的常见方案，线程池就是把线程提前从系统中申请好，放到一个地方，后面需要使用线程的时候，直接从这个地方来取，而不是从系统重新申请，线程用完之后，也是回到刚才的地方（资源是进程申请好了的，创建线程本身不需要资源分配）

1.3.1 内核态 & 用户态

操作系统 = 操作系统内核（操作系统的核心功能部分，负责完成一个操作系统的核心工作 “管理”） + 操作系统配套的应用程序

执行很多代码逻辑是需要用户态和内核态的代码配合完成的，实际场景中，应用程序有很多，其都是由内核统一负责管理和服务的，内核中非常繁忙

假设场景：让同学帮忙带饭，和自己去买饭

从系统创建线程，就相当于让同学帮忙带饭，这样的逻辑就是调用系统 api，由系统内核执行一系列逻辑来完成这个过程（同学不光给你带饭，还要买自己的，可能他还想去买个饮料，整个过程不可控）

直接从线程池里取，就相当于自己去买饭，整个过程都是纯用户态代码，整个过程更可控，效率更高

因此通常认为，纯用户态操作就比经过内核的操作效率更高

1.3.2 Java 标准库中的线程池

标准库提供了类 ThreadPoolExecutor（构造方法）

经典面试题：上面红框中构造方法的参数含义

1) int corePoolSize（核心线程数）， int maximumPoolSize（最大线程数）

此线程池中支持 ”线程扩容“，某个线程池初始状态下可能有 M 个线程，实际使用中发现 M 不够用，就会自动增加 M 个数

tip：在 Java 标准库的线程中的线程分为两类

a) 核心线程（也可理解成最少有多少个线程，相当于正式员工，一旦录用，不会轻易辞掉）

b) 非核心线程（线程扩容过程中新增的，相当于临时工，一段时间不干活就被辞退）

出现上述线程分类的原因是：cpu 上的核心数目是有限的

最大线程数 = 核心线程数 + 非核心线程数

2) long keepAliveTime（数值），TimeUnit unit（单位秒、分钟、小时、天...）

非核心线程会在线程空闲的时候被销毁，该参数就是允许非核心线程摸鱼的最大时间

3) BlockingQueue<Runnable> workQueue（工作队列）

线程池的工作过程是典型的“生产者消费者模型”，程序员使用的时候，通过形如 “submit” 这样的方法把要执行的任务设定到线程池里，线程池内部的工作线程负责执行这些任务

Runnable 接口本身的含义就是一段可以执行的任务

此处的阻塞队列由我们自行指定：a) 队列的容量 capacity；b) 队列的类型

4) ThreadFactory threadFactory（线程工厂）

就是 Thread 类的工厂类，通过整个类完成 Thread 实例的创建和初始化操作，此处的 ThreadFactory 就可以针对线程池里的线程进行批量的设置数据（一般使用标准库提供的 ThreadFactory 的默认值）

tip：“工厂” 指的是 “工厂设计模式”，是一种常见的设计模式，是在创建类的实例时使用的设计模式，因为构造方法有 “坑”，所以通过工厂模式来 “填坑”

详细分析：

构造方法是一个特殊的方法，必须和类名一样，要想实现多个版本的构造方法，必须通过 “重载”（overload），如下在一个平面描述一个点，可以通过平面坐标 “x y”，也可以通过极坐标 “α r” 两种构造方法来实例化：

class Point {public Point (double x, double y) {...}public Point (double α, double r) {...}
}

发现这两个构造方法无法构成重载，这就是构造方法的 “坑”（局限性）

为解决以上问题，引入了 “工厂设计模式”。通过 “普通方法”（通常是静态方法）完成对象构造和初始化的操作，如下：

class Point {}class PointFactory {public static Point makePointByXY (double x, double y) {Point p;p.setX(x);p.setY(y);return p;}public static Point makePointByRA (double r, double α) {Point p;p.setR(r);p.setA(α);return p;} 
}

此处用来创建对象的 static 方法就称为 “工厂方法”，有时工厂方法也会放到单独的类里实现，该类成为 “工厂类”

5) RejectedExecutionHandler handler（拒绝策略）（最重要！）

如果线程池的任务队列满了，还要继续给这个队列添加任务的话，不会阻塞，而是直接拒绝

Java 标准库给出了四种不同的拒绝策略：

ThreadPoolExecutor 的功能很强大，使用也很麻烦，所以 Java 标准库对这个类进一步封装了一下，Executors 提供了一些工厂方法，可以更方便的构造出线程池，如下：

如下线程池的示例代码：

发生了变量捕获，只需要再创建一个变量来接收 i 即可，如下：

运行结果：

执行这个代码，虽然 100 个任务都执行完毕了，但是整个进程并没有结束，这是因为此处线程创建出来的线程，默认都是 “前台线程”，虽然 main 线程结束了，但是这些线程池里的前台线程仍然存在

此时就需要使用 shutdown() 方法手动结束线程池里的线程，如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class thread {public static void main(String[] args) throws InterruptedException {ExecutorService service = Executors.newFixedThreadPool(4);for (int i = 0; i < 100; i++) {int id = i;service.submit(() -> {Thread current = Thread.currentThread();System.out.println("hello thread" + id + ", " + current.getName());});}//最好不要立即就终止，可能使任务还没执行完呢，线程就被终止了Thread.sleep(2000);//把线程池里所有的线程都终止掉service.shutdown();System.out.println("程序退出！");}
}

运行结果：

指定线程个数

使用线程池的时候需要指定线程个数，因为

1) 一台主机上，并不只是运行这一个程序

2) 这个程序也不是 100% 每个线程都跑满 cpu，线程工作过程中可能会涉及到一些 IO 操作/阻塞操作，而主动放弃 cpu（如果线程代码里都是算数运算，确实能跑满 cpu，如果代码中包含了 sleep、wait、加锁、打印、网络通信、读写硬盘等等操作，就会使线程主动放弃 cpu 一会）

在实际开发中更建议通过 “实验” 的方式找到一个合适的线程个数值（给线程池设置不同的线程数，分别进行性能测试，关注响应时间/消耗资源，挑选一个比较合适的数值）

模拟实现线程池

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;class MyThreadPool {private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);//此处 n 表示创建几个线程public MyThreadPool(int n) {//先创建出 n 个线程for (int i = 0; i < n; i++) {Thread t = new Thread(() -> {//循环的从队列中取任务while (true) {Runnable runnable = null;try {runnable = queue.take();runnable.run();} catch (InterruptedException e) {throw new RuntimeException(e);}}});t.start();}}//添加任务public void submit(Runnable runnable) {try {queue.put(runnable);} catch (InterruptedException e) {throw new RuntimeException(e);}}
}public class Demo {public static void main(String[] args) {MyThreadPool pool = new MyThreadPool(4);for (int i = 0; i < 1000; i++) {int id = i;pool.submit(() -> {System.out.println("执行任务" + id + ", " + Thread.currentThread().getName());});}}
}

1.4 实例：定时器

定时器相当于一个 “闹钟”，像网络通信中就经常需要设定一个 “超时时间”，此时就需要用到这个 “闹钟”机制

1.4.1 Java标准库中也提供了定时器实现：

import java.util.Timer;
import java.util.TimerTask;public class Demo32 {public static void main(String[] args) {Timer timer = new Timer();timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello");}}, 3000);System.out.println("程序开始运行！");}
}

运行结果：

从运行结果可以看出，当我们传入第二个参数 3000ms 之后，它就会等待 3s，然后再执行打印操作

而且 Timer 也支持管理多个任务：

import java.util.Timer;
import java.util.TimerTask;public class Demo32 {public static void main(String[] args) {Timer timer = new Timer();timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello3");}}, 3000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello2");}}, 2000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello1");}}, 1000);System.out.println("程序开始运行！");}
}

运行结果：

tip：

1.4.2 模拟实现定时器

对于定时器来说，其核心主要有：

1) 创建类，描述一个要执行的任务（任务的内容，任务的时间）

//表示一个任务
class MyTimerTask {private Runnable runnable; //任务的具体实现//此处的 time 通过毫秒时间戳来表示这个任务具体什么时候执行//意思是：此时是 9：00，时间设为 一小时，就是 10：00 执行（绝对时间）//而不是像 Demo32 中一样，只是 3000ms 之后执行（相对时间），没有具体时间private long time;public MyTimerTask(Runnable runnable, long delay) {this.runnable = runnable;this.time = System.currentTimeMillis() + delay;}public void run() {runnable.run();}public long getTime() {return time;}
}

2) 管理多个任务，通过某个数据结构，把多个任务存起来

若使用 List：

因为这里的 MyTimerTask 是按照时间来执行任务的，只要能够确定所有任务中，时间最小的任务，判定它是否到时间该执行即可（时间最小的任务还没到时间的话，其他任务就不必考虑了）

所以使用堆来保存任务是更好的选择，能够很方便的找到“最小值/第二小/第三小”

采用这个优先级队列来实现，但是 MyTimerTask 不能直接传入，需要指定明确的比较规则（方式一：实现 Comparable 接口；方式二：指定 Comparator 传入），下面实现一个 Comparable：

//表示一个任务
class MyTimerTask implements Comparable<MyTimerTask> {private Runnable runnable; //任务的具体实现//此处的 time 通过毫秒时间戳来表示这个任务具体什么时候执行//意思是：此时是 9：00，时间设为 一小时，就是 10：00 执行（绝对时间）//而不是像 Demo32 中一样，只是 3000ms 之后执行（相对时间），没有具体时间private long time;public MyTimerTask(Runnable runnable, long delay) {this.runnable = runnable;this.time = System.currentTimeMillis() + delay;}public void run() {runnable.run();}public long getTime() {return time;}@Overridepublic int compareTo(MyTimerTask o) {//此处的 - 的顺序，决定了是大堆还是小堆//本代码实现中需要小堆//这里 - 的顺序，不要背，通过“实验”的方式来确认return (int) (this.time - o.time);}
}//表示自己实现定时器
class MyTimer {//private List<MyTimerTask> list = new ArrayList<>(); //不是最优选private PriorityQueue<MyTimerTask> queue = new PriorityQueue<>();
}

3) 有专门的线程，执行这里的任务

//表示自己实现定时器
class MyTimer {//private List<MyTimerTask> list = new ArrayList<>(); //不是最优选private PriorityQueue<MyTimerTask> queue = new PriorityQueue<>();public MyTimer() {//创建线程，负责执行上述队列中的内容Thread t = new Thread(() -> {while (true) {if (queue.isEmpty()) {continue;}MyTimerTask current = queue.peek();//查看栈顶元素//比较当前时间与需执行时间，若当前为12：00，需执行时间为11：55，则应该执行if (System.currentTimeMillis() >= current.getTime()) {//要执行任务current.run();//把执行过的任务从队列中删除queue.poll();} else {//时间未到，不执行任务continue;}}});t.start();}//创建任务到队列中public void schedule(Runnable runnable, long delay) {MyTimerTask myTimerTask = new MyTimerTask(runnable, delay);queue.offer(myTimerTask);}
}