Java之阻塞队列和消息队列

刺骨的言语ヽ痛彻心扉 2024-03-22 15:07 71阅读 0赞

**目录**

一.上节复习

1.什么是单列模式

2.饿汉模式

3.懒汉模式

二.阻塞队列

1.什么是阻塞队列

三.消息队列

1.什么是消息队列

2.消息队列的作用

1.解耦

2.削峰填谷

3.异步

四.JDK中的阻塞队列

1.常见的阻塞队列

2.向阻塞队列中放入元素---put()

3.向阻塞队列中拿出元素---take()

五.手动实现阻塞队列

1.普通队列的实现

2.堵塞队列的实现

六.实现生产者和消费者模型

1.消费速度大于生产速度

2.生产速度大于消费速度

3.虚假唤醒

--------------------

## 一.上节复习 ##

上节内容指路:[Java之单例模式][Java]

### 1.什么是单列模式 ###

单例模式是一种设计模式(设计模式:就是在特定的场景下,解决问题最优的方式,类似于棋谱),单例:顾名思义,全局只有一个实例对象

### 2.饿汉模式 ###

**饿汉模式:类加载的时候就完成初始化,DCL双重检查锁**

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

优点:书写简单,不容易出错

### 3.懒汉模式 ###

**懒汉模式:程序使用对象的时候才进行初始化**

public class SingletonLazy {
        private static volatile SingletonLazy instance = null;
     
        private SingletonLazy() {
     
        }
     
        public static SingletonLazy getInstance() {
            if (instance == null) {
                //在获取单例对象的时候,判断是否已经被创建,没有创建则创建
                synchronized (SingletonLazy.class) {
                    if (instance == null) {
                        instance = new SingletonLazy();
                    }
                }
            }
     
            return instance;
        }
    }

优点:避免了资源的浪费

## 二.阻塞队列 ##

### 1.什么是阻塞队列 ###

和之前学习过的队列一样,也是FIFO(先进先出).

**入队**元素时,先判断队列是否**满了**,如何满了就**阻塞(等待)**,直到队列中**有空余空间再入队.**

**出队**元素时,先判断队列是否**为空**,如果空了就**阻塞(等待)**,直到队列中**有元素使再出队**.

实例:包饺子:分为擀饺子皮和包饺子两个操作

当放饺子皮的盘子满了,擀饺子皮的人停止擀皮(等待)--入队列操作,等待有空间了再工作

当放饺子皮的盘子空了,包饺子的人就停止包饺子(等待)---出队列操作,等待有饺子皮再工作

![f99fe12556c943328ba3da367c181c79.png][]

这种模式叫做生产者消费者模型

## 三.消息队列 ##

### 1.什么是消息队列 ###

消息队列本质上就是阻塞队列,在此基础上为放入阻塞队列的消息打上了标签

为不同的消息进行了分组的操作

在基础数据结构上，做了一些针对应用场景的优化和实现，那么我们把这样的框架和软件，称为“**中间件**”

消息队列就是中间件

### 2.消息队列的作用 ###

#### **1.解耦**  ####

**一个良好的程序应该是"高内聚,低耦合"的.**

**高内聚:将功能强相关的代码写在一块,方便维护**

**低耦合:两个相关的模块尽可以能把依赖的部分降低到最小,不要让两个系统产生强依赖**

实例:当我们订单支付,订单系统会对支付系统发起请求支付的指令,调用支付系统的相关命令,这样子就是一个高耦合的例子,如果支付系统崩溃,会直接对订单系统产生影响,支付系统修改相关的代码,订单系统也要发生修改

![121a8364a673426d97741ce13302a51d.png][]

这个时候我们维护一个消息队列,可以对两个系统进行解耦操作,即使支付系统崩溃了,订单系统也能进行正常的工作,代码修改也不会牵一发而动全身

![347f6e80f445456abf5c7befecdd6387.png][]

#### 2.削峰填谷 ####

峰和谷是指消息的密集程度

例如现实中的三峡大坝

汛期:蓄水,防止下游遭遇洪峰的冲击 削峰

旱期:把存的水进行灌溉使用 填谷

例如在双十一等流量很大的时间点,我们使用消息队列可以存储订单的信息,然后慢慢的处理订单的信息,过了时间点流量慢慢小的时候,我们依旧可以处理订单信息.没必要在大流量的时候一下子处理所有的订单,这样可能会造成服务器崩溃.

#### 3.异步 ####

同步:请求方必须死等对方的响应才能开始下一步操作.

异步:请求方发出请求之后,可以进行其他的操作,没必须等待对方的响应才开始操作.

比如订单系统对支付系统发出请求之后,没必须死等支付系统成功的响应就才开始其他订单的操作,而是进行其他订单的操作,同时等待这个订单的相应结果.

## 四.JDK中的阻塞队列 ##

### 1.常见的阻塞队列 ###

1.  LinkedBlockingQueue 基于链表
2.  ArrayBlockingQueue 基于数组
3.  PriorityBlockingQueue 基于优先级队列

在创建的时候可以指定队列的大小

![b522d0652576475d81c01c4d18b623fd.png][]

### 2.向阻塞队列中放入元素---put() ###

会抛出InterruptedException异常,阻塞队列中专用的入队的方法,不能使用offer()和add()

打印的结果:

![fdc28855d3d3456fbddb8d4f1db7855e.png][]

因为我们指定的阻塞队列的大小为3,当插入第四个元素的时候就会进入到阻塞等待的状态

### 3.向阻塞队列中拿出元素---take() ###

会抛出InterruptedException异常,阻塞队列中专用的出队的方法,不能使用poll()

public class Demo01_BlockingQueue {
        public static void main(String[] args) throws InterruptedException {
            //定义一个阻塞队列
            BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(3);
            //使用put()方法,不能使用add()和offer()
            queue.put(1);
            queue.put(2);
            queue.put(3);
            System.out.println("此时插入了三个元素");
            System.out.println(queue);
    //        queue.put(4);
    //        System.out.println("此时插入了四个元素");
            //一定要使用take()方法,不能使用poll()方法
            System.out.println(queue.take());
            System.out.println(queue.take());
            System.out.println(queue.take());
            //进入阻塞等待状态
            System.out.println(queue.take());
    
        }
    }

**打印结果:**

![ecb0c8974a8e4ba7984a3026a379400a.png][]

获取第四个元素的时候队列为空,因此进入阻塞等待的状态.

## 五.手动实现阻塞队列 ##

### 1.普通队列的实现 ###

在实现阻塞队列之前,我们现在实现普通队列

public class MyBlockingQueue {
        private int[] element;
        //队首下标
        private int head;
        //队尾下标
        private int tail;
        //元素个数
        private int size;
    
        public MyBlockingQueue() {
            this(3);
        }
    
        public MyBlockingQueue(int capacity) {
            element = new int[capacity];
        }
    
        /**
         * 入队一个元素
         *
         * @param val
         */
        public void put(int val) {
            if (size >= element.length) {
                return;
            }
            //向队尾插入元素
            element[tail] = val;
            //向后移动
            tail = (tail + 1) % element.length;
            size++;
    
        }
    
        /**
         * 出队一个元素
         *
         * @return
         */
        public int take() {
            if (size == 0) {
                return -1;
            }
            int val = element[head];
            head = (head + 1) % element.length;
            size--;
            return val;
        }
    
    }

### 2.堵塞队列的实现 ###

1.之前实现一个普通队列，底层用到了两种数据结构，一个是**链表**，一个是**循环数组**

2.阻塞队列就是在普通的队列上加入了**阻塞等待(wait())**和**唤醒操作(notify()),与synchronized相关**

**确定synchronized的范围,如果一个对象需要new出来使用,锁对象一般是this,其他情况具体分析**

**1.(线程1)当执行入队(put)操作时,判断阻塞队列满了,执行wait()操作,进入阻塞等待操作,当之后(别的线程)执行了出队列操作完成时,队列此时不满,这个时候唤醒队列.(线程1)继续完成put操作**

**2.(线程1)当执行出队(take)操作时,判断阻塞队列为空,执行wait()操作,进入阻塞等待操作,当之后(别的线程)执行了入队列操作完成时,队列此时不空,这个时候唤醒队列.(线程1)继续完成take操作**

public class MyBlockingQueue {
        private int[] element;
        //队首下标
        private int head;
        //队尾下标
        private int tail;
        //元素个数
        private int size;
    
        public MyBlockingQueue() {
            this(3);
        }
    
        public MyBlockingQueue(int capacity) {
            element = new int[capacity];
        }
    
        /**
         * 入队一个元素
         *
         * @param val
         */
        public void put(int val) throws InterruptedException {
            //加入修改的范围加锁
            synchronized (this) {
                if (size >= element.length) {
                    this.wait();
                }
    
    
                //向队尾插入元素
                element[tail] = val;
                //向后移动
                tail = (tail + 1) % element.length;
                size++;
                //做唤醒操作
                this.notifyAll();
            }
    
        }
    
        /**
         * 出队一个元素
         *
         * @return
         */
        public int take() throws InterruptedException {
            synchronized (this) {
                if (size == 0) {
                    this.wait();
                }
    
    
                int val = element[head];
                head = (head + 1) % element.length;
                size--;
                //唤醒操作
                this.notifyAll();
                return val;
            }
        }
    
    }

## 六.实现生产者和消费者模型 ##

我们使用两个线程,一个线程模仿生产者,向阻塞队列中放元素,一个线程模仿消费者,从阻塞队列中取元素.

![5dd5bd3a61ef45978b521d9b347722cf.png][]

### 1.消费速度大于生产速度 ###

public class Demo3_ProducerConsumer {
        static MyBlockingQueue queue = new MyBlockingQueue(3);
    
        public static void main(String[] args) {
            Thread producer = new Thread(() -> {
                int num = 1;
                while (true) {
                    //生产一条消息
                    try {
                        queue.put(num);
                        System.out.println("生产者生产了消息" + num);
                        num++;
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
    
                }
            });
            //启动生产者
            producer.start();
            Thread consumer = new Thread(() -> {
                while (true) {
                    //消费一条消息
                    try {
                        int num = queue.take();
                        System.out.println("消费者消费了消息" + num);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
    
                }
            });
            consumer.start();
    
        }
    }

**结果打印:**

![e735bc94d25249fc96c12594a26ebd0c.png][]

### 2.生产速度大于消费速度 ###

public class Demo3_ProducerConsumer {
        static MyBlockingQueue queue = new MyBlockingQueue(10);
    
        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) {
                        throw new RuntimeException(e);
                    }
    
                }
            });
            //启动生产者
            producer.start();
            Thread consumer = new Thread(() -> {
                while (true) {
                    //消费一条消息
                    try {
                        int num = queue.take();
                        System.out.println("消费者消费了消息" + num);
                        TimeUnit.SECONDS.sleep(1);
    
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
    
                }
            });
            //消费者后启动,让阻塞队列满
            TimeUnit.SECONDS.sleep(2);
            consumer.start();
    
        }
    }

**打印的结果:**

![1e3b52dc46cd40dcb73c83ebd98b0800.png][]

### 3.虚假唤醒 ###

这是Java官方文档中给出虚假唤醒的定义

![dbb77ede135846b297d1be2f2f929918.png][]

翻译成中文为: 线程也可以在没有被通知、中断或超时的情况下唤醒，这就是所谓的虚假唤醒。虽然这种情况在实践中很少发生，但应用程序必须通过测试本应导致线程被唤醒的条件，并在条件不满足时继续等待来防止这种情况的发生。换句话说，等待应该总是发生在循环中，就像下面这样：

所以在实践中wait条件的判断要加while循环.

阻塞队列中put和take方法的优化,并且多线程环境下共享变量要加voliatile,最终的阻塞队列的代码为:

public class MyBlockingQueue {
        private volatile int[] element;
        //队首下标
        private volatile int head;
        //队尾下标
        private volatile int tail;
        //元素个数
        private volatile int size=0;
    
        public MyBlockingQueue() {
            this(3);
        }
    
        public MyBlockingQueue(int capacity) {
            element = new int[capacity];
        }
    
        /**
         * 入队一个元素
         *
         * @param val
         */
        public void put(int val) throws InterruptedException {
            //加入修改的范围加锁
            synchronized (this) {
                while (size >= element.length) {
                    this.wait();
                }
    
    
                //向队尾插入元素
                element[tail] = val;
                //向后移动
                tail = (tail + 1) % element.length;
                size++;
                //做唤醒操作
                this.notifyAll();
            }
    
        }
    
        /**
         * 出队一个元素
         *
         * @return
         */
        public int take() throws InterruptedException {
            synchronized (this) {
                while (size == 0) {
                    this.wait();
                }
    
    
                int val = element[head];
                head = (head + 1) % element.length;
                size--;
                //唤醒操作
                this.notifyAll();
                return val;
            }
        }
    
    }

[Java]: https://blog.csdn.net/qq_64580912/article/details/130611342
[f99fe12556c943328ba3da367c181c79.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/968a8a58c8a24360832e110f66f1af37.png
[121a8364a673426d97741ce13302a51d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/46c56da4bbcd4b33876f8d687a5cb340.png
[347f6e80f445456abf5c7befecdd6387.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/f5ae164be59844f788034c0d53460280.png
[b522d0652576475d81c01c4d18b623fd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/a8a8b0bc625449f291eca1aae8add26e.png
[fdc28855d3d3456fbddb8d4f1db7855e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/7255b413a6de46ee84e020b1b6e3b0e9.png
[ecb0c8974a8e4ba7984a3026a379400a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/b65de5023845422da3148d00a9db41ec.png
[5dd5bd3a61ef45978b521d9b347722cf.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/1e4f19f188e045b0bbd906f6677adbf0.png
[e735bc94d25249fc96c12594a26ebd0c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/142ff37c638a4b8795421e283f972859.png
[1e3b52dc46cd40dcb73c83ebd98b0800.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/57e5e932c61e4563aa6d495f8011b89d.png
[dbb77ede135846b297d1be2f2f929918.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/18/d24f042ed03045c6aeaae2196cbd7679.png