前言:
大家好,我是良辰丫,今天我们来学习一下阻塞队列,这是一种基于线程安全的特殊队列,那么它到底有哪些应用呢?我们慢慢往下看.💞💞💞

🧑个人主页：良辰针不戳
📖所属专栏：javaEE初阶
🍎励志语句：生活也许会让我们遍体鳞伤，但最终这些伤口会成为我们一辈子的财富。
💦期待大家三连，关注，点赞，收藏。
💌作者能力有限，可能也会出错，欢迎大家指正。
💞愿与君为伴，共探Java汪洋大海。

阻塞队列

• 1. 简述阻塞式队列
• 2. 集合里的阻塞队列
• 3. 生产者消费者模型
• 4. 模拟实现阻塞队列

1. 简述阻塞式队列

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

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

2. 集合里的阻塞队列

首先我们先学习一下集合中的阻塞队列.

public static void main(String[] args) throws InterruptedException {
        BlockingDeque<Integer> blockingDeque = new LinkedBlockingDeque<>();
        blockingDeque.put(1);
        blockingDeque.put(2);
        blockingDeque.put(3);
        blockingDeque.put(4);
        System.out.println(blockingDeque.take());
        System.out.println(blockingDeque.take());
        System.out.println(blockingDeque.take());
        System.out.println(blockingDeque.take());
    }

阻塞队列一个重要的特性就是拥有阻塞状态,一般的队列,take次数大于put次数的时候就会抛出异常;然而我们的阻塞队列,take次数大于put次数的时候,不会报异常,会进入阻塞状态.如下图运行所示.

3. 生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。生产者消费者这样的模型是针对某个资源来说的.
生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取.

• 阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力.
• 阻塞队列也能使生产者和消费者之间 解耦.

下面代码我们来模拟一下生产者和消费者,让生产者, 每隔 1s 生产一个元素,有元素消费者进行消费,没有元素阻塞等待.

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class Test16 {
    public static void main(String[] args) {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>();

        // 消费者
        Thread t1 = new Thread(() -> {
            while (true) {
                try {
                    int value = blockingQueue.take();
                    System.out.println("消费元素: " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t1.start();

        // 生产者
        Thread t2 = new Thread(() -> {
            int value = 0;
            while (true) {
                try {
                    System.out.println("生产元素: " + value);
                    blockingQueue.put(value);
                    value++;
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t2.start();

      
    }
}

生产者消费者模型是一种非常重要的思想,我们写代码追求低耦合高内聚.低耦合表示各个代码文件有很少的关联,避免一个代码损坏,其余代码收到很大的牵连;高内聚表示一个相关联的代码分类管理,便于查找.

看了上面,大家或许还是不明白耦合性,那么我们再简单介绍一下耦合性.举个简单的例子,我们买一个二手房子,没有通过中介,直接和商家进行交易,结果呢,商家跑路了,这是一个不幸的消息;如果我们有中介公司做保证,商家无论出现怎样的情况,我们都有中介公司为我们承担.因此呢?有中介公司的时候是我们与商家耦合性比较低的情况.这就是解耦.

还有一种解决方案叫削峰填谷,什么意思呢?我们慢慢往下看,如下图,服务器接收客户端的请求,当然,客户端不止一个,如果直接进行通信,可能突然,客户端的请求大量增加,这时候服务器就反应不过来,此时服务器可能就会面临奔溃的的情况.因为服务器没有考虑峰值,那么在这种情况下就会挂了.因为服务器处理每个请求都需要消耗硬件资源(资源包括CPU,硬盘,内存,带宽等),如果资源达到瓶颈,就会出现故障,系统稳定性会带来很大的风险.

我们来看一下削峰填谷方案的本意.

那么我们就可以采用削峰填谷的方式,我们可以采用一个队列帮助服务器提取元素,可以叫做中间代理服务器.中间代理帮服务器承受了压力,这样就使得服务器不容易挂.如下图所示.

4. 模拟实现阻塞队列

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

class MyBlockingQueue2 {
    private int[] items = new int[1000];
    volatile private int head = 0;
    volatile private int tail = 0;
    volatile private int size = 0;

    // 入队列
    synchronized public void put(int elem) throws InterruptedException {
        while (size == items.length) {
            // 队列满了, 插入失败.
            // return;
            this.wait();
        }
        // 把新元素放到 tail 所在位置上
        items[tail] = elem;
        tail++;
        // 万一 tail 达到末尾, 就需要让 tail 从头再来.
        if (tail == items.length) {
            tail = 0;
        }
        // tail = tail % items.length;
        size++;
        this.notify();
    }

    // 出队列
    synchronized public Integer take() throws InterruptedException {
        while (size == 0) {
            // return null;
            this.wait();
        }
        int value = items[head];
        head++;
        if (head == items.length) {
            head = 0;
        }
        size--;
        this.notify();
        return value;
    }
}

public class Test {
    public static void main(String[] args) {
        MyBlockingQueue2 queue = new MyBlockingQueue2();
        // 消费者
        Thread t1 = new Thread(() -> {
            while (true) {
                try {
                    int value = queue.take();
                    System.out.println("消费: " + value);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        // 生产者
        Thread t2 = new Thread(() -> {
            int value = 0;
            while (true) {
                try {
                    System.out.println("生产: " + value);
                    queue.put(value);
                    value++;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        t1.start();
        t2.start();

    }
}

上述代码空队列阻塞和满队列阻塞不可能同时执行,因为一个队列不可能既是空又是满.