1. 什么是阻塞队列
阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。
- 支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满。
- 支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。
阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。
在阻塞队列不可用时，这两个附加操作提供了4种处理方式，如下所示:
阻塞队列方法
2. Java里的阻塞队列
常见的4种阻塞队列。
- ArrayBlockingQueue:由数组支持的有界队列
- LinkedBlockingQueue:由链接节点支持的可选有界队列
- PriorityBlockingQueue:由优先级堆支持的无界优先级队列
- DelayQueue:由优先级堆支持的、基于时间的调度队列
2.1 ArrayBlockingQueue
ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。
默认情况下不保证线程公平的访问队列，所谓公平访问队列是指阻塞的线程，可以按照 阻塞的先后顺序访问队列，即先阻塞线程先访问队列。非公平性是对先等待的线程是非公平 的，当队列可用时，阻塞的线程都可以争夺访问队列的资格，有可能先阻塞的线程最后才访问队列。为了保证公平性，通常会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列。
2.1.1 队列创建
ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
访问者的公平性是使用可重入锁实现的，代码如下
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
2.1.2 应用场景
在线程池中有比较多的应用，生产者消费者场景
2.1.3 工作原理
基于ReentrantLock保证线程安全，根据Condition实现队列满时的阻塞
2.2 LinkedBlockingQueue
LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为 Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。
2.3 PriorityBlockingQueue
PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。也可以自定义类实现compareTo()方法来指定元素排序规则，或者初始化 PriorityBlockingQueue时，指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。
2.4 DelayQueue
DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队 列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。
2.4.1 队列创建
BlockingQueue<String> blockingQueue = new DelayQueue();
2.4.2 应用场景
- 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询 DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。
- 定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从 DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的。
3. 多线程生产者-消费者示例
接下来我们创建一个由两部分组成的程序 - 生产者 ( Producer ) 和消费者 ( Consumer ) 。
创建一个生产者用于生产苹果，一个消费者用于吃苹果，用阻塞队列存放。
Produce类:
package BlockingQueue;
import java.util.concurrent.BlockingQueue;
public class Product implements Runnable {
private BlockingQueue<String> queue;
public Product(BlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 开始生产苹果 ");
try {
queue.put("国光");
System.out.println(Thread.currentThread().getName() + " 放入成功");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
Consumer类:
package BlockingQueue;
import java.util.concurrent.BlockingQueue;
public class Consumer implements Runnable {
private BlockingQueue<String> queue;
public Consumer(BlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 开始吃苹果");
try {
queue.take();
System.out.println(Thread.currentThread().getName() + " 消费成功");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
测试类:
package BlockingQueue;
import java.util.concurrent.*;
public class Test {
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(100);
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);
for (int i = 0; i < 20; i++) {
Thread thread = new Thread(new Product(blockingQueue));
executorService.submit(thread);
}
for (int i = 0; i < 20 ; i++) {
Thread.sleep(1000);
Thread thread1 = new Thread(new Consumer(blockingQueue));
executorService.submit(thread1);
}
executorService.shutdown();
}
}
输出结果:
/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/bin/java "-javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=54787:/Applications/IntelliJ IDEA.app/Contents/bin" -Dfile.encoding=UTF-8 -classpath /Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/cldrdata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/jaccess.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/nashorn.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/zipfs.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jce.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jfxswt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/packager.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/tools.jar:/Users/zhangyunzheng/IdeaProjects/Thread/target/classes BlockingQueue.Test
pool-1-thread-1 开始生产苹果
pool-1-thread-2 开始生产苹果
pool-1-thread-1 放入成功
pool-1-thread-3 开始生产苹果
pool-1-thread-4 开始生产苹果
pool-1-thread-2 放入成功
pool-1-thread-5 开始生产苹果
pool-1-thread-4 放入成功
pool-1-thread-3 放入成功
pool-1-thread-6 开始生产苹果
pool-1-thread-5 放入成功
pool-1-thread-6 放入成功
pool-1-thread-7 开始生产苹果
pool-1-thread-8 开始生产苹果
pool-1-thread-9 开始生产苹果
pool-1-thread-7 放入成功
pool-1-thread-9 放入成功
pool-1-thread-8 放入成功
pool-1-thread-10 开始生产苹果
pool-1-thread-10 放入成功
pool-1-thread-11 开始生产苹果
pool-1-thread-12 开始生产苹果
pool-1-thread-13 开始生产苹果
pool-1-thread-14 开始生产苹果
pool-1-thread-15 开始生产苹果
pool-1-thread-16 开始生产苹果
pool-1-thread-17 开始生产苹果
pool-1-thread-18 开始生产苹果
pool-1-thread-19 开始生产苹果
pool-1-thread-20 开始生产苹果
pool-1-thread-21 开始吃苹果
pool-1-thread-21 消费成功
pool-1-thread-11 放入成功
pool-1-thread-22 开始吃苹果
pool-1-thread-22 消费成功
pool-1-thread-12 放入成功
pool-1-thread-23 开始吃苹果
pool-1-thread-23 消费成功
pool-1-thread-13 放入成功
pool-1-thread-24 开始吃苹果
pool-1-thread-24 消费成功
pool-1-thread-14 放入成功
pool-1-thread-25 开始吃苹果
pool-1-thread-25 消费成功
pool-1-thread-15 放入成功
pool-1-thread-26 开始吃苹果
pool-1-thread-26 消费成功
pool-1-thread-16 放入成功
pool-1-thread-27 开始吃苹果
pool-1-thread-27 消费成功
pool-1-thread-17 放入成功
pool-1-thread-28 开始吃苹果
pool-1-thread-28 消费成功
pool-1-thread-18 放入成功
pool-1-thread-29 开始吃苹果
pool-1-thread-29 消费成功
pool-1-thread-19 放入成功
pool-1-thread-30 开始吃苹果
pool-1-thread-30 消费成功
pool-1-thread-20 放入成功
pool-1-thread-31 开始吃苹果
pool-1-thread-31 消费成功
pool-1-thread-32 开始吃苹果
pool-1-thread-32 消费成功
pool-1-thread-33 开始吃苹果
pool-1-thread-33 消费成功
pool-1-thread-34 开始吃苹果
pool-1-thread-34 消费成功
pool-1-thread-35 开始吃苹果
pool-1-thread-35 消费成功
pool-1-thread-36 开始吃苹果
pool-1-thread-36 消费成功
pool-1-thread-37 开始吃苹果
pool-1-thread-37 消费成功
pool-1-thread-38 开始吃苹果
pool-1-thread-38 消费成功
pool-1-thread-39 开始吃苹果
pool-1-thread-39 消费成功
pool-1-thread-40 开始吃苹果
pool-1-thread-40 消费成功
Process finished with exit code 0
可以看到由于阻塞队列的大小是10，那么当放入10个线程以后，剩下的线程会被阻塞，直到队列中有空间(消费线程)之后，才会继续放入。
