当前位置：首页 > news >正文

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

news 2025/5/20 6:18:30

一、阻塞队列

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

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() {@Overridepublic void run() {//时间到了之后要执行的代码System.out.println("hello timer 3000");}},3000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello timer 2000");}}, 2000);timer.schedule(new TimerTask() {@Overridepublic void run() {System.out.println("hello timer 1000");}},1000);System.out.println("hello main");}
}