阻塞队列和生产消费模型

news2025/1/4 19:00:58

阻塞队列

阻塞队列的概念

队列相信我们已经不陌生了之前也学过很多队列比如: 普通队列 和 优先级队列 两种
这两种队列都是线程不安全的
而我们讲的阻塞队列 刚好可以解决线程安全问题也是先进先出并且带有阻塞功能.
阻塞功能是怎么回事呢

就是如果入队的时候阻塞队列为满 ,则当前线程就会进入阻塞状态,阻塞到有出队列操作才会再唤醒线程继续执行 ,相同如果出队时阻塞队列为空,则当前线程也会进入阻塞状态直到有新元素入队操作才会唤醒继续执行该线程.

在Java标准库中,BlockingQueue这个就是标准库提供的阻塞队列. 阻塞队列在实际开发中使用的场景十分广泛. 在阻塞方法中有独特的进队列和出队列的方法其他一些普通的方法阻塞队列也有 ,但是使用阻塞队列一般都不会再使用普通队列的方法了.

消息队列

消息队列这种数据结构其实也是阻塞队列的一种 ,也带有阻塞特性.它不是普通的先进先出,而是通过topic这样的参数来对数据进行分类,出队列的时候,指定topic,每个topic下的数据是先进先出. 消息队列只要应用于生产者消费者模型.

生产者消费者模型

什么是生产者和消费者模型呢 ?
举个例子:
就相当于两个人制作饺子要分为两步一步是擀饺子皮一步是包饺子. 要两个人分工合作,实际上要包饺子就得先擀饺子皮那擀饺子皮的人相当于生产者包饺子皮的人就相当于消费者擀好的饺子皮要用一个容器来盛放 ,而这个容器就相当于阻塞队列/消息队列. 有了这个容器就会有阻塞功能 ,现在A擀饺子皮 B包饺子如果B包饺子包得快 A擀饺子皮慢那容器里面的饺子皮很快就会空了这时候B就要阻塞等待A擀饺子皮每擀一个就包一个饺子 ,相反如果A擀饺子皮擀得比较快 B包饺子慢,那么容器很快就会满了,这样A就会阻塞等待 B包饺子 B每包一个 A就擀一个.

在这里插入图片描述
上面图片就是生产者和消费者模型

你们想一想在实际开发中为什么要用到这个生产者消费者模型 ? 相当于在开发中有两种服务器 :请求服务器Q和应用服务器Y 如果QY直接传递消息 ,没有阻塞队列的情况下,如果Q突然请求有很多的时候,Y的请求也就会跟着暴涨 ,但Y服务器是应用服务器,他处理任务的过程很复杂是重量级的这样就会让Y服务器直接G.

这时我们如果我们引入生产者消费者模型,这样即使Q服务器的请求再怎么多,也不会影响到Y服务器最坏的情况也就是Q服务器挂了但Y服务器不会有任何影响因为Q的请求都会放在阻塞队列中 , 有界的阻塞队列就会引起阻塞 Y服务器还是和原来一样处理这些请求这样就起到了"削峰填谷"的作用.
这个模型还有一个优势就是高内聚低耦合两个服务器不会直接接触一个挂了不会对另一个产生影响.

用循环队列实现一个简单的阻塞队列用于实现一个生产者消费者模型

第一种就是生产者生产快而消费者慢就会出现程序一运行生产者咔咔就会生产满了接下来就会阻塞消防者每消费一个生产者才会生产一个. 下面我们通过代码来实现一下:

class MyBlockingQueue{
    private String[] elems = null;
    private volatile int head = 0;
    private volatile int tail = 0;
    private volatile int size = 0;

    public MyBlockingQueue(int capacity) {
        elems = new String[capacity];
    }

    void put(String elem) throws InterruptedException {
        //先加锁
        synchronized (this) {
            while (size >= elems.length) {
                //队列满了 ,就要等待
                this.wait();
            }

            //把新的元素放到tail所在的位置上
            elems[tail] = elem;
            tail++;
            if (tail >= elems.length) {
                //到达末尾  ,就回到开头
                tail = 0;
            }
            //更新size的值
            size++;
            //唤醒下面take的wait
            this.notify();
        }
    }

    String take() throws InterruptedException {
        synchronized (this) {
            while (size == 0) {
                //为空要等待
                this.wait();
            }
            //取出 head指向的元素
            String result = elems[head];
            head++;
            if (head >= elems.length) {
                head = 0;
            }

            size--;
            //唤醒put上面的wait
            this.notify();
            return result;
        }

    }
}

public class Text2 {
    //实现一个阻塞队列 模拟一个生产者消费者模型
    public static void main(String[] args) {
        //实例化一个阻塞队列的对象   利用该阻塞队列实现一个生产者消费者模型
        MyBlockingQueue queue = new MyBlockingQueue(1000);

        //生产者线程
        Thread t1 = new Thread(() -> {
            try {
                int count = 0;
                while (true) {
                    queue.put(count + "");
                    System.out.println("生产:" + count);
                    count++;
                }
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        //消费者线程
        Thread t2 = new Thread(() ->{
            try {
                while (true) {
                    String result = queue.take();
                    System.out.println("消费:" + result );
                    Thread.sleep(1000);
                }
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        t1.start();
        t2.start();
    }
}