1. 什么是阻塞队列
阻塞队列,从名字上看,它首先应该是一个队列,满足先进先出的原则.其次,我们来理解一下"阻塞"这个词.我们之前其实见过这个词,在介绍线程的状态时,我们讲过有一种状态就是阻塞状态.所谓的"阻塞",也就是一种等待,需要其他的线程进行唤醒,在这里也确实是这层意思.阻塞队列中存放的是Runnable对象,代表着需要进行的操作.
总结一下,阻塞队列就是一个可以内置阻塞状态的队列.
2. 阻塞队列的结构
我们刚刚提过:阻塞队列是一个队列,但它是一个普通的队列吗?并不是,阻塞队列实际上底层使用的是循环队列.为什么呢?因为阻塞队列需要"阻塞",什么时候阻塞呢?就是队列为满的时候,使用一个普通的队列,就不存在一个满的概念,除非人为设置一个阈值.在这种情况下,不如使用循环队列进行实现更为直观.
3. put和take方法
在阻塞队列中,内置了两个非常重要的方法put方法和take方法.
put方法
在put方法中,我们需要做的就是把Runnable对象添加进入阻塞队列.当队列不为满时,我们就可以将这个过程简要理解为对一个不满的循环队列添加元素.
但是,当队列满时,我们就没有办法再把Runnable对象添加进阻塞队列,那么调用put方法的这个线程就要进入阻塞等待状态.
take方法
在take方法中,我们需要做的就是把Runnable对象从阻塞队列中取出,当队列不为空时,我们就可以将这个过程简要理解为对一个不为空的循环队列删除元素.
同样,当队列空时,我们没法取出其中的Runnable对象,调用take方法的这个线程也要进入阻塞等待状态.
两个方法的联系
我们刚才还没有解决一个问题,调用put方法的线程和调用take方法的线程进入阻塞等待后,谁去唤醒它们,答案就是-------除了自己的另一个线程.
我们思考一下,对同一个循环队列进行put和take,上述两个线程中至多只有一个线程进入阻塞等待状态,因为同一个循环队列不能又空又满,这就意味着两个线程必定可以互相唤醒,相辅相成.
代码实现:
当然在这个过程中,要解决线程安全问题:
public class MyBlockingQueue {
private String[] elems;
int head = 0;
int tail = 0;
int size = 0;//通过计数实现循环队列
public MyBlockingQueue(int capacity) {
elems = new String[capacity];
}
public void put(String elem) throws InterruptedException {
synchronized (this) {
if (size >= elems.length) {
//队列满了
this.wait();
}
elems[tail] = elem;
tail++;
if (tail >= elems.length) {
tail = 0;
}
size++;
this.notify();
}
}
public String take() throws InterruptedException {
synchronized (this) {
if (size == 0) {
this.wait();
}
String res = elems[head];
head++;
if (head >= elems.length) {
head = 0;
}
size--;
this.notify();
return res;
}
}4. 阻塞队列的经典应用-----生产者消费者模型
所谓生产者消费者模型,就是利用循环队列,开启两个线程t1,t2.t1线程负责调用put方法,t2线程负责调用take方法,相当于线程t1进行生产,线程t2进行消耗,相辅相成.
下面直接给出代码:
public static void main(String[] args) {
MyBlockingQueue q = new MyBlockingQueue(10);
Thread t1 = new Thread(()->{
try {
int count = 1;
while(true) {
q.put(count+" ");
System.out.println("t1生产了"+count);
count++;
}
}catch(InterruptedException e) {
e.printStackTrace();
}
});
Thread t2 = new Thread(()->{
try {
while (true) {
String s = q.take();
System.out.println("t2消费了"+s);
}
}catch(InterruptedException e) {
e.printStackTrace();
}
});
t1.start();
t2.start();
}
}得到的效果如下图:
