javaEE -3（12000字详解多线程）

一：单例模式

首先啥是设计模式?

设计模式好比象棋中的 “棋谱”. 红方当头炮, 黑方马来跳. 针对红方的一些走法, 黑方应招的时候有一些固定的套路. 按照套路来走局势就不会吃亏.

软件开发中也有很多常见的 “问题场景”. 针对这些问题场景, 大佬们总结出了一些固定的套路. 按照这个套路来实现代码, 也不会吃亏.

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

而单例模式具体的实现方式, 分成 “饿汉” 和 “懒汉” 两种.

1.1 饿汉模式

类加载的同时, 创建实例.

class Singleton {private static Singleton instance = new Singleton();private Singleton() {}public static Singleton getInstance() {return instance;}
}

首先我们通过new创建了一个对象，但是这个对象是静态的，所以对于这整个类来说，这个对象只会存在一份，并且由于构造方法Singleton()被私有化，外部无法直接创建Singleton对象，从而保证了单例模式的实现

而getInstance()方法作为公共静态方法，返回了instance对象，以提供对单例对象的访问。

通过将自身的对象实例化并保存在静态成员变量中，通过公共的静态方法返回该实例，我们可以确保在整个应用程序中只有一个Singleton对象的存在。这就是单例模式的核心思想。

1.2懒汉模式

1.2.1懒汉模式-单线程版

类加载的时候不创建实例. 第一次使用的时候才创建实例.

class Singleton {private static Singleton instance = null;private Singleton() {}public static Singleton getInstance() {if (instance == null) {instance = new Singleton();}return instance;}
}

这段代码的关键在于，通过私有化构造函数和静态的 getInstance() 方法，保证了外部无法直接创建 Singleton 类的对象，并且每次调用 getInstance() 方法时都返回同一个实例。

因此，当第一次调用 getInstance() 方法时，instance 为 null，会创建一个新的 Singleton 对象并将其赋值给 instance。而当后续调用 getInstance() 方法时，由于 instance 已经被赋值为 Singleton 对象，不再为 null，直接返回 instance。这样就确保了只有一个 Singleton 实例被创建并返回。

请注意，这段代码是线程不安全的，如果多个线程同时调用 getInstance() 方法，有可能会创建多个 Singleton 实例。线程安全问题发生在首次创建实例时. 如果在多个线程中同时调用 getInstance 方法, 就可能导致创建出多个实例.一旦实例已经创建好了, 后面再多线程环境调用 getInstance 就不再有线程安全问题了(不再修改instance 了)

1.2.2 懒汉模式-多线程版

加上 synchronized 可以改善这里的线程安全问题：

class Singleton {private static Singleton instance = null;private Singleton() {}public synchronized static Singleton getInstance() {if (instance == null) {instance = new Singleton();}return instance;}
}

1.2.3 懒汉模式-多线程版(改进)

以下代码在加锁的基础上, 做出了进一步改动:

使用双重 if 判定, 降低锁竞争的频率.
给 instance 加上了 volatile.

class Singleton {private static volatile Singleton instance = null;private Singleton() {}public static Singleton getInstance() {if (instance == null) {synchronized (Singleton.class) {if (instance == null) {instance = new Singleton();}}}return instance;}
}

理解双重 if 判定 / volatile:
加锁 / 解锁是一件开销比较高的事情. 而懒汉模式的线程不安全只是发生在首次创建实例的时候.因此后续使用的时候, 不必再进行加锁了.

外层的 if 就是判定下看当前是否已经把 instance 实例创建出来了.同时为了避免 “内存可见性” 导致读取的 instance 出现偏差, 于是补充上 volatile .

当多线程首次调用 getInstance, 大家可能都发现 instance 为 null, 于是又继续往下执行来竞争锁,其中竞争成功的线程, 再完成创建实例的操作.

当这个实例创建完了之后, 其他竞争到锁的线程就被里层 if 挡住了. 也就不会继续创建其他实例.

二：阻塞式队列

阻塞队列是一种特殊的队列. 也遵守 “先进先出” 的原则.
阻塞队列能是一种线程安全的数据结构, 并且具有以下特性:

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

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

2.1 生产者和消费者模型

生产者和消费者模型是一种常用的多线程协作模型，用于解决生产者与消费者之间的同步和协作问题。在这个模型中，一个或多个生产者生成数据，并将其放入一个共享的缓冲区中，而一个或多个消费者从缓冲区中取出数据进行消费。这种模型可以在多线程环境下实现生产者和消费者之间的有效通信和数据交换。

为了更好地理解生产者和消费者模型，我们可以通过一个生活上的例子来说明。假设有一个餐厅，菜品的制作是由厨师（生产者）完成的，而食客（消费者）则将这些菜品消费掉。

在这个例子中，厨师是生产者，负责制作菜品，并将其放入菜品架（缓冲区）上。食客是消费者，从菜品架上取走菜品进行消费。

在这个模型中，生产者和消费者之间需要进行协调和同步。如果菜品架已满，厨师需要等待一段时间，直到有空的位置放置菜品。如果菜品架为空，食客需要等待一段时间，直到有菜品可供消费。

在这个例子中，使用一个容量有限的菜品架模拟了缓冲区的概念。生产者和消费者之间通过对菜品架的交互来实现数据的传递。

那么这个生产者和消费者模型有什么作用呢？

阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力

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

阻塞队列也能使生产者和消费者之间解耦.

比如过年一家人一起包饺子. 一般都是有明确分工, 比如一个人负责擀饺子皮, 其他人负责包. 擀饺子皮的人就是 “生产者”, 包饺子的人就是 “消费者”.

擀饺子皮的人不关心包饺子的人是谁(能包就行, 无论是手工包, 借助工具, 还是机器包), 包饺子的人也不关心擀饺子皮的人是谁(有饺子皮就行, 无论是用擀面杖擀的, 还是拿罐头瓶擀, 还是直接从超市买的).

2.2 标准库中的阻塞队列

在 Java 标准库中内置了阻塞队列. 如果我们需要在一些程序中使用阻塞队列, 直接使用标准库中的即可.

BlockingQueue 是一个接口. 真正实现的类是 LinkedBlockingQueue.
put 方法用于阻塞式的入队列, take 用于阻塞式的出队列.
BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性.

BlockingQueue<String> queue = new LinkedBlockingQueue<>();
// 入队列
queue.put("abc");
// 出队列. 如果没有 put 直接 take, 就会阻塞.
String elem = queue.take();

生产者消费者模型

public static void main(String[] args) throws InterruptedException {// 创建一个阻塞队列，使用 LinkedBlockingQueue 实现BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>();// 创建消费者线程Thread customer = new Thread(() -> {while (true) {try {// 从阻塞队列中获取元素并打印int value = blockingQueue.take();System.out.println("消费元素: " + value);} catch (InterruptedException e) {e.printStackTrace();}}}, "消费者");// 启动消费者线程customer.start();// 创建生产者线程Thread producer = new Thread(() -> {Random random = new Random();while (true) {try {// 生成一个随机数作为元素int num = random.nextInt(1000);System.out.println("生产元素: " + num);// 将元素放入阻塞队列中blockingQueue.put(num);// 生产者线程休眠 1 秒Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}, "生产者");// 启动生产者线程producer.start();// 等待消费者线程和生产者线程执行结束customer.join();producer.join();
}

2.3阻塞队列实现

通过 “循环队列” 的方式来实现.
使用 synchronized 进行加锁控制.
put 插入元素的时候, 判定如果队列满了, 就进行 wait. (注意, 要在循环中进行 wait. 被唤醒时不一定队列就不满了, 因为同时可能是唤醒了多个线程).
take 取出元素的时候, 判定如果队列为空, 就进行 wait. (也是循环 wait)

public class BlockingQueue {private int[] items = new int[1000];  // 存放元素的数组private volatile int size = 0;  // 队列中元素的数量private int head = 0;  // 队列头部索引private int tail = 0;  // 队列尾部索引// 向队列中放入元素public void put(int value) throws InterruptedException {synchronized (this) {// 当队列已满时，等待// 此处最好使用 while，否则 notifyAll 的时候, 该线程从 wait 中被唤醒,但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能又已经队列满了，就只能继续等待while (size == items.length) {wait();}items[tail] = value;  // 将元素放入队列尾部tail = (tail + 1) % items.length;  // 更新尾部索引size++;  // 元素数量加1notifyAll();  // 唤醒其他等待线程}}// 从队列中取出元素public int take() throws InterruptedException {int ret = 0;synchronized (this) {// 当队列为空时，等待while (size == 0) {wait();}ret = items[head];  // 取出队列头部元素head = (head + 1) % items.length;  // 更新头部索引size--;  // 元素数量减1notifyAll();  // 唤醒其他等待线程}return ret;}// 获取队列中元素的数量public synchronized int size() {return size;}// 测试代码public static void main(String[] args) throws InterruptedException {BlockingQueue blockingQueue = new BlockingQueue();// 定义消费者线程Thread customer = new Thread(() -> {while (true) {try {int value = blockingQueue.take();  // 从队列中取出元素System.out.println(value);  // 打印取出的元素值} catch (InterruptedException e) {e.printStackTrace();}}}, "消费者");customer.start();  // 启动消费者线程// 定义生产者线程Thread producer = new Thread(() -> {Random random = new Random();while (true) {try {blockingQueue.put(random.nextInt(10000));  // 向队列中放入随机生成的元素} catch (InterruptedException e) {e.printStackTrace();}}}, "生产者");producer.start();  // 启动生产者线程customer.join();  // 等待消费者线程结束producer.join();  // 等待生产者线程结束}
}

该代码实现了一个阻塞队列，其中包括以下方法和功能：

put(int value) 方法用于向队列中放入元素。
- 在队列已满时，使用 wait() 方法等待。
- 使用 notifyAll() 方法唤醒其他等待的线程。
take() 方法用于从队列中取出元素。
- 在队列为空时，使用 wait() 方法等待。
- 使用 notifyAll() 方法唤醒其他等待的线程。
size() 方法用于获取队列中元素的数量。
main 方法用于测试代码：
- 创建一个 BlockingQueue 实例。
- 定义一个消费者线程，不断从队列中取出元素并打印。
- 定义一个生产者线程，不断向队列中放入随机生成的元素。
- 启动消费者线程和生产者线程。
- 使用 join() 方法等待消费者线程和生产者线程结束。

三：定时器

定时器也是软件开发中的一个重要组件. 类似于一个 “闹钟”. 达到一个设定的时间之后, 就执行某个指定好的代码.

定时器是一种实际开发中非常常用的组件，比如网络通信中, 如果对方 500ms 内没有返回数据, 则断开连接尝试重连，比如一个 Map, 希望里面的某个 key 在 3s 之后过期(自动删除)，类似于这样的场景就需要用到定时器。

3.1标准库中的定时器

标准库中提供了一个 Timer 类. Timer 类的核心方法为 schedule
schedule 包含两个参数. 第一个参数指定即将要执行的任务代码, 第二个参数指定多长时间之后执行 (单位为毫秒).

Timer类的schedule方法用于安排指定的任务在指定的时间开始执行，或者在指定的延迟后开始执行。它有多种重载形式，这里将详细解释最常用的一个重载形式：

public void schedule(TimerTask task, Date time)

参数解释：

task：要被调度的任务，通常是继承自TimerTask类并实现其run方法的一个对象。
time：任务开始执行的时间，是一个java.util.Date对象。

schedule方法的执行流程：

在指定的时间点time之后，Timer会创建一个新线程，并在新线程中启动任务task的执行。
在任务task执行之前，Timer会使用一个内部定时器线程来等待，直到指定的时间点到达。
任务task在指定的时间点到达后会被执行，此时Timer会调用task对象的run方法。

示例代码如下：

import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;public class TimerExample {public static void main(String[] args) {Timer timer = new Timer();// 创建一个任务，继承自TimerTask类，并实现其run方法TimerTask task = new TimerTask() {@Overridepublic void run() {System.out.println("任务开始执行，当前时间：" + new Date());}};// 设定任务在5秒后开始执行Date startTime = new Date(System.currentTimeMillis() + 5000);// 安排任务在指定时间开始执行timer.schedule(task, startTime);}
}

以上代码创建了一个Timer对象，并定义了一个任务task，该任务在5秒后开始执行。通过schedule方法将任务安排在指定时间开始执行。在任务执行时，会打印出当前的时间。

3.2 实现定时器

public class Timer {// 内部类，表示要执行的任务static class Task implements Comparable<Task> {private Runnable command; // 要执行的任务private long time; // 任务执行的时间// 构造方法，初始化任务和执行时间public Task(Runnable command, long time) {this.command = command;// time 中存的是绝对时间, 超过这个时间的任务就应该被执行this.time = System.currentTimeMillis() + time;}// 执行任务的方法public void run() {command.run();}// 实现 Comparable 接口，根据任务执行时间进行排序@Overridepublic int compareTo(Task o) {// 谁的时间小谁排前面return (int)(time - o.time);}}// 核心结构private PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<Task>();// 存在的意义是避免 worker 线程出现忙等的情况private Object mailBox = new Object();// 内部类，表示工作线程class Worker extends Thread {@Overridepublic void run() {while (true) {try {Task task = queue.take(); // 从队列中取出一个任务long curTime = System.currentTimeMillis();if (task.time > curTime) {// 时间还没到, 就把任务再塞回去queue.put(task);synchronized (mailBox) {// 指定等待时间 waitmailBox.wait(task.time - curTime);}} else {// 时间到了, 可以执行任务task.run();}} catch (InterruptedException e) {e.printStackTrace();break;}}}}// 构造方法，创建并启动工作线程public Timer() {Worker worker = new Worker();worker.start();}// 安排要执行的任务public void schedule(Runnable command, long after) {Task task = new Task(command, after);queue.offer(task); // 将任务插入队列synchronized (mailBox) {mailBox.notify(); // 唤醒工作线程}}// 主函数，用于测试public static void main(String[] args) {Timer timer = new Timer();Runnable command = new Runnable() {@Overridepublic void run() {System.out.println("我来了");timer.schedule(this, 3000); // 3秒后再次调度任务}};timer.schedule(command, 3000); // 3秒后执行任务}
}

四：线程池

在Java中，线程池是一种用于管理和调度线程的机制。线程池由一组预先创建的线程组成，这些线程可以被重复利用，从而减少了线程的创建和销毁开销。

线程池的主要作用是提高程序的性能和资源利用率。它可以在系统初始化时创建一定数量的线程放入线程池中，当任务到达时，从线程池中取出一个线程来处理任务，当线程完成任务后，再将该线程放回线程池中，等待下一个任务的到来。通过有效地管理线程的生命周期，线程池可以控制并发线程的数量，避免创建过多的线程导致资源耗尽。

线程池的好处包括以下几个方面：

降低资源消耗：线程的创建和销毁是一项开销较大的操作，使用线程池可以避免频繁地创建和销毁线程，从而降低了系统的资源消耗。
提高响应速度：线程池中的线程是预先创建的，当有任务到达时，可以立即执行，提高了任务的响应速度。
提高线程的可管理性：线程池可以对线程进行统一的管理，如线程的创建、销毁、监控等，提供了更好的线程管理方式。
提供线程复用：线程池中的线程可以被重复利用，避免了频繁创建和销毁线程的开销，提高了系统的性能和资源利用率。

4.1 标准库中的线程池

ExecutorService 是 Java 提供的一个线程池框架，用于管理和执行多线程任务。它是 Executor 的子接口，提供了更丰富的功能和更灵活的线程管理。

首先，我们需要使用 Executors 工厂类提供的方法来创建 ExecutorService。以下是几种常用的创建线程池的方式：

newFixedThreadPool(int nThreads)：创建包含固定线程数的线程池。
代码示例：
```
ExecutorService executorService = Executors.newFixedThreadPool(10);
```
newCachedThreadPool()：创建线程数目动态增长的线程池，适用于执行短期异步任务的程序。
代码示例：
```
ExecutorService executorService = Executors.newCachedThreadPool();
```
newSingleThreadExecutor()：创建只包含单个线程的线程池，保证任务按顺序执行。
代码示例：
```
ExecutorService executorService = Executors.newSingleThreadExecutor();
```
newScheduledThreadPool(int corePoolSize)：创建可以定期执行任务的线程池，具有延迟或周期执行任务的能力。
代码示例：
```
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
```

创建了 ExecutorService 后，我们可以通过调用其 submit() 方法将任务提交给线程池执行。submit() 方法接受一个 Callable 或 Runnable 对象作为参数，并返回一个 Future 对象用于获取任务的执行结果。

以下是一个简单的示例代码，演示了如何创建一个固定线程数的线程池，并提交多个任务执行：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;public class ExecutorServiceExample {public static void main(String[] args) {// 创建固定线程数的线程池ExecutorService executorService = Executors.newFixedThreadPool(10);// 提交任务到线程池执行for (int i = 0; i < 10; i++) {final int taskId = i;executorService.submit(() -> {try {// 模拟任务执行耗时Thread.sleep(1000);System.out.println("Task " + taskId + " completed.");} catch (InterruptedException e) {e.printStackTrace();}});}// 关闭线程池executorService.shutdown();}
}

在这个示例中，我们创建了一个包含固定线程数为 10 的线程池，并通过循环提交了 10 个任务。每个任务执行的操作是打印任务编号和完成信息。最后，我们调用 executorService.shutdown() 来关闭线程池。

通过使用 ExecutorService，我们可以更方便地管理线程和任务，提高程序的效率和性能，Executors 本质上是 ThreadPoolExecutor 类的封装，ThreadPoolExecutor 提供了更多的可选参数, 可以进一步细化线程池行为的设定. (后面再介绍)

4.2实现线程池

class Worker extends Thread {// 创建一个阻塞队列来保存任务private LinkedBlockingQueue<Runnable> queue = null;// Worker 的构造函数，传入任务队列public Worker(LinkedBlockingQueue<Runnable> queue) {// 调用父类 Thread 的构造函数，给线程命名为 "worker"super("worker");this.queue = queue;}// 线程执行的方法@Overridepublic void run() {try {// 循环判断线程是否被中断while (!Thread.interrupted()) {// 从任务队列中取出一个任务Runnable runnable = queue.take();// 执行任务的 run 方法runnable.run();}} catch (InterruptedException e) {// 捕获 InterruptedException 异常}}
}public class MyThreadPool {// 设置线程池最大工作线程数为 10private int maxWorkerCount = 10;// 创建一个任务队列private LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue();// 提交任务的方法public void submit(Runnable command) {// 如果当前工作线程数小于最大工作线程数if (workerList.size() < maxWorkerCount) {// 继续创建新的 worker 线程Worker worker = new Worker(queue);worker.start();}// 将任务添加到任务队列中queue.put(command);}public static void main(String[] args) throws InterruptedException {// 创建一个 MyThreadPool 实例MyThreadPool myThreadPool = new MyThreadPool();// 执行一个任务，输出 "吃饭"myThreadPool.execute(new Runnable() {@Overridepublic void run() {System.out.println("吃饭");}});// 线程休眠 1000 毫秒Thread.sleep(1000);}
}