阻塞队列：线程安全与生产者消费者模型解析

news2025/5/21 2:33:40

一、阻塞队列

阻塞队列就是基于普通队列做出扩展

1.线程安全的

如果针对一个已经满了的队列进行入队列，此时入队列操作就会阻塞，一直阻塞到队列不满（其他线程出队列元素）之后

如果针对一个已经空了的队列进行出队列，此时出队列操作就会阻塞，一直阻塞到队列不空（其他线程入队列元素）之后

包饺子

A和B和C在过年期间要包饺子，为了快一点三个人决定一起包饺子

但包饺子得先擀饺子皮，现在就只有一个擀面杖，所以A擀饺子皮是BC就陷入了阻塞，显然这就会出现竞争擀面杖问题（锁竞争）。

我们加入了一个桌子（阻塞队列），A负责擀饺子皮，BC通过桌子来包饺子，

如果A干的特别快，BC赶不上他的速度，很快桌子就会被放满（队列满了），这下A就会停下来（阻塞）

如果BC干的特别快，A赶不上他俩的速度，桌子为空（队列为空），BC就得等待（阻塞）

生产者消费者模型

1.引入生产者消费者模型，就可以更好的做到“解耦合”

上述过程中，A和B，A和C的耦合性较强，如果B或者C挂了，对于A的影响是很大的

加了阻塞队列A和B，C几乎几乎就没有交集了，B，C就算是挂了，对于A的影响也是微乎其微

阻塞队列（消息队列）

1.上述的阻塞队列，是基于这个数据结构实现的服务器程序，又被部署到单独的主机上

2.整个系统的结构更复杂了，你要维护的服务器更多了

3.效率，请求从A发过来B受到，这个过程经过转发要有一定开销

2.削峰填谷

当外网的请求数量就像洪水一样，急剧增加，这时A的请求数量就会增加很多，即使工作一般比较简单，每个请求消耗的资源少，但架不住多。B和C完成的工作更复杂，消耗的资源更多，这就可能导致BC挂了。

我们加入了阻塞队列之后

队列就像抗洪用的水库，只是用来存储数据的，消耗资源很少，抗压能力很强，在数据多的时候，存下来保持稳定的速度，少的是否就会放出数据，同样保持稳定请求速度，需要注意的是这个队列只是防止BC挂掉了。

Java标准库提供了现成的阻塞队列数据结构

使用put 和 offer一样都是入队列

但是put是带有阻塞功能，offer 没带阻塞（队列满了会返回结果）

take 方法用来出队列，同样带有阻塞功能

阻塞队列没有提供带有阻塞功能的获取队首元素的方法

尝试实现一个阻塞队列

既然涉及到了出队列，那数据结构就得用到循环队列，保证空间利用率

class MyBlockingQueue{
    private String[] elems = null;
    private int head = 0;
    private int tail = 0;
    private int size = 0;
    //准备锁对象，使用this也可以
    private Object locker = new Object();
    public MyBlockingQueue(int capacity){
        elems= new String[capacity];
    }
    public void put(String elem) throws InterruptedException {
            while(size >= elems.length){
           
            }
            //新的元素要放到tail指向的位置上
            elems[tail] = elem;
            tail++;
            if(tail >= elems.length){
                tail = 0;
            }
            size++;

    }
    public String take() throws InterruptedException {
        String elem = null;
            while(size == 0){
                
            }
            //取出 head 位置的元素返回
            elem = elems[head];
            head++;
            if(head >= elems.length){
                head = 0;
            }
            size--;
            return elem;
        }

    
}

现在我们就相当于实现了一个循环队列，我们需要加上阻塞效果，我们知道只有当队列满和队列空才会阻塞，而解除阻塞则需要队列中有一个元素出去和队列中有一个元素进来。

所以我们可以在put结束时加上notify，并在判断队列满了的时候进行阻塞，至于在判断条件加上while循环，是为了防止在多线程情况下被不小心解锁，毕竟，每次put，take之后都对解一次锁，为了保证代码的原子性，我们为整个方法都加上锁。试想一下如果没有这么加锁，就会引发数组越界等一系列问题。

take方法也是同理

至此我们可以来验证一下阻塞队列是否成功，我们创建了两个线程分别是t1，t2，t1线程为生产元素的线程并且要比t2线程消费元素快的多。

Thread t1 = new Thread(() -> {
            int n = 1;
            while(true){
                try {
                    queue.put(n + "");
                    System.out.println("生产元素" + n);
                    n++;

                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        //消费者
        Thread t2 = new Thread(() -> {
            while (true){
                try {
                    String n = queue.take();
                    System.out.println("消费元素" + n);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        t2.start();
        t1.start();

我们发现在一开始运行时，t1线程十分快都到了一千多了，但t2线程才到2，这就导致了t1线程要想继续添加元素，就得等t2线程，这就出现了消费一个元素，就生产一个元素。

定时器

public class ThreadDemo30 {
    public static void main(String[] args) {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                //时间到了之后要执行的代码
                System.out.println("hello timer 3000");
            }
        },3000);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("hello timer 2000");
            }
        }, 2000);
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("hello timer 1000");
            }
        },1000);
        System.out.println("hello main");
    }
}