阻塞队列概念 

阻塞队列是一种特殊的队列，同样遵循“先进先出”的原则，支持入队操作和出队操作和一些基础方法。在此基础上，阻塞队列会在队列已满或队列为空时陷入阻塞，所以它具有如下特性：

当队列已满时，继续入队列就会阻塞，直到有其他线程从队列中取走元素。
当队列为空时，继续出队列也会阻塞，直到有其他线程向队列中插入元素。
并且它是线程安全的，就是多线程中使用它是不会引发线程安全bug的
那么我们用它是干嘛呢？一般用它实现生产者消费者模型，对于该概念我们下面详细说下：
 

生产者消费者模型

生产者消费者模型有两种角色，生产者和消费者，两者之间通过缓冲容器来达到解耦合和削峰填谷的效果。类似于厂商和客户与中转仓库之间的关系，如下图：

 

厂家生产的商品堆积在中转仓库，当中转仓库满时，入仓阻塞，当中转仓库为空时，出仓阻塞。通过上述结构，生产者和消费者摆脱了“产销一体”的运作模式，即解耦合。同时，无论是客户需求暴增，还是厂家产量飙升，都会被中央仓库协调，避免突发情况导致结构崩溃，达到削峰填谷的作用。

同理，根据生产者消费者模型，我们将线程带入到消费者和生产者的角色，阻塞队列带入到缓冲空间的角色，一个类似的模型很容易就搭建起来了。
 

阻塞队列的作用

①解耦合

作为生产者消费者模式的缓冲空间，将线程（其他）之间分隔，通过阻塞队列间接联系起来，起到降低耦合性的作用，这样即使其中一个挂掉，也不会使另一个也跟着挂掉。（就是降低它们之间的联系性）

 

②削峰填谷

因为阻塞队列本身的大小是有限的，所以能起到一个限制作用，即在消费者面对突发暴增的入队操作，依然不受影响。

如电商平台在每年双十一时都会出现请求峰值的情况，电商平台对请求的处理流程是这样的：

因为处理请求需要消耗硬件资源，如果没有消息队列，面对双十一这种请求暴增的情况，请求处理服务器很可能就直接挂掉了。
而有了消息队列之后，请求处理服务器不必直接面对大量请求的冲击，仍旧可以按原先的处理速度来处理请求，避免了被冲爆，这就是‘削峰’。
没有被处理的请求也不是不处理了，而是当消息队列有空闲时再继续流程，即高峰请求被填在低谷中，这就是‘填谷’
 

阻塞队列的使用

在 Java 标准库中就提供了现成阻塞队列这样的数据结构：BlockingQueue ，这里 BlockingQueue 是一个接口，实现这个接口的类也有很多：

ArrayBlockingQueue: 基于数组的阻塞队列。
LinkedBlockingQueue: 基于链表的阻塞队列。
PriorityBlockingQueue: 支持优先级的阻塞队列。
 

阻塞队列一般用put和take方法。
put 方法用于阻塞式的入队列, take 用于阻塞式的出队列. BlockingQueue 也有 offer, poll, peek 等方法, 但是这些方法不带有阻塞特性，所以不用

阻塞队列的实现

实现阻塞队列，我们可以从浅到深的来实现，先实现一个普通队列，再在普通队列的基础上，添加上线程安全，再增加阻塞功能，那么就来普通队列的实现吧。这里我们实现一个环形队列（之前讲过怎么实现，这里直接给代码）

class MyBlockingQueue {
    //对象公用锁
    private Object lock = new Object();
    //String类型的数组，存储队列元素
    private String[] elems = null;
    //队首位置
    private int head = 0;
    //队尾位置
    private int tail = 0;
    //存储的元素个数
    private int size = 0;
    
    //构造方法，用于构建定长数组，数组长度由参数指定
    public MyBlockingQueue(int capacity) {
        elems = new String[capacity];
    }
 
    //入队方法
    public void put(String elem) throws InterruptedException {
        synchronized(lock) {
            //已满时入队操作阻塞
            while(size == elems.length) {
                lock.wait();
            }
            //将元素存入队尾
            elems[tail] = elem;
            //存入后，队尾位置后移一位
            tail++;
            //实现环形队列的关键，超过数组长度后回归数组首位
            if(tail >= elems.length) {
                //回归数组首位
                tail = 0;
            }
            //存入后元素总数加一
            size++;
            //当出队操作阻塞时，入队后为其解除阻塞
            //（入队后队列不为空了）
            lock.notify();
        }
    }
    
    //出队方法
    public String tack() throws InterruptedException {
        //存储取出的元素，默认为null
        String elem = null;
        synchronized (lock) {
            //队列为空时出队操作阻塞
            while (size == 0) {
                lock.wait();
            }
            //出队，取出队首值（不用置空，队尾存入时覆盖）
            elem = elems[head];
            //出队后，队首位置后移一位
            head++;
            //实现环形队列的关键，超过数组长度后回归数组首位
            if(head == elems.length) {
                //回归数组首位
                head = 0;
            }
            //存入后元素总数加一
            size--;
            //当入队操作阻塞时，出队后为其解除阻塞
            //（出队后队列不满）
            lock.notify();
        }
        //返回取出的元素
        return elem;
    }
}

在普通队列上再进行修改，修改思路是这样

首先，我们上述代码中用的是if而非while，但是wait被唤醒的方法不止notify一种，它还可能被interrupt所‘唤醒’，这样的话可能队列还是空的或者满的就进行的取和入的操作，所以要用while，再进行一次判断，判断它是不是由于notify所唤醒
再者，如果这里用的的try catch的话，用if和while的区别就很大了，因为用throw抛出异常的话，会直接报错，而用try case捕捉异常的话就没有报错，后面会留下祸患
 

 

还有一个问题，因为我们在进行实现代码的时候，会进行多次判断例如：

这样的判断，当t1进行读取内存中的数据的时候，t2可能刚修改完cpu中的数据，还没来的及去修改内存中的数据，这样的话，t1就还是原来的size，这个就是内存可见性的案例，所以我们可以在变量前面添加volatile进行修饰，来避免这个问题
 

综上所述，我们得出的最终代码如下所示:

package thread;
 
// 此处不考虑泛型参数, 只是基于 String 进行存储.
class MyBlockingQueue {
    private String[] data = null;
    private volatile int head = 0;
    private volatile int tail = 0;
    private volatile int size = 0;
 
    private Object locker = new Object();
 
    public MyBlockingQueue(int capacity) {
        data = new String[capacity];
    }
 
    public void put(String s) throws InterruptedException {
        // 加锁的对象, 可以单独定义一个, 也可以直接就地使用 this.
        synchronized (this) {
            while (size == data.length) {
                // 队列满了
                // return;
                this.wait();
            }
            data[tail] = s;
            tail++;
            if (tail >= data.length) {
                tail = 0;
            }
            size++;
 
            this.notify();
        }
    }
 
    public String take() throws InterruptedException{
        String ret = "";
        synchronized (this) {
            while (size == 0) {
                // 队列为空
                // return null;
                this.wait();
            }
            ret = data[head];
            head++;
            if (head >= data.length) {
                head = 0;
            }
            size--;
 
            this.notify();
        }
        return ret;
    }
}
 
 
public class Demo28 {
    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(1000);
 
        // 生产者线程
        Thread t1 = new Thread(() -> {
            int i = 1;
            while (true) {
                try {
                    queue.put("" + i);
                    System.out.println("生产元素 " + i);
                    i++;
 
                    // 给生产操作, 加上 sleep, 生产慢点, 消费快点
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
 
        // 消费者线程
        Thread t2 = new Thread(() -> {
            while (true) {
                try {
                    Integer i = Integer.parseInt(queue.take());
                    System.out.println("消费元素 " + i);
 
                    // 给消费操作, 加上 sleep, 生产快点, 消费慢点
                    // Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
 
        t1.start();
        t2.start();
    }
}