一.阻塞队列的介绍

1.线程是安全的

2.当进行入队操作的时候,队列为满,入队操作就阻塞,直到队列非满的时候入队操作才完成

3.当进行出队操作的时候,队列为空,出队操作就阻塞,直到队列非空的时候出队操作才完成

生产者消费者模型:

例如:3个人包饺子,其中有一个人需要生产饺子皮,他就是生产者,另外的两个人就是消费者,而生产者生产的饺子皮放在桌子上,桌子就是"交易场所".如果生产者生产过快饺子皮已经放满了桌子,他就能进行阻塞等待,如果是饺子皮的生产速度慢于包饺子的速度,那消费者就能够进行阻塞等待

这饺子看起来有点不太一样hhh

二.使用阻塞队列/生产者消费者模型的好处

1.使用阻塞队列,有利于代码"解耦合"

如果服务器A和服务器B直接进行通信,那么A的代码里面需要知道B的存在(A知B),B的代码里也需要知道A的存在(B知A),此时的耦合度就比较高,此时如果再多一个服务器C与A进行通讯,则A的代码又要更改,所以此处使用阻塞队列有很大好处

2.削峰填谷

按照没有生产者消费者模型的写法,如果外面用户请求量突增,外面流量过来的压力就会直接压到每一个服务器上,如果某一个服务器抗压能力不行,服务器就容易挂掉.

服务器每处理一个请求,都需要消耗一定的硬件资源(CPU,内存,硬盘,带宽......),同一个时刻的请求越多,所消耗的资源也就越多,一台主机的硬件资源是有限的,当资源耗尽的时候,服务器也就挂了.

分布式系统本质上就是加入了更多的硬件资源

例如:河流上游连续降雨,就会产生汛期,也就相当于是流量暴增,此时如果没有修水库的话,上游河流的水就会直接流到下游去,下游就会发生洪涝灾害,而修水库就是相当于阻塞队列,能够在汛期蓄水,减少河流留到下游的水量,旱期放水,保证下游的河流流量

※所以当流量突然暴涨的时候,有使用阻塞队列的话受到冲击的就服务器A和阻塞队列,而其他的服务器就比较不受到影响

三.阻塞队列的使用

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class Demo {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<>(100);
        //带有阻塞的入队操作
        blockingQueue.put(1);
        blockingQueue.put(2);
        blockingQueue.put(3);
        //带有阻塞的出队操作
        Integer ret = blockingQueue.take();
        System.out.println(ret);
        ret = blockingQueue.take();
        System.out.println(ret);
        ret = blockingQueue.take();
        System.out.println(ret);

        ret = blockingQueue.take();//此时就会阻塞等待了
        System.out.println(ret);
    }
}

运行结果:

当前面的三个数字都出队后,队列中就没有元素了,所以带有阻塞的出队操作就会一直等待下去

四.模拟实现阻塞队列

队列的特点是先进先出

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

    //入队列
    public void put(int elem) throws InterruptedException {
        synchronized (this){
            //判断队列是否为满,满了则不能插入
            while (size >= items.length){
                this.wait();//由take方法里的notify来唤醒
            }
            //进行插入操作,将elem放到items里,放到tail所指向的位置
            items[tail] = elem;
            tail++;
            if(tail >= items.length){
                tail = 0;
            }
            size++;
            this.notify();
        }
    }

    //出队列,返回删除的元素内容
    public Integer take() throws InterruptedException {
        synchronized (this){
            //判断队列是否为空,为空则不能出队
            while (size == 0){
                this.wait();//由put方法里的notify来唤醒
            }
            //非空,取元素
            int ret = items[head];
            head++;
            if(head >= items.length){
                head = 0;
            }
            size--;
            this.notify();
            return ret;
        }
    }
}

public class Demo2 {
    public static void main(String[] args) {
        MyBlockingQueue myBlockingQueue = new MyBlockingQueue();
        Thread producer = new Thread(() ->{//生产者
            int n = 1;
            while (true){
                try {
                    myBlockingQueue.put(n);
                    System.out.println("生产元素:" + n);
                    n++;
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        Thread customer = new Thread(() -> {//消费者
            while (true){
                try {
                    int n = myBlockingQueue.take();
                    System.out.println("消费元素:" + n);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        producer.start();
        customer.start();
    }
}

运行结果: