【JavaEE】多线程 -- 阻塞队列

阻塞队列是什么

• 在数据结构的中我们已经讲过队列这种数据结构. 我们知道队列遵守先进先出的原则(FIFO), 同理, 我们阻塞队列也遵守这个原则.

• 阻塞队列是一个线程安全(为啥线程安全后面说)的数据结构. 具有下面两个特性

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

• 阻塞队列的⼀个典型应⽤场景就是 “⽣产者消费者模型”. 这是⼀种⾮常典型的开发模型.

生产者消费者模型

好处1:避免抢夺资源导致效率降低

• 我们过年的时候在北方通常都要包饺子. 而包饺子通常涉及这几个步骤

  • 擀饺子皮
  • 用饺子皮包肉馅, 生成饺子.

• 方案1: 一个擀面杖, 我和小美和千均. 在桌子上谁先拿到擀面杖, 就先做出一个饺子皮. 让后包饺子. 所以我们三个要争夺这个擀面杖. 如果我抢到了这个擀面杖, 他们就只能干等着. 因为他们没有饺子皮(这个行为就浪费时间效率)

• 方案2: 我负责擀饺子皮, 小美和千均负责包饺子. 我把擀好的饺子皮放在簸箕上. 小美和千均用的时候就从簸箕上拿. 他们两个不用和我争夺擀面杖, 他们的任务是只是包饺子, 不需要擀饺子皮. 这个行为他们他们就可以不用花费很多时间等我, 因为我只是擀饺子皮, 不用包饺子. (和上个方案对比)

• 这里我就是生产者, 小美和千均就是消费者. 簸箕是阻塞队列

• 这两个方案明显第二个方案更好

• 因为第二个方案不需要争夺擀面杖(抢夺资源), 导致没有抢夺到资源的线程干等着浪费时间.

好处2:降低耦合

第二个就是耦合性低

• 解耦合我们用一个具体的例子来说: 可以看到我们如果有了商品订单, 那么库存的商品数量就应该减少. 那么有了商品订单我们管理订单的服务器会知道, 这个时候会通知库存服务器修改库存数量. 但是这个时候突然库存管理服务器出现了BUG, 就需要修改库存服务器的代码, 由于订单服务器肯定涉及到一些访问库存服务器的代码, , 那么库存服务器代码修改了. 我们订单服务器也需要进行修改. 这样的操作就非常复杂, 我们不希望程序之间相互影响太大.

• 我们采用消息队列(阻塞队列的plus升级版), 让他作为订单服务器和库存服务器之间的通信桥梁. 订单服务器要访问库存服务器就从消息队列中拿. 库存服务器需要通知订单服务器就往消息队列中加载数据通知. 这个时候库存服务器再出现了什么BUG. 就不需要也修改订单服务器的代码.(他们之间不是之间交交互了), 有人可能说, 那订单服务器和消息队列耦合了啊. 其实我们主要怕的不是耦合, 而是耦合后的代码修改, 像之前订单服务器和库存服务器直接交互的时候. 因为他们两个都涉及全部业务. 那么直接修改他们之间的一个就可能会影响另一个服务器的业务. 这个时候就需要也同时修改另一个业务逻辑代码. 而消息队列只负责出队列和入队列数据, 没有任何业务逻辑代码. 所以我们不必担心修改另一个服务器的代码会影响消息队列.
  

好处3:削峰填谷

• 我们经常在开学的那几天都要进选课系统进行选课, 那么这个系统平时没有什么人用. 就基本上可以正常运行.
• 但是到了选课的时候, 几乎全校的人都要在同一个时间进这个系统进行选课. 这个时候我们服务器需要处理的请求量就递增升高了.
  
• 这个时候B服务器的代码任务重, 请求量又在同个时间多这么多. 服务器就有可能因为处理不过来导致崩溃
  
• 在我们生产者消费者模型中, 我们引入了阻塞队列. 这个时候既然请求量非常多, 可能会导致我们B服务器崩溃, 那么我们就不直接把请求一下子全部发给B服务器处理. 而是全部放在阻塞队列中(阻塞队列的任务只有出入队列, 所以可以承载这么多请求), 让B服务器按照自己的处理速度来处理请求, 保证B服务器不会因为处理不过来崩溃, 根据我们阻塞队列的特点, 当阻塞队列满了, 那么就不再让A服务器把请求交给阻塞队列.
  

模型弊端

• 我们浏览器发送单个请求, 这个时候由于模型结构的阻塞队列, 我们这个请求不知道要在这个阻塞队列中待多久, 有可能他前面的请求排了很多, 就要等很久. 那么这个时候我们可能浏览器的请求都超时了, 发送的请求还在阻塞队列中排队.
• 所以这个模型只适合异步操作, 也就是我们的浏览器客户端不需要一直等待服务器发送给他结果才继续执行其他事情, 而是浏览器客户端发送了这个请求就继续做其他事情了(比如打开网页), 在做其他事情的过程中收到服务器处理请求后的结果.
• 而同步操作则是浏览器客户端发送请求后, 就什么事情都不干一直等待服务器的处理结果, 这个时候就可能等待超时了.

自主实现阻塞队列

• 阻塞队列要求线程安全, 我们这里put和take方法在同一个对象(成员变量一样)多线程场景下, 对同一个成员变量就行修改操作如data数组, 就有可能因为修改不具有原子性导致线程安全, 这里就通过加锁来保证了线程安全.
• 第二点就是阻塞队列要求队列满了不能进行put操作, 队列空了不能进行take操作, 那么我们判断一下通过wait方法让线程进入等待状态. 如队列满了, 让这个A线程进入等待状态, B线程去调用take出队列后唤醒A线程, 就能保证队列没有满, 这个时候A线程就可以继续入队列了

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 19182
 * Date: 2025-08-18
 * Time: 15:51
 */
public class MyBlockingQueue {
    private String[] data;  //存储数据
    private int head = 0;//队头
    private int tail = 0;   //队尾
    private int size = 0;   //有效元素个数
    private Object locker = new Object();
    public MyBlockingQueue(int capacity){
        if (capacity <= 0) {
            throw new IllegalArgumentException("capacity must be positive.");
        }
        data = new String[capacity];
    }
    public void put(String elem) throws InterruptedException {
        synchronized (locker){
            while(size == data.length){
                locker.wait(); //队列满了, 那么进入等待状态, 等待数据入队列
            }
            data[tail] = elem;
            tail++;
            if(tail >= data.length){    //循环队列满了
                tail = 0;
            }
            size++;
            locker.notify();
        }
    }
    public String take() throws InterruptedException {
        synchronized (locker){
            while(size == 0){
                locker.wait(); //队列空了, 没有数据可以出队列了, 进入等待状态
            }
            String ret = data[head];
            head++;
            if(head >= data.length){
                head = 0;
            }
            size--;
            locker.notify();
            return ret;
        }
    }

}

• 注意
  
• 这里不能使用if, 因为在多线程的情况下. 我们有可能导致每个线程被唤醒后, 由于另外一个线程的改变同一个数据. 导致唤醒后, 还是满足进入线程等待的条件.
• 例如: 有A, B, C三个线程. 这个时候阻塞队列满了, A, B线程执行put操作的时候都不能进行put, 那么A, B两个线程进入等待状态(不参与锁竞争). 这个时候只有C能拿到这把锁了, C进行take操作, 此时队列就不是满的了. 这个时候执行notify(随机唤醒线程), 唤醒到了A线程, A线程由于是if执行完了, 继续往下执行成功完成put操作, 这个时候队列又是满的了. 这个时候A线程执行notify唤醒到了B线程, B线程由于也是if直接往下执行. 结果去入队列就错误了. 这个时候队列是满的.
• 所以我们被唤醒后不能直接往下执行, 因为有可能还是满足等待的条件. 需要循环判断一下是否不满足等待的条件, 才能往下执行.