【多线程】阻塞队列、定时器、线程安全的单例模式的原理及实现

快来打我* 2024-04-08 11:04 5阅读 0赞

#### 文章目录 ####

*  1. 线程安全版本的单例模式
 *   *  1.1 单例模式介绍
     *  1.2 实现线程安全版本的懒汉模式
 *  2. 阻塞队列
 *   *  2.1 阻塞队列介绍
     *  2.2 标准库中的阻塞队列
     *  2.3 实现阻塞队列
     *  2.4 生产者消费者模型
 *  3. 定时器
 *   *  3.1 定时器介绍
     *  3.2 标准库中的定时器
     *  3.3 实现定时器

## 1. 线程安全版本的单例模式 ##

### 1.1 单例模式介绍 ###

单例模式（Singleton Pattern）是 Java 最简单的设计模式之一。这种类型的设计模式属于创建型模式，它提供了一种创建对象的最佳方式。

这种模式涉及到一个单一的类，该类负责创建自己的对象，同时确保只有单个对象被创建。这个类也只提供一种访问其唯一对象的方式，可以直接访问，不需要实例化该类的对象。

单例模式具体的实现有以下两种：

*  饿汉模式（线程安全）
    
        class Singleton {
              
            // 被 static 修饰，则该类的对象只有一份（并且完成了初始化）
            private static Singleton instance = new Singleton();
        
            // 构造方法设为私有，保证外部类无法调用进行构造
            private Singleton(){
              }
        
            // 外部类唯一得到该类对象的方法
            public static Singleton getInstance(){
              
                return instance;
            }
        }
 *  懒汉模式（线程不安全）
    
        class Singleton {
              
            // 被 static 修饰，则该类的对象只有一份（未完成初始化）
            private static Singleton instance = null;
        
            // 构造方法设为私有，保证外部类无法调用进行构造
            private Singleton(){
              }
        
            // 外部类唯一得到该类对象的方法
            public static Singleton getInstance(){
              
                // 如果该类未有对象则进行创建
                if (instance == null) {
              
                    instance = new Singleton();
                }
                return instance;
            }
        }

饿汉模式和懒汉模式的区别：

*  饿汉模式的实例的创建在类加载的时候，而懒汉模式的实例创建是在第一次调用 getInstance 方法的时候。
 *  饿汉模式是线程安全地，因为 getInstance 方法中只存在读取变量。而懒汉模式是线程不安全的，因为在第一次调用 getInstance 的时候会进行读和写两个操作，如果多个线程同时调用 getInstance 则会导致同时读取和修改变量，可能会产生 bug。

### 1.2 实现线程安全版本的懒汉模式 ###

针对上述线程不安全的懒汉模式的代码，可以通过下面几个操作来解决：

1.  给读写的操作包起来，通过 synchronized 加锁，保证读写的原子性。
2.  给类的静态对象加上 volatile 关键字，保证内存的可见性。
3.  给原有加了 synchronized 的代码块再加上一个判断，保证只有第一次调用 getInstance 方法的时候会进行读取操作而加锁，而之后由于对象不会 null，则不会进行多余的获取锁的操作而降低开销。

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

## 2. 阻塞队列 ##

### 2.1 阻塞队列介绍 ###

阻塞队列（BlockingQueue）是一种特殊的队列，它遵循先进先出的原则。

阻塞队列是一种线程安全的数据结构，并且具有以下特性：

*  当队列满的时候，入队阻塞，直到有其它线程从队列中取走元素。
 *  当队列空的时候，出队阻塞，直到有其它线程从队列中插入元素。

**阻塞队列的作用：**

使用阻塞队列使我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，BlockingQueue 会帮我们负责。

### 2.2 标准库中的阻塞队列 ###

在 Java 标准库中内置了阻塞队列 BlockingQueue，它是一个接口，底下有七个实现类：

<table> 
 <thead> 
  <tr> 
   <th>实现类</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>ArrayBlockQueue</code></td> 
   <td>由数组结构组成的有界阻塞队列</td> 
  </tr> 
  <tr> 
   <td><code>LinkedBlockingQueue</code></td> 
   <td>由链表结构组成的有界的阻塞队列（有界，默认大小 Integer.MAX_VALUE，相当于无界）</td> 
  </tr> 
  <tr> 
   <td><code>PriorityBlockQueue</code></td> 
   <td>支持优先级排序的无界阻塞队列</td> 
  </tr> 
  <tr> 
   <td><code>DelayQueue</code></td> 
   <td>使用优先级队列实现的延迟无界阻塞队列</td> 
  </tr> 
  <tr> 
   <td><code>SynchronousQueue</code></td> 
   <td>不存储元素的阻塞队列，即单个元素的队列，生产一个，消费一个，不存储元素，不消费不生产</td> 
  </tr> 
  <tr> 
   <td><code>LinkedTransferQueue</code></td> 
   <td>由链表结构组成的无界阻塞队列</td> 
  </tr> 
  <tr> 
   <td><code>LinkedBlockingDeque</code></td> 
   <td>由链表结构组成的双向阻塞队列</td> 
  </tr> 
 </tbody> 
</table>

BlockingQueue 核心方法：

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>put(e)</code></td> 
   <td>将元素 e 阻塞式的入队列</td> 
  </tr> 
  <tr> 
   <td><code>take()</code></td> 
   <td>用于阻塞式的出队列</td> 
  </tr> 
 </tbody> 
</table>

BlockingQueue 也有其它方法，但是都不带有阻塞特性。

### 2.3 实现阻塞队列 ###

接下来将基于数组实现一个阻塞队列。

// 通过数组实现阻塞队列
    class MyBlockingQueue<T>{
        
        // 初始化队列大小
        private T[] items = (T[]) new Object[1000];
        // 队列中有效元素的个数
        private int size = 0;
        // 记录队首的位置
        private int head = 0;
        // 记录队尾的位置
        private int tail = 0;
    
        // 入队列
        public void put(T e) throws InterruptedException {
        
            synchronized (this) {
        
                if(size == items.length){
        
                    // 阻塞等待
                    this.wait();
                }
                if (tail == items.length) {
        
                    tail = 0;
                }
                items[tail++] = e;
                size++;
                // 唤醒 take 的阻塞等待
                this.notify();
            }
        }
    
        // 出队列
        public T take() throws InterruptedException {
        
            synchronized (this) {
        
                if(size == 0) {
        
                    // 阻塞等待
                    this.wait();
                }
                if (head == items.length) {
        
                    head = 0;
                }
                size--;
                // 唤醒 put 的阻塞等待
                this.notify();
                return items[head++];
            }
        }
    }

### 2.4 生产者消费者模型 ###

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

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

**生产者消费者模型的作用：**

*  在开发中起到服务器之间的解耦合效果
 *  在请求突然暴增的峰值中，起到削峰填谷的效果

**代码展示生产者消费者模型：**

// 使用阻塞队列作为交易场所
    private static MyBlockingQueue queue = new MyBlockingQueue();
    public static void main(String[] args) throws InterruptedException {
        
        // 生产者
        Thread producer = new Thread(() -> {
        
            int num = 1;
            while (true){
        
                try {
        
                    queue.put(num);
                    System.out.println("生产者生产了：" + num);
                    num++;
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        });
        producer.start();
        // 消费者
        Thread customer = new Thread(() -> {
        
            while (true){
        
                try {
        
                    System.out.println("消费者消费了：" + queue.take());
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        });
        customer.start();
    }

## 3. 定时器 ##

### 3.1 定时器介绍 ###

定时器是软件开发中的一个重要组件，类似于一个闹钟，当达到设定的时候后，就执行某个具体的任务。

### 3.2 标准库中的定时器 ###

Java 标准库中提供了一个 Time 类作为定时器，它有一个核心方法 schedule，能够在指定多长时间后执行某个任务。`schedule(TimerTask taks, long delay)` 包含两个参数，第一个参数是即将要执行的代码（通过 TimerTask 重写 run 方法来完成），第二个参数是指定多长事件之后执行（单位是毫秒）。

Timer timer = new Timer();
    timer.schedule(new TimerTask() {
        
        @Override
        public void run() {
        
            System.out.println("3s 后执行指定任务");
        }
    }, 3000);

### 3.3 实现定时器 ###

实现定时器，需要完成以下几个工作：

1.  创建一个描述任务的类（包括执行的具体任务，和多久后执行的时间，注意该类要实现 Comparable 接口，成为可比较的类才能保存到优先级队列中）
2.  使用优先级队列保存任务（可以使用 PriorityBlockingQueue，它是具有优先级的线程安全的阻塞队列）
3.  实现将任务注册到队列中的方法
4.  创建一个新的线程，执行扫描队列任务，对于时间到达的任务则进行执行（为了减少资源消耗，在为到达下一个最先执行的任务之前，可以通过 wait 方法来阻塞该线程）

class MyTimer {
        
        // 使用这个类表示当前的一个任务
        static class Task implements Comparable<Task>{
        
            // 具体要执行的任务
            private Runnable runnable;
            // 多久后执行该任务
            private long delay;
    
            public Task(Runnable runnable, long delay){
        
                this.runnable = runnable;
                this.delay = System.currentTimeMillis() + delay;
            }
    
            public void run(){
        
                runnable.run();
            }
    
            @Override
            public int compareTo(Task o) {
        
                return (int)(this.getDelay() - o.getDelay());
            }
    
            public Runnable getRunnable() {
        
                return runnable;
            }
    
            public void setRunnable(Runnable runnable) {
        
                this.runnable = runnable;
            }
    
            public long getDelay() {
        
                return delay;
            }
    
            public void setDelay(long delay) {
        
                this.delay = delay;
            }
        }
    
        // 通过带有优先级的阻塞队列存储任务，并且通过小根堆能够快速找到时间优先的任务
        private PriorityBlockingQueue<Task> tasks = new PriorityBlockingQueue<Task>();
    
        // 通过该方法往定时器中注册一个任务
        public void schedule(Runnable runnable, long delay){
        
            tasks.put(new Task(runnable, delay));
        }
    
        // 创建一个扫描线程，不停的扫描队首元素，判定任务是否可以执行
        // 在初始化 MyTimer 时，将扫描线程创建出来
        public MyTimer(){
        
            Thread scanner = new Thread(() -> {
        
                while(true) {
        
                    if (tasks.size() != 0) {
        
                        try {
        
                            Task task = tasks.take();
                            if (task.getDelay() > System.currentTimeMillis()) {
        
                                // 时间还没到
                                tasks.put(task);
                                synchronized (this){
        
                                    this.wait(task.getDelay() - System.currentTimeMillis());
                                }
                            }else {
        
                                task.run();
                            }
                        } catch (InterruptedException e) {
        
                            e.printStackTrace();
                        }
                    }
                }
            });
            scanner.start();
        }
    }