[Java EE] 多线程(五):单例模式与阻塞队列

1. 单例模式

单例模式是校招中最长考的设计模式之一,首先我们来谈一谈什么是设计模式:

设计模式就好像象棋中的棋谱一样,如果红方走了什么样的局势,黑方就有一定地固定地套路,来应对这样的局势,按照固定地套路来,可以保证在该局势下不会吃亏.

软件开发也是同样的道理,有很多的问题场景,针对这些问题场景,一些编程界的大佬就总结出了一套固定的解决模版,只要按着这个套路来,就会避免出现各种各样的bug.

软件设计中的单例模式就是保证每个程序运行的过程中只存在一份实例,而不会创建多个实例.这种设计模式在我们的日常开发中比比皆是,使用单例模式也可以大大提高程序的效率,下面我们来举个例子:
假如服务器中药从硬盘上读取100G的固定数据到内存中,这样的操作肯定是通过一个类来封装,这样的类最好是单例的,因为数据一样,在创建多个实例的时候,就会对系统资源开销很大,而且多次读取同样数据这样的操作也没必要,让他们共享同一份数据就可以了.

但是如何保证每个类只创建一个实例出来呢?这可不是程序员口头保证就行,只要是人去做,都不靠谱,我们应该把这样的操作交给计算机来处理,需要让编译器来帮我们做一个强制的检查.

通常的单例模式有以下两种,懒汉模式和饿汉模式:

1.1 饿汉模式

所谓饿汉模式,就突出一个字:急,就好像一个饿汉被饿了好几天想要得到食物一样.我们需要在JVM启动的时候,即类刚刚加载的时候,就创建这个类的实例.其次,我们为了该类只创建一个实例,防止通过构造方法来创建多余的实例,所以我们把构造方法设置为私有.之后要想获取这个类的实例,我们可以通过一个普通方法来获取类中创建的这个实例.代码如下:

/**
 * 饿汉模式
 */
class Singleton{
    public static Singleton instance = new Singleton();//在加载类的时候就创建实例
    private Singleton(){}//通过把构造方法设计为private修饰的来避免调用构造方法
    public static Singleton getInstance(){
        return instance;
    }
}

在多线程情况下,饿汉模式仍然是线程安全的:因为在创建这个类的时候,这个实例就已经有了,它的创建比主线程的启动都要早,所以在主线程中启动其他线程的时候,只需要读取已经创建好的线程就可以,我们前面说到,针对同一个变量进行修改的时候容易产生线程安全问题,但是现在只有读取操作,所以不存在线程安全问题.

1.2 懒汉模式

所谓懒汉模式,就突出一个字:懒,在第一次调用这个类的实例的时候才创建实例.如果没有被创建,引用一直为null.
单线程情况:

/**
 * 懒汉模式
 */
class SingletonLazy{
    Object object = new Object();
    public static SingletonLazy instance = null;
    private SingletonLazy(){}
    public static SingletonLazy getInstance(){
         if (instance == null){//判断实例是否被创建过,如果被创建过,就不创建了,直接返回
               instance = new SingletonLazy();
             }
      return instance;
   }
}

多线程情况:
上述代码虽然说在单线程情况下不会出现什么bug,但是在多线程的情况下,就会难免出现bug,下面我们就通过画图的方式解释以下为什么会出现bug.

在线程1和线程2创建实例的时候,线程2的条件判断有可能在线程1创建实例执之前,此时线程1和线程2都会创建实例,从而违背了单例模式的初衷.这里产生线程安全问题的原因是操作不符合原子性.
那么为了避免在多线程中的安全问题,我们可以通过加锁的方式,把创建对象这个操作打包成一个原子的操作.

/**
 * 懒汉模式
 */
class SingletonLazy{
    Object object = new Object();
    public static SingletonLazy instance = null;
    private SingletonLazy(){}
    public static SingletonLazy getInstance(){
        synchronized (instance.object){
            if (instance == null){//防止多个线程同时创建多个对象
                instance = new SingletonLazy();
            }
        }
        return instance;
    }
}

但是上述代码在一些方面还是显得有些赘余:

1. 如果线程比较多,会加重锁竞争,在最初调用getInstance的时候会存在线程安全问题,但是后续调用,就只是读取操作,并不存在线程安全问题,所以此时上锁就是赘余,上锁本身也是一个开销比较大的操作.
2. 存在内存可见性问题和指令重排序问题导致读取instance的偏差.
   为了解决上述问题,我们对上述代码进行优化:

/**
 * 懒汉模式
 */
class SingletonLazy{
    Object object = new Object();
    public static volatile SingletonLazy instance = null;//加上volatile方式指令重排序和内存可见性引起的bug
    private SingletonLazy(){}
    public static SingletonLazy getInstance(){
        if (instance == null){//节省上锁的开支
            synchronized (instance.object){
                if (instance == null){//防止多个线程同时创建多个对象
                    instance = new SingletonLazy();
                }
            }
        }
        return instance;
    }
}

这里两层判断的条件虽然是一样的,但是它们所起到的作用,却是大相径庭:
第一层判断是为了创建好实例之后,在后续获取实例的时候重复加锁,已节省系统开销.
第二层判断是为了多个线程如果都在创建好实例之前运行,防止这几个线程同时创建多个实例.

举例说明:
有请助教:荧,温迪,莱欧斯利,原神其他男性角色
现在温迪和莱欧斯利(两个线程)都知道了荧没有男朋友(没有创建实例),他们都突破了第一层if,于是开始竞争统一把锁.

假如现在风神巴巴托斯(线程1)更胜一筹:温迪就通过了里层的if,成为了荧的男朋友(创建了实例):

现在莱欧斯利(线程2)拿到了锁,但是被里层if排除在了外面,因为荧已经有了男朋友(实例已经创建了).

其他人听说了荧已经有了男朋友(已经创建了实例),纷纷散去了(被外层if排除在了外面).

2. 阻塞队列

2.1 什么是阻塞队列

阻塞队列是⼀种特殊的队列.也遵守"先进先出"的原则.
阻塞队列能是⼀种线程安全的数据结构(但是像我们前面学习的普通队列和优先级队列都是线程不安全的),并且具有以下特性:
• 当队列满的时候,继续⼊队列就会阻塞,直到有其他线程从队列中取⾛元素.
• 当队列空的时候,继续出队列也会阻塞,直到有其他线程往队列中插⼊元素.
阻塞队列的⼀个典型应⽤场景就是"⽣产者消费者模型".这是⼀种⾮常典型的开发模型.

拓展:消息队列
还有一种队列,叫做消息队列,首先通过topic这样的参数,对数据进行归类,出队列的时候,指定topic,每个topic下的所有数据先进先出,消息队列往往也带有阻塞特性.由于这种队列实在是太好用了,在部署服务器的时候,一般对这种队列会进行单独部署.

2.2 标准库中的阻塞队列

在Java标准库中内置了阻塞队列.如果我们需要在⼀些程序中使⽤阻塞队列.直接使⽤标准库中的即
可.
• BlockingQueue是⼀个接⼝.真正实现的类是LinkedBlockingQueue / ArrayBlockingQueue / PriorityBlockingQueue.
• put方法用于阻塞式的入队列,take用于阻塞式的出队列.
• BlockingQueue也有offer,poll,peek等⽅法,但是这些⽅法不带有阻塞特性.

2.3 生产者消费者模型

⽣产者消费者模式就是通过⼀个容器来解决⽣产者和消费者的强耦合问题。
⽣产者和消费者彼此之间不直接通讯, 而通过阻塞队列来进行通讯，所以⽣产者⽣产完数据之后不⽤等待消费者处理，直接扔给阻塞队列，消费者不找⽣产者要数据，⽽是直接从阻塞队列⾥取.

生产者和消费者模型有以下好处:

1. 阻塞队列就相当于⼀个缓冲区，平衡了⽣产者和消费者的处理能⼒.(削峰填谷)
   场景: 天猫11·11秒杀商品,在双十一的零点,阿里巴巴的服务器会接收到大量的支付请求,但是这样请求又是复杂的,很有可能服务器就被这样的一波请求冲垮了,这时候便会用到阻塞队列,让服务器慢慢处理这些指令.
   这样做可以有效进⾏"削峰",防⽌服务器被突然到来的⼀波请求直接冲垮.
   

2. 阻塞队列也能使⽣产者和消费者之间解耦.
   如果直接连接,服务器A和服务器B之间的逻辑关联会非常强,如果A或者B出现了bug或者出现了修改,另一个服务器就会被牵连到.这样的耦合性就比较高.我们希望通过阻塞队列来降低耦合性.让两个服务器和阻塞队列关联.

举例:过年包饺子
有请助教:空,荧
包饺子需要两步,擀皮,包馅,假如荧擀皮(生产者),空包馅(消费者),荧擀完皮会放在他们中间的盖帘上(阻塞队列).

假如荧擀皮慢,空包馅快,空在盖帘空的时候,空就会阻塞等待.
假如空包馅慢,荧擀皮快,空在盖帘满的时候,荧就会阻塞等待.

下面我们来用代码实现一个生产者消费者模型:

import java.util.concurrent.LinkedBlockingQueue;

/**
 * 生产者与消费者模型
 */
public class Demo23 {
    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();
        Thread thread = new Thread(()->{
            int count = 1;
            while (true){
                try {
                    blockingQueue.put(count);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("put:" + count);
                count++;
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        Thread thread1 = new Thread(()->{
            int num = 0;
            while (true){
                try {
                    num = blockingQueue.take();//当队列为空的时候就阻塞等待
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("take:" + num);
            }
        });
        thread.start();
        thread1.start();
    }
}

2.4 阻塞队列的实现

• 通过"循环队列"的方式实现.
  循环队列描述
• 使用synchronized进行加锁控制.
• put插⼊元素的时候,判定如果队列满了,就进⾏wait.(注意,要在循环中进⾏wait.被唤醒时不⼀定
  队列就不满了,因为还有可能是其他的线程的interrupt唤醒了该线程).
• take取出元素的时候,判定如果队列为空,就进⾏wait.(也是循环wait)

/**
 * 实现阻塞队列
 */
public class Block {
    public int size = 0;
    public int tail = 0;
    public int head = 0;
    public int[] item = new int[10];//默认循环队列的容积是10
    public int take() throws InterruptedException {//取队列元素
        synchronized (this){
            int ret = 0;
            while (size == 0){
                this.wait();
            }
            ret = item[head];//处队头元素
            head = (head+1)%item.length;
            size--;
            this.notify();//当出了一个元素之后,唤醒put的wait()
            return ret;
        }
    }
    public void put(int num) throws InterruptedException {
        synchronized (this){
            while (size == item.length){//队列满,阻塞等待
                this.wait();
            }
            item[tail] = num;
            tail = (tail+1)%item.length;
            size++;
            this.notify();//添加一个元素之后,唤醒take的阻塞
        }
    }
/**
*开始测试
**/

    public static void main(String[] args) {
        Block block = new Block();
        Thread thread = new Thread(()->{
            int num = 1;
            while (true){
                try {
                    block.put(num);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("put:"+num);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                num++;
            }
        });
        Thread thread1 = new Thread(()->{
            int ret = 0;
            while (true){
                try {
                    ret = block.take();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("take:"+ret);
            }
        });
        thread.start();
        thread1.start();
    }
}

测试结果: