【多线程-从零开始-捌】阻塞队列，消费者生产者模型

什么是阻塞队列

阻塞队里是在普通的队列（先进先出队列）基础上，做出了扩充

1. 线程安全
   • 标准库中原有的队列 Queue 和其子类，默认都是线程不安全的
2. 具有阻塞特性
   • 如果队列为空，进行出队列操作，此时就会出现阻塞。一直阻塞到其他线程往队列里添加元素为止
   • 如果队列满了，进行入队列操作，此时就会出现阻塞。一直阻塞到其他线程从队列里取走元素为止

基于阻塞队列，最大的应用场景，就是实现“生产者消费者模型”（日常开发中，常见的编程手法）

生产者消费者模型

比如：

小猪佩奇一家准备包饺子，成员有佩奇，猪爸爸和猪妈妈，外加一个桌子

• 佩奇负责擀面皮
• 猪爸爸和猪妈妈负责包饺子
• 桌子用来放你擀好的面皮 每次佩奇擀好一个面皮后，就放在桌子上，猪爸爸和猪妈妈就用这个面皮包出一个饺子

此时：

• 佩奇就是面皮的生产者——生产者
• 猪爸爸和猪妈妈就是面皮的消费者——消费者
• 桌子就是阻塞队列——阻塞队列

为什么是是阻塞队列而不是普通队列？

因为阻塞队列可以很好的协调生产者和消费者

• 若佩奇擀面皮很快，不一会桌子上就满了 - 阻塞队列：佩奇就休息一下，等面皮被消耗一些之后继续再擀 - 普通队列：不会停，放不下了也一直擀
• 若猪爸爸和猪妈妈包的很快，不一会桌子上就空了 - 阻塞队列：猪爸爸和猪妈妈休息一下，等到面皮擀出来之后再包 - 普通队列：不会停，没面皮了也一直包

好处

上述生产者消费者模型在后端开发中，经常会涉及到

当下后端开发，常见的结构——“分布式系统”，不是一台服务器解决所有问题，而是分成了多个服务器，服务器之间相互调用

主要有两方面的好处

1. 服务器之间解耦合

我们希望见到“低耦合”

• 模块之间的关联程度/影响程度

通常谈到的“阻塞队列”是代码中的一个数据结构

但是由于这个东西太好用了，以至于会把这样的数据结构单独封装成一个服务器程序，并且在单独的服务器机器上进行部署

此时，这样的饿阻塞队列有了一个新的名字，“消息队列”（Message Queue，MQ）

如果是直接调用：

image.png|354

编写 A 和 B 代码中，会出现很多对方服务器相关的代码

并且，此时如果 B 服务器挂了，A 可能也会直接受到影响

再并且，如果后续想加入一个 C 服务器，此时对 A 的改动就很大

如果是通过阻塞队列：

image.png|526

A 之和队列通信

B 也只和队列通信

A 和 B 互相不知道对方的存在，代码中就更没有对方的影子

看起来，A 和 B 之间是解耦合了，但是 A 和队列，B 和队列之间，不是引入了新的耦合吗？

耦合的代码，在后续的变更工程中，比较复杂，容易产生 bug

但消息队列是成熟稳定的产品，代码是稳定的，不会频繁更改。A、B 和队列之间的耦合，对我们的影响微乎其微

再增加 C 服务器也很方便，也不会影响到原有的 A 和 B 服务器

2. “削峰填谷”的效果

通过中间的阻塞队列，可以起到削峰填谷的效果，在遇到请求量激增突发的情况下，可以有效保护下游服务器，不会被请求冲垮

阻塞队列的作用就相当与三峡大坝在三峡的防汛作用

image.png

A 向队列中写入数据变快了，但是 B 仍然可以按照原有的速度来消费数据

阻塞队列扛下了这样的压力，就像三峡大坝抗住上游的大量水量的压力

如果是直接调用，A 收到多少请求，B 也收到多少，那很可能直接就把 B 给搞挂了

当 A 不再写入数据的时候，但队列中还存有数据，可以继续工给 B

问题

1. 为啥一个服务器，收到的请求变多，就容易挂？
   • 一台服务器，就是一台“电脑”，上面就提供了一些硬件资源（包括但不限于 CPU，内存，硬盘，网络带宽......）
   • 就算你这个及其配置再好，硬件资源也是有限的
   • 服务器每次收到一个请求，处理这个请求的过程，就都需要执行一系列的代码，在执行这些代码的过程中，就需要消耗一定的硬件资源（CPU，内存，硬盘，网络带宽......）
   • 这些请求小号的总的硬件资源的量，超过了及其能提供的上限，那么此时机器就会出现（卡死，程序直接崩溃等......）
2. 在请求激增的时候，A 为啥不会挂？队列为啥不会挂？反而是 B 更容易挂呢？
   • A 的角色是一个“网关服务器”，收到客户端的请求，再把请求转发给其他的服务器 - 这样的服务器里的代码，做的工作比较简单（单纯的数据转发），消耗的硬件资源通常更少 - 处理一个请求，消耗的资源更少，同样的配置下，就能支持更多的请求处理
   • 同理，队列其实也是比较简单的程序，单位请求消耗的硬件资源，也是比较少见的
   • B 这个服务器，是真正干活的服务器，要真正完成一系列的业务逻辑 - 这一系列的工作，代码量非常庞大，消耗的时间很多，消耗的系统硬件资源，也是更多的

类似的，像 MySQL 这样的数据库，处理每个请求的时候，做的工作就是比较多的，消耗的硬件资源也是比较多的，因此 MySQL 也是后端系统中，容易挂的部分

对应的，像 Redis 这种内存数据库，处理请求，做的工作远远少于 MySQL，消耗的资源更少，Redis 就比 MySQL 硬朗很多，不容易挂

代价

1. 需要更多的机器来部署这样的消息队列（小代价）
2. A 和 B 之间的通信延迟会变长
   • 对于 A 和 B 之间的调用，要求响应时间比较短就不太适合了

每个技术都有优缺点，不能无脑吹，也不能无脑黑

比如：微服务

• 本质上就是把分布式系统服务拆的更细了，每个服务都很小，只做一项功能
• 非常适合大公司，部门分的很细
• 但需要更多的机器，处理请求需要更多的响应时间，更复杂的后端结构，运维成本水涨船高

Java 自带的阻塞队列

阻塞队列在 Java 标准库中也提供了现成的封装——BlockingQueue

image.png|565

BlockingQueue 本质上是一个接口，不能直接 new，只能 new 一个类

因为是继承与 Queue，所以 Queue 的一些操作，offer、poll 这些，在 BlockingQueue 中同样可以使用（不过不建议使用，因为都不能阻塞）

BlockingQueue 提供了另外两个专属方法，都能阻塞

- `put`——入列

- `take`——出队列

BlockingQueue<String> queue = new ArrayBlockingQueue<>(1000);

capacity 指的是容量，是一个需要加上的参数

public class Demo10 {  
    public static void main(String[] args) throws InterruptedException {  
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);  
        queue.put("111");  
        System.out.println("put成功");  
        queue.put("111");  
        System.out.println("put成功");  
      	
    }
}
//运行结果
put成功
put成功
put成功

只打印了三个，说明第四次 put 的时候容量不够，阻塞了

public class Demo10 {  
    public static void main(String[] args) throws InterruptedException {  
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(3);  
        queue.put("111");  
        System.out.println("put 成功");  
        queue.put("111");  
        System.out.println("put 成功");  
        
        queue.take();  
        System.out.println("take 成功");  
        queue.take();  
        System.out.println("take 成功");  
        queue.take();  
        System.out.println("take 成功");  
    }
}
//运行结果
put 成功
put 成功
take 成功
take 成功

由于只有 put 了两次，所以也只有两次 take，随后阻塞住了

public class Demo11 {  
    public static void main(String[] args) {  
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(1000);  
  
        Thread t1 = new Thread(() -> {  
            int i = 1;  
            while(true){  
                try {  
                    queue.put(i);  
                    System.out.println("生产者元素"+i);  
                    i++;  
                    Thread.sleep(1000);  
                } catch (InterruptedException e) {  
                    throw new RuntimeException(e);  
                }            
            }        
        });        
        Thread t2 = new Thread(() -> {  
            while(true) {  
                try {  
                    Integer i = queue.take();  
                    System.out.println("消费者元素"+i);  
                } catch (InterruptedException e) {  
                    throw new RuntimeException(e);  
                }            
            }        
        });        
    	t1.start();  
        t2.start();  
    }
}

上述程序中，一个线程生产，一个线程消费

实际开发中，通常可能是多个线程生产，多个线程消费

自己实现一个阻塞队列

普通队列

基于数组的队列

实现一个基础的队列

//此处不考虑泛型参数，只是基于 String 进行存储  
class MyBlockingQueue {  
    private String[] data = null;  
    private int head = 0;  
    private int tail = 0;  
    private int size = 0;  
    
    public MyBlockingQueue(int capacity) {  
        data = new String[capacity];  
    }    
    
    public void put(String s) {  
        if(size == data.length) {  
            //队列满了  
            return;  
        }        
        data[tail] = s;  
        tail++;  
        if(tail >= data.length){  
            tail = 0;  
        }        
        size++;  
    }    
    
    public String take() {  
        if(size == 0) {  
            //队列为空  
            return null;  
        }        
        String ret = data[head];  
        head++;  
        if(head >= data.length){  
            head = 0;  
        }        
        size--;  
        return ret;  
    }
}

阻塞队列

• 队列为空，take 就要阻塞，在其他线程 put 的时候唤醒//此处不考虑泛型参数，只是基于 String 进行存储 class MyBlockingQueue { private String[] data = null; private int head = 0; private int tail = 0; private 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 (locker) { if (size == data.length) { //队列满了，需要阻塞 //return; locker.wait(); } data[tail] = s; tail++; if (tail >= data.length) { tail = 0; } size++; //唤醒 take 的阻塞 locker.notify(); } } public String take() throws InterruptedException { String ret = ""; synchronized (locker) { if (size == 0) { //队列为空,需要阻塞 //return null; locker.wait(); } ret = data[head]; head++; if (head >= data.length) { head = 0; } size--; //唤醒 put 的阻塞 locker.notify(); } return ret; } }
• 队列未满，put 就要阻塞，在其他线程 take 的时候唤醒