简介
说明
本文用示例介绍Java中阻塞队列（BlockingQueue）的用法。
队列类型
BlockingQueue有这几种类型：ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue、PriorityBlockingQueue、DelayedWorkQueue。
| 队列类型 | 说明 |
| ArrayBlockingQueue | 基于数组的FIFO队列；有界；创建时必须指定大小； 入队和出队共用一个可重入锁。默认使用非公平锁。 |
| LinkedBlockingQueue | 基于链表的FIFO队列；有/无界；默认大小是 Integer.MAX_VALUE（无界），可自定义（有界）； 两个重入锁分别控制元素的入队和出队，用Condition进行线程间的唤醒和等待。 吞吐量通常要高于ArrayBlockingQueue。 默认大小的LinkedBlockingQueue将导致所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了）。当每个任务相互独时，适合使用无界队列；例如， 在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 |
| SynchronousQueue | 无缓存的等待队列；无界；可认为大小为0。 不保存提交任务，直接提交出去。若超出corePoolSize个任务，直接创建新线程来执行任务，直到(corePoolSize＋新建线程)> maximumPoolSize。 此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线 程具有增长的可能性。 吞吐量通常要高于LinkedBlockingQueue。 //也有地方说：是一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态 详见下边的：CachedThreadPool的execute流程 |
| PriorityBlockingQueue | 基于链表的优先级队列；有/无界；默认大小是 Integer.MAX_VALUE，可自定义； 类似于LinkedBlockingQueue，但是其所含对象的排序不是FIFO，而是依据对象的自然顺序或者构造函数的Comparator决定。 |
| DelayedWorkQueue |
常用方法
放入数据
| 方法 | 说明 |
| offer(E e) | 向队列尾部插入一个元素。该方法是非阻塞的。 如果队列中有空闲：插入成功后返回 true。 如果队列己满：丢弃当前元素然后返回false。 如果e元素为null：抛出NullPointerException异常。 |
| offer(E o, long timeout, TimeUnit unit) | 可以设定等待的时间，若在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。 |
| add(E e) | 内部调用offer方法。 与直接调用offer的区别： add：失败时，抛出异常 offer：失败时，返回false |
| put(E e) | 向队列尾部插入一个元素。 如果队列中有空闲：插入后直接返回。 如果队列己满：阻塞当前线程，直到队列有空闲插入成功后返回。 如果在阻塞时被其他线程设置了中断标志：被阻塞线程会抛出InterruptedException异常而返回。 如果e元素为null：抛出NullPointerException异常。 |
获取数据
| 方法 | 说明 |
| poll() | 获取当前队列头部元素并从队列里面移除它。 如果队列为空则返回null。 |
| poll(long timeout, TimeUnit unit) | 从BlockingQueue取出（会删除对象）一个队首的对象。 一旦在指定时间内有数据可取，则立即返回队列中的数据。 若直到时间超时还没有数据可取，返回失败。 |
| take() | 获取当前队列头部元素并从队列里面移除它。 如果队列为空则阻塞当前线程直到队列不为空然后返回元素； 如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。 |
| drainTo() | 一次性从BlockingQueue获取（会删除对象）所有可用的数据对象（可指定获取数据的个数）。 本方法可提升获取数据效率，不需要多次分批加锁或释放锁。 |
其他方法
| 方法 | 说明 |
| remainingCapacity() | 获取队列中剩余的空间 |
| contains(Object o) | 判断队列中是否拥有该值。 |
| remove(Object o) | 从队列中移除指定的值。 |
| size() | 获得队列中有多少值。 （返回AtomicLong的值） |
ArrayBlockingQueue
简介
ArrayBlockingQueue通过数组实现的FIFO有界阻塞队列，它的大小在实例被初始化的时候就被固定了，不能更改。
该类支持一个可选的公平策略，用于被阻塞等待的线程获取独占锁的排序，因为ArrayBlockingQueue内部的操作都需要获取一个ReentrantLock锁，该锁是支持公平策略的，所以ArrayBlockingQueue的公平策略就直接作用于ReentrantLock锁，决定线程是否有公平获取锁的权利。默认情况下是非公平的，公平模式下队列按照FIFO顺序授予线程访问权。公平性通常会降低吞吐量，但会降低可变性并避免饥饿。
ArrayBlockingQueue的缺陷
通过源码可以看见，ArrayBlockingQueue内部的几乎每一个操作方法都需要先获取同一个ReentrantLock独占锁才能进行，这极大的降低了吞吐量，几乎每个操作都会阻塞其它操作，最主要是插入操作和取出操作相互之间互斥。所以ArrayBlockingQueue不适用于需要高吞吐量的高效率数据生成与消费场景。LinkedBlockingQueue就能弥补其低吞吐量的缺陷。
实例
创建一个corePoolSize为2，maximumPoolSize为3的线程池。其中ArrayBlockingQueue设置缓存2个任务。执行6个任务。ArrayBlockingQueue为有界队列：
- 任务1和2在核心线程中执行；
- 任务3和4进来时，放到ArrayBlockingQueue缓存队列中，并且只能放2个（ArrayBlockingQueue设置的大小为2）；
- 任务5和6进来的时候，任务5新建线程来执行任务，已经达到最大线程数3，所以任务6拒绝；
- 当有线程执行完的时候，再将任务3和4从队列中取出执行
创建线程池代码如下：
/**
* ArrayBlockingQueue
*/
private static void arrayQueue() {
System.out.println("\n\n =======ArrayBlockingQueue====== \n\n");
Executor executors = new ThreadPoolExecutor(
2, 3, 30, TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(2),
new RejectHandler());
execute(executors);
}
执行结果如下
1 is running... 2 is running... 6 is rejected ^^ //6被拒 5 is running... //5新建线程执行 1 is end !!! 2 is end !!! 3 is running... //1和2执行完之后3和4才执行 4 is running... 5 is end !!!
LinkedBlockingQueue
简介
LinkedBlockingQueue和ArrayBlockingQueue的相同点：
- 是FIFO队列，不允许插入null值。
- 容量在实例被构造完成之后不允许被更改
不同点
| 项 | LinkedBlockingQueue | ArrayBlockingQueue |
| 大小指定 | 实例化时可指定队列大小。 若不指定大小，会采用默认的Integer.MAX_VALUE。 | 实例化时必须指定大小。 |
| 吞吐量 | 大。 采用了“双锁队列” 算法，元素的入队和出队分别由putLock、takeLock两个独立的可重入锁来实现。 | 小。 几乎每一个方法都需要先获取同一个ReentrantLock独占锁才能进行。 |
实例
创建一个corePoolSize为2，maximumPoolSize为3的线程池。无界队列。同样执行6个任务
- 核心线程执行任务1和2，其它的任务3～6放到队列中
- 执行完1和2，将3和4从队列中取出执行
- 执行完3和4，将5和6从队列中取出
创建线程池代码如下：
/**
* LinkedBlockingQueue
*/
private static void linkedQueue() {
System.out.println("\n\n =======LinkedBlockingQueue====== \n\n");
Executor executors = new ThreadPoolExecutor(
2, 3, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new RejectHandler());
execute(executors);
}
运行结果如下：
1 is running... 2 is running... //中间线程休眠 2 is end !!! //10s之后才运行完 1 is end !!! 3 is running... //任务3和4才执行 4 is running... 4 is end !!! 3 is end !!! 6 is running... 5 is running... 5 is end !!! 6 is end !!!
SynchronousQueue
说明
SynchrousQueue是个一个无缓存的队列。因为：SynchrousQueue源码可以看到：isEmpty()始终为true；size()始终返回0。
示例
说明
创建一个corePoolSize为2，maximumPoolSize为3的线程池。执行6个任务。
根据参数设置应该只可以执行3个任务：
- 2个核心线程执行2个任务；
- 第3个任务的时候，创建线程来执行任务3；
- 当第4个任务来的时候，此时已经超过了maximumPoolSize，所以拒绝任务。
代码
/**
* SynchronousQueue
*/
private static void syncQueue() {
System.out.println("\n\n =======SynchronousQueue====== \n\n");
Executor executors = new ThreadPoolExecutor(
2, 3, 30, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(),
new RejectHandler());
execute(executors);
}
执行结果
1 is running... 4 is rejected ^^ //4被拒 2 is running... 3 is running... 5 is rejected ^^ //5被拒 6 is rejected ^^ //6被拒 3 is end !!! 1 is end !!! 2 is end !!!
PriorityBlockingQueue
简介
PriorityBlockingQueue是一个无限容量的阻塞队列。
容量是无限的，所以put等入队操作其实不存在阻塞，只要内存足够都能够立即入队成功，当然多个入队操作的线程之间还是存在竞争唯一锁的互斥访问。虽然PriorityBlockingQueue逻辑上是无界的，但是尝试添加元素时还是可能因为资源耗尽而抛出OutOfMemoryError。
该队列也不允许放入null值，它使用与类java.util.PriorityQueue 相同的排序规则，也不允许放入不可比较的对象，这样做会导致ClassCastException。
值得注意的是，虽然PriorityBlockingQueue叫优先级队列，但是并不是说元素一入队就会按照排序规则被排好序，而是只有通过调用take、poll方法出队或者drainTo转移出的队列顺序才是被优先级队列排过序的。所以通过调用 iterator() 以及可拆分迭代器 spliterator() 方法返回的迭代器迭代的元素顺序都没有被排序。如果需要有序遍历可以通过 Arrays.sort(pq.toArray()) 方法来排序。注意peek方法永远只获取且不删除第一个元素，所以多次调用peek都是返回同样的值。
PriorityBlockingQueue其实是通过Comparator来排序的，要么入队的元素实现了Comparator接口（即所谓的自然排序），要么构造PriorityBlockingQueue实例的时候传入一个统一的Comparator实例，如果两者兼备那么以后者为准
PriorityBlockingQueue不保证具有相同优先级的元素顺序，但是你可以定义自定义类或比较器，通过辅助属性来决定优先级相同的元素的顺序，后文会举例说明。
DelayedWorkQueue
简介
为什么不直接使用DelayQueue而要重新实现一个DelayedWorkQueue呢，可能是了方便在实现过程中加入一些扩展。
使用场景
- 实现重试机制
- 比如当调用接口失败后，把当前调用信息放入delay=10s的元素，然后把元素放入队列，那么这个队列就是一个重试队列。一个线程通过take方法获取需要重试的接口，take返回则接口进行重试，失败则再次放入队列，同时也可以在元素加上重试次数。
- TimerQueue的内部实现
请先 ！