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

单例模式与消费者生产者模型，以及线程池的基本认识与模拟实现

news 来源：原创 2025/8/6 12:48:32

前言

今天我们就来讲讲什么是单例模式与线程池的相关知识，这两个内容也是我们多线程中比较重要的内容。其次单例模式也是我们常见设计模式。

单例模式

那么什么是单例模式呢？上面说到的设计模式又是什么？

其实单例模式就是设计模式的一种。我们在学习过程中会不断编程，设计合理的代码结构和逻辑。后来就有许多种比较常用的结构，这时就有一些大佬总结这些常用的代码结构，逻辑，并给这些结构命名成不同的模式，而这些模式就是我们所说的单例模式。那么接下来我们就来学习一下，其中之一的单例模式。

单例模式就是在某个类程序中保证某个实类对象在程序中只有一个实例，而且他的实例是不能new出来的，你只能通过其提供的getInstance方法来获取他的实例化对象。单例模式也分为两个经典的代码编写方式。“懒汉模式”和“饿汉模式”。

饿汉模式

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

如上代码：我们在要实现单例模式的类中，直接创建一个本类的成员变量，然后就是最牛的“点睛之笔”了。

我们直接将构造方法私有化，那么这时其它类就无法通过构造方法new出这个类的实例化对象了。

什么是饿汉模式呢？这里重点注意“饿”这个字，因为“饿”所以非常急，在类中直接将成员变量在定义时就直接实例化。

懒汉模式

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;}
}

我们可以很清楚的看到，懒汉模式相较于饿汉模式代码多了些，因为懒汉模式在多线程下需要加锁，如果不加锁可能有多个线程同时调用getInstance（）方法就有可能创建出多个实例，这就不符合单例模式的设定了。而且还加上了volatile 保证了内存的可见性也是防止线程安全的发生。

如何理解两层 if判断？

最里层的 if ，判断该类是否被实例化，如果没有实例化即 ==null 我们就new一个对象并返回，如果已经实例化过就直接返回。

最外层的 if 我们知道加锁和开锁也是一个开销比较高的事情，我们就要经尽量减少加锁开锁的次数，当我们的实例已经创建了，我们在就可以直接返回了，但是如果不加这一层 if，程序就会加锁判断部分，这就导致无用开销，所以我们就可以再加一层判断。这一层就是为了防止无用的加锁。

阻塞队列

什么是阻塞队列呢？

正如字面他是一个可以阻塞的队列，他跟普通队列一样有着FIFO（“先进先出”）出的性质。但是他会在两种情况下发生阻塞。

当队列满时，如果还有元素要入队列那么就会发生阻塞。
当队列为空时，如果有出队列的请求那么也会发生阻塞。

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

生产者消费者模型

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

生产者和消费者之间不直接通讯，通过阻塞队列来进行通讯，所以生产者生产完数据后不在需要等待消费者处理，而是直接将数据丢给阻塞队列，消费者也不找生产者要数据，而是直接从阻塞队列中取数据。这样就成功的对生产者和消费者进行解耦。

除了解耦，阻塞队列还起到了缓冲区的作用，阻塞队列就平衡了生产者和消费者的处理能力，起到了消峰填谷。比如在某些购物日，或秒杀抢购的情况下，如果没有阻塞队列，突然暴增的请求，如果让服务器直接去处理，我们的服务器有可能会处理不过来，而导致奔溃，但是如果有了阻塞队列我们就可以把请求放进阻塞队列中，再由消费者线程慢慢处理订单。

Java标准库中的阻塞队列

在Java标准库中我们有阻塞队列 BlockingQueue（这是一个接口），其中他的实现类是LinkedBlockingQueue

其中这个队列中我们有put（）方法表示入队列，还有take（）出队列，者两个方法是具有阻塞功能的.

当然这个队列也有offer，poll，peek，等方法，但这都是不具有阻塞功能的。

下面我们通过编写一段代码，象形的展示了阻塞队列的阻塞功能，非常直观。

我们先是创建第一个消费者线程，让其不断的从阻塞队列中去数据，但刚开始阻塞队列中没有数据，所以他就会进行阻塞，所以我们创建了第二个线程（生产者线程）我们每隔一段时间生成一个数据，并放进队列中，这时生产者线程就可会马上将队列里的数据给取出来了。

import java.util.Random;
import java.util.concurrent.BlockingQueue;import java.util.concurrent.LinkedBlockingQueue;public class Main {public static void main(String[] args) throws InterruptedException {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);Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}, "⽣产者");producer.start();customer.join();producer.join();}
}

阻塞队列的模拟实现

这里主要通过wait（）进行阻塞，当我们发现队列满时，或空时，我们的put方法和take（）方法就要进行阻塞，也就是调用wait（）方法以及我们的synchronize，但是需要注意的是我们这里的判断尽量还是用while循环来进行判断，因为在我们notifyAll 的时候, 该线程从 wait 中被唤醒,但是紧接着并未抢占到锁. 当锁被抢占的时候, 可能⼜已经队列满了

假设队列初始是满的，生产者 P1 调用 wait() 并释放锁。
消费者 C1 抢到锁，消费一个数据，调用 notifyAll()，唤醒 P1。
但此时可能有另一个生产者 P2 抢先抢到锁，并插入数据，导致队列又满了。
如果 P1 用 if，它会直接执行 items[tail] = value（因为之前检查 size == items.length 是 true，但被唤醒后未重新检查），导致队列溢出。
用 while 会重新检查条件，发现队列还是满的，继续 wait()。

如下代码：我们通过synchroniz关键字和wait（）方法，完成了阻塞队列。

class BlockingQueue1{private int[] items = new int[1000];private volatile int size = 0;private volatile int head = 0;private volatile 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++;notifyAll();}}public int take() throws InterruptedException {int ret = 0;synchronized (this) {while (size == 0) {wait();}ret = items[head];head = (head + 1) % items.length;size--;notifyAll();}return ret;}public synchronized int size() {return size;}// 测试代码public static void main(String[] args) throws InterruptedException {BlockingQueue1 blockingQueue = new BlockingQueue1();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();}
}

定时器

定时器是什么呢？就是一个可以根据时间定时执行任务的容器吧

定时器的主要构成：

一个带优先级的队列（ (不要使⽤ PriorityBlockingQueue, 容易死锁）
其中队列中的每一个元素是一个Task
Task中存在一个带有时间属性，其中队首元素是即将执行的元素。
还有存在一个工作线程worker不断扫描队首元素,,检查时间是否以及到了，是否开始执行

定时器的模拟实现：

首先我们先对MyTask重写我们的compareTo方法，如果不在这里重写，就要在创建队列的时候对其构造方法传入一个比较器的参数即（Comparable<MyTask>）。

且MyTask中是定义一个long类型的属性time，我们就可以利用时间戳来表示他要在何时执行任务。

然后定义构造方法，其中包含两个参数，第一个就是当前的时间戳，第二个是多少毫秒后执行当前任务。

然后我们在提交任务去定时器时，只需要传入这两个参数给schedule即可。在方法内部我们会先根据这两个参数

构造出一个Mytask,然后放进优先级队列中。

class MyTask implements Comparable<MyTask> {public Runnable runnable;// 为了⽅便后续判定, 使⽤绝对的时间戳.public long time;public MyTask(Runnable runnable, long delay) {this.runnable = runnable;// 取当前时刻的时间戳 + delay, 作为该任务实际执⾏的时间戳this.time = System.currentTimeMillis() + delay;}@Overridepublic int compareTo(MyTask o) {// 这样的写法意味着每次取出的是时间最⼩的元素.// 到底是谁减谁?? 俺也记不住!!! 随便写⼀个, 执⾏下, 看看效果~~return (int)(this.time - o.time);}
}class MyTimer {// 核⼼结构private PriorityQueue<MyTask> queue = new PriorityQueue<>();// 创建⼀个锁对象private Object locker = new Object();public void schedule(Runnable command, long after) {// 根据参数, 构造 MyTask, 插⼊队列即可.synchronized (locker) {MyTask myTask = new MyTask(command, after);queue.offer(myTask);locker.notify();}}// 在这⾥构造线程, 负责执⾏具体任务了.public MyTimer() {Thread t = new Thread(() -> {while (true) {try {synchronized (locker) {// 阻塞队列, 只有阻塞的⼊队列和阻塞的出队列, 没有阻塞的查看队⾸元素.while (queue.isEmpty()) {locker.wait();}MyTask myTask = queue.peek();long curTime = System.currentTimeMillis();if (curTime >= myTask.time) {// 时间到了, 可以执⾏任务了queue.poll();myTask.runnable.run();} else {// 时间还没到locker.wait(myTask.time - curTime);}}} catch (InterruptedException e) {e.printStackTrace();}}});t.start();}}

线程池

线程池是什么？通俗来讲就是先创建好一些线程，当我们需要创建线程时，直接从线程池里取即可，不需要在创建，而且用完后直接将线程返回给线程池，这样就减少了我们创建和销毁线程的开销。

Java标准库中的线程池

Executors是一个工厂类，提供了创建各种类型线程池的静态方法

固定大小线程池

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(int nThreads);
固定数量的线程
无界任务队列(LinkedBlockingQueue)
适用于负载较重的服务器

单线程线程池

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
只有一个工作线程
保证任务顺序执行
无界任务队列

可缓存线程池

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
线程数量根据需要自动调整
空闲线程60秒后回收
适用于执行大量短期异步任务

定时任务线程池

ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(int corePoolSize);
核心线程数固定，非核心线程数无限制
支持定时及周期性任务执行

如下代码就可以创建一个含有10个线程的线程池，其中

ExecutorService executorService = Executors.newFixedThreadPool(10);

从上上面代码可以很容易看出Executors.newFixedThreadPool(10)的返回值时一个ExecutorService，然后我们可以往线程池里提交任务执行了。

如下截图：我们调用submit方法时，只需要传入一个Runable的对象即可，跟创建线程的方法相似。

mport java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Main {public static void main(String[] args) {ExecutorService executorService = Executors.newFixedThreadPool(10);executorService.submit(()->System.out.println("一个人任务"));}
}

前言