JavaEE多线程:阻塞队列BlockingQueue

目录

1.了解阻塞队列

2.生产者消费者模型又是什么？ 

2.1生产者消费者模型的优点

2.1.1降低服务器与服务器之间耦合度

 2.1.2“削峰填谷”平衡消费者和生产的处理能力

3.标准库中的阻塞队列（BlockingQueue）

 3.1基于标准库（BlockingQueue）实现一个简单的阻塞队列的应用

 4.基于循环队列模拟是实现阻塞队列

 4.1 put()方法

4.2 take()方法

4.3put方法和take方法是如何被唤醒

---

1.了解阻塞队列

阻塞队列是一种特殊的队列. 也遵守 "先进先出" 的原则. 

阻塞队列顾名思义就是带有阻塞特性的队列的，它是如何运行的呢？

答： 1.当队列为空时，尝试出队列，就会发生阻塞，直到队列不空为止。

        2.当队列为满时，尝试入队列，就会发送阻塞，直到队列不满为止

阻塞队列在 多线程中非常有用，它可以调节生产者和消费者之间的关系，当生产者生产的资源太多，以至于消费者无法完全消费完，那么可以让生产者阻塞，达到生产、消费平衡的作用。

2.生产者消费者模型又是什么？ 

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

它将所有工作分工明确，参考下面这个例子：

有4个人我们一起包饺子，我们每个人都各顾各的，每个人都得做擀面皮和包饺子，如果这样的话，每个人的包饺子的效率都不是很高。若我们进行分工，1个人负责擀面皮，3个人负责包饺子，擀面皮的不管包饺子的是如何包，包饺子的不管擀面皮的是如何擀的。在第二种方式中擀面皮的就可以看作生产者，包饺子的就可以看作消费者。

2.1生产者消费者模型的优点

2.1.1降低服务器与服务器之间耦合度

看如下解读：开始时A服务器和B服务器通过阻塞对象进行交互，若B服务器挂了，那么我们可以迅速的做出反应，参考阻塞队列进行重新加载B服务器。还有就是若A服务器业务服务器，B服务器是实现业务的服务区，因为业务服务器需要时常更新代码，因此可以阻塞队列相当于一个缓存区。

 2.1.2“削峰填谷”平衡消费者和生产的处理能力

 比如在 "秒杀" （购买物品）场景下, 服务器同一时刻可能会收到大量的支付请求. 如果直接处理这些支付请求,服务器可能扛不住(每个支付请求的处理都需要比较复杂的流程). 这个时候就可以把这些请求都放到一个阻塞队列中, 然后再由消费者线程慢慢的来处理每个支付请求.在这里就起到了削峰的效果。

“ 削峰填谷”我们可以联想到三峡大坝：

三峡可谓是“功在当代，利在千秋”，为何这样说?三峡大坝它可不仅仅只是发电用的，它还有其他的功能。当到达雨季时，三峡上游的水向下流的会比往常多，这时开始关闸蓄水，防止下游发生洪灾，这就是“削峰”，过了雨季，下游需要灌溉庄稼，但是下游水又少了，这时打开闸门进行放水，造福下游，这就是“填谷”。

在购物系统中，我们就可以将秒杀时突然猛增的用户请求比喻成雨季的雨水，我们通过阻塞队列进行蓄洪，当在平常时，没那么多的用户请求时，我们就是慢慢的解决用户的请求比喻成开闸放水。

3.标准库中的阻塞队列（BlockingQueue）

 BlockingQueue的成员：

• ArrayBlockingQueue：基于数组的阻塞队列实现
• LinkedBlockingQueue：基于链表的阻塞队列
• PriorityBlockingQueue：基于优先队列的阻塞队列
• BlockingQueue：这个是实现阻塞队列的接口，继承与Queue

常用方法介绍：

注意：在这个些方法中，只有put()方法和take()方法带有阻塞特性，其他的方法不具备阻塞特性。 

 3.1基于标准库（BlockingQueue）实现一个简单的阻塞队列的应用

1.  
   import java.util.concurrent.BlockingDeque;
2.  
   import java.util.concurrent.BlockingQueue;
3.  
   import java.util.concurrent.LinkedBlockingDeque;
4.  
    
5.  
   public class testDemo1 {
6.  
   public static void main(String[] args) throws InterruptedException {
7.  
   BlockingQueue<Integer> queue=new LinkedBlockingDeque<>();
8.  
   //生产者
9.  
   Thread producer=new Thread(()->{
10.  
    int i=0;
11.  
    while (true) {
12.  
    try {
13.  
    System.out.println("生产：" i);
14.  
    queue.put(i);
15.  
    i ;
16.  
    Thread.sleep(1000);
17.  
    } catch (InterruptedException e) {
18.  
    e.printStackTrace();
19.  
    }
20.  
    }
21.  
    });
22.  
    producer.start();
23.  
    //消费者
24.  
    Thread customer=new Thread(()-> {
25.  
    while (true) {
26.  
    try {
27.  
    int value=queue.take();
28.  
    System.out.println("消费:" value);
29.  
    } catch (InterruptedException e) {
30.  
    e.printStackTrace();
31.  
    }
32.  
    }
33.  
    });
34.  
     
35.  
    customer.start();
36.  
    }
37.  
    }

代码运行结果：

 4.基于循环队列模拟是实现阻塞队列

这此次模拟实现中，仅模拟了put()方法，和take()方法。

若有对循环队列不了解的话，可以参考下方链接了解循环队列：【数据结构趣味多】循环队列https://blog.csdn.net/qq_65228171/article/details/128613550?spm=1001.2014.3001.5501

代码实现如下： 

1.  
   public class MyBlockingDeque {
2.  
   private int[] items=new int[1000];
3.  
   //头指针
4.  
   volatile private int head;
5.  
   //尾指针
6.  
   volatile private int tail;
7.  
   //队列中元素数量
8.  
   volatile private int size;
9.  
   synchronized public void put(int value) throws InterruptedException {
10.  
    //如果队列已满，阻塞等待
11.  
    while (size == items.length) {
12.  
    this.wait();
13.  
    }
14.  
    items[tail]=value;
15.  
    tail ;
16.  
    //如果队列满了，从头再来（循坏队列思想）
17.  
    if(tail==items.length) {
18.  
    tail=0;
19.  
    }
20.  
    size ;
21.  
    //唤醒take
22.  
    this.notify();
23.  
    }
24.  
    synchronized public int take() throws InterruptedException {
25.  
    while (size==0){
26.  
    //等待放入元素，等待put唤醒
27.  
    this.wait();
28.  
    }
29.  
    int value=items[head];
30.  
    head ;
31.  
    //如果头到了队尾，重新加载
32.  
    if(head == items.length) {
33.  
    head=0;
34.  
    }
35.  
    size--;
36.  
    this.notify();
37.  
    return value;
38.  
    }
39.  
    }

 4.1 put()方法

put 方法是向队列中存放数据，因此第一步我们需要判断队列满没满，若队列满了的话，就让队列通过wait方法进入阻塞状态，等待唤醒。若队列没有满，就将数据放入到尾指针所指位置，让尾指针向后移动一位；当尾指针达到数组的长度时，我们就需要重置尾指针了。

1.  
   synchronized public void put(int value) throws InterruptedException {
2.  
   //如果队列已满，阻塞等待
3.  
   while (size == items.length) {
4.  
   this.wait();
5.  
   }
6.  
   items[tail]=value;
7.  
   tail ;
8.  
   //如果队列满了，从头再来（循坏队列思想）
9.  
   if(tail==items.length) {
10.  
    tail=0;
11.  
    }
12.  
    size ;
13.  
    //唤醒take
14.  
    this.notify();
15.  
    }

4.2 take()方法

take 方法是在队列中拿取数据，有两种情况，若队列空，发生阻塞，直到put进来数据，若不为空，拿到数据，头指针向后走，如果头指针达到数组长度，重置头指针，最后返回value。

1.  
   synchronized public int take() throws InterruptedException {
2.  
   while (size==0){
3.  
   //等待放入元素，等待put唤醒
4.  
   this.wait();
5.  
   }
6.  
   int value=items[head];
7.  
   head ;
8.  
   //如果头到了队尾，重新加载
9.  
   if(head == items.length) {
10.  
    head=0;
11.  
    }
12.  
    size--;
13.  
    //唤醒put
14.  
    this.notify();
15.  
    return value;
16.  
    }

4.3put方法和take方法是如何被唤醒的

1.当存入数据时，队列为满的时候，队列就会通过wait方法使线程进入阻塞态，它什么时候被唤醒呢？队列被拿走一个数据时，被唤醒，这个唤醒是在take中唤醒的，take拿走一个数据队列不为满，唤醒put。

2. 当拿去数据时，队列为空的时候，队列会通过while判断，让线程进入阻塞态，当有新的数据被放入时，take就会被唤醒。

---

写在最后：

这篇好文章是转载于：学新通技术网