# Queue接口
Queue 队列，也是Collection的一个重要分支，是一个先进先出的结构，其一个重要的子接口为BlockingQueue（阻塞队列），阻塞队列在多线程的场景中有着广泛的应用，在这里主要学习BlockingQueue的实现类。
阻塞队列中的阻塞意为等待，与非阻塞队列相比的话有如下的不同点：
1. 出队：
* 非阻塞队列：当队列满的时候，放入新的元素时，数据丢失。
* 阻塞队列：当队列满的时候，放入新的元素时，线程阻塞等待，等待队列中有出队的元素，再继续运行放进去。
2. 出队：
* 非阻塞队列：当队列没有元素的时候，取数据时得到的是null。
* 阻塞队列：当队列没有元素的时候，取数据时，线程阻塞等待，什么时候有元素入队了，才可以继续运行取出元素。
常用API：
1. 添加元素
~~~
public boolean add(E, e); // 不能添加null，成功时返回true，不成功时抛出Queue Full异常
public boolean offer(E, e); // 不能添加null，成功时返回true，不成功时返回false
public void put(E e); // 阻塞操作
~~~
2. 查询
~~~
public E take(); // 获取并移除此队列的头部，在元素变得可用之前一直等待
public E poll(long time, TimeUnit unit); // 获取并移除此队列的头部，在指定时间内等待可用的元素
public E peek(); // 获取队列的头部，不移除
~~~
*****
BlockingQueue的常见实现类：
## ArrayBlockingQueue
ArrayBlockingQueue底层是一个基于数组的有边界的阻塞队列，其只用了一把锁来同时阻塞读写操作，读写不分离。
源码分析：
~~~
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 底层使用数组保存数据
final Object[] items;
// 取元素时用到的索引，初始为0
int takeIndex;
// 放元素时用到的索引，初始为0
int putIndex;
// 数组中元素的个数
int count;
// 可重入锁，用于出队和入队等操作中
final ReentrantLock lock;
// lock伴随的一个不为空的等待池，当队列有数据时会唤醒在该等待队列中的线程获取数据
private final Condition notEmpty;
// lock伴随的一个队列不满的等待池，当队列中还没有满时会唤醒在该等待队列中的线程往队列添加数据
private final Condition notFull;
// -----------------------构造方法-------------------------
public ArrayBlockingQueue(int capacity) { // 必须指定初始化容量
this(capacity, false); // 默认为非公平锁
}
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();
}
//------------------两个关键的操作-------------------------
// 入队
private void enqueue(E x) {
final Object[] items = this.items;
// 将x放入putIndex索引指向的位置
items[putIndex] = x;
if (++putIndex == items.length)
// 循环数组的作用，可以不断的利用底层数组存放数据
putIndex = 0;
// 元素个数+1
count++;
// 添加一个元素则证明底层数组有数据了，唤醒在空队列的等待池中阻塞的线程取出数据，即唤醒take方法的阻塞
notEmpty.signal();
}
// 入队
private E dequeue() {
final Object[] items = this.items;
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
// 循环数组
takeIndex = 0;
count--; // 元素个数-1
if (itrs != null)
itrs.elementDequeued();
// 取出一个元素则证明队列中又有空间可以存放数据了，唤醒在满队列的等待池中阻塞的线程，可以往队列中继续添加数据，即唤醒put方法的阻塞
notFull.signal();
return x;
}
// -------------------阻塞操作-------------------------
// 添加元素
public void put(E e) throws InterruptedException {
checkNotNull(e); // 检查是否为空
// 线程安全的添加元素
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
// 元素满了的话将当前线程放入已满等待池中阻塞
notFull.await();
// 入队操作-可能会唤醒在非空等待池中阻塞的队列
enqueue(e);
} finally {
lock.unlock();
}
}
// 取出元素
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
// 如果没有元素的话，则会将当前线程放入非空等待池中阻塞
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
}
~~~
【更正：Condition中等待池这个词汇换成等待队列可能会更合适一点。】
总结：
1. 两个Conditional变量的作用，notEmpty用来阻塞获取元素操作，当队列非空时（enqueue中唤醒）会唤醒该等待池中的线程；notFull用来阻塞添加元素操作，当队列未满时（dequeue中唤醒）会唤醒该等待池中的线程。
2. 读写操作只用到一把锁，读写不分离。
3. 必须指定初始化容量。对于put操作当队列已满的时候会阻塞线程，对于take操作当队列为空时也会阻塞线程。
> 面试：能否将while(count == items.length)和while(count == 0)中的while换成if？
不能，因为notFull中的线程被唤醒的瞬间，有其他线程放入元素，此时队列又满了，如果采用if的话只会判断一次，线程被唤醒之后就会继续执行enqueue操作，但是此时队列是满的，执行enqueue就会造成数据的丢失了。因此需要不断的判断，直到唤醒的线程判断出当前队列真的有位置了才能继续运行。
## LinkedBlockingQueue
底层基于链表结构，支持读写同时操作，并发情况下，效率比ArrayBlockingQueue高。是一个`可选择`的有界队列，可以指定链表的容量，当不指定时默认为整型的最大值。
其API的使用与ArrayBlockingQueue一样。
源码分析：
~~~
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
// 链表的节点结构
static class Node<E> {
E item; // 数据域
Node<E> next; // 下个节点的指针域
Node(E x) { item = x; }
}
// 链表的最大容量，不指定时默认为Integer.MAX_VALUE
private final int capacity;
// 链表中元素的个数，因为读写操作是分离的，所以要使用原子操作来改变元素的个数
private final AtomicInteger count = new AtomicInteger();
// 头结点
transient Node<E> head;
// 尾节点
private transient Node<E> last;
// 获取数据的锁
private final ReentrantLock takeLock = new ReentrantLock();
// 队列非空等待池
private final Condition notEmpty = takeLock.newCondition();
// 添加数组的锁
private final ReentrantLock putLock = new ReentrantLock();
// 队列不满时的等待池
private final Condition notFull = putLock.newCondition();
//---------------------构造方法----------------------
public LinkedBlockingQueue() {
// 默认容量为整型最大值
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
//--------------------关键操作----------------------------
// 入队，
private void enqueue(Node<E> node) {
// 尾插法，直接添加到最后一个元素，并改变last的指向
last = last.next = node;
}
// 出队
private E dequeue() {
Node<E> h = head;
// 第一个节点，会成为新的头节点，
Node<E> first = h.next;
h.next = h; // help GC，这样操作没有别的引用h这个节点，垃圾回收会进行回收
head = first;
// 返回第一个节点的数据
E x = first.item;
// head的数据域为空
first.item = null;
return x;
}
// ----------------put和take操作-------------------------
// 添加元素
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
// 封装成新的节点结构
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
// 队列已满，则阻塞当期线程，采用while的原因与ArrayBlockingQueue相同
notFull.await();
}
enqueue(node);
// 递增元素个数
c = count.getAndIncrement();
if (c + 1 < capacity)
// 再次判断队列是否满的，因为count可能同时被读线程操作
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
// 队列中有元素了，唤醒take阻塞的线程，这里的c是旧值，count=1
signalNotEmpty();
}
// 取出元素
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
// 原子的递减元素个数
c = count.getAndDecrement();
if (c > 1)
// 再次判断是否有元素
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
// c是count的旧值，c==capacity就证明队列有空余出来的位置了，唤醒put阻塞的线程
signalNotFull();
return x;
}
}
~~~
> 在FixedThreadPool和SingleThreadExecutor中被使用。
> 疑问，为什么使用了锁还要使用while进行排队队列是否为空或者队列是否满了？
> 这是因为底层使用的是ReetrantLock，支持可重复入的。
**ArrayBlockingQueue和LinkedBlockingQueue的对比**
假如有10000个线程，分别有5000个线程进行读操作，有5000个线程进行写操作，其性能对比如下：
~~~
public static void main(String[] args) throws InterruptedException {
ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10000);
// LinkedBlockingQueue<String> linkedBlockingQueue = new LinkedBlockingQueue<>(10000);
List<String> dataList = new Vector<>();
// 创建10000个线程，分别进行500次的读写操作
long start = System.currentTimeMillis();
for (int i = 0; i < 10000; i++) {
if (i % 2 == 0) {
new Thread(() -> {
for (int j = 0; j < 500; j++) {
try {
blockingQueue.put(Thread.currentThread().getName() + "j");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "Thread-" + i).start();
} else {
new Thread(() -> {
for (int j = 0; j < 500; j++) {
try {
dataList.add(blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "Thread-" + i).start();
}
}
long end = System.currentTimeMillis();
System.out.println("用时：" + (end - start) + "ms");
System.out.println(dataList.size());
}
~~~
结果：
~~~
ArrayBlockingQueue用时：2331ms
LinkedBlockingQueue用时：2319ms
~~~
两者性能差不多，有时LinkedBlockingQueue的性能还比不上ArrayBlockingQueue。
## SynchronousQueue
使用这个队列必须先从队列中取出元素（即先调用take方法），才可以向队列中加入元素（再调用put方法），SynchronousQueue队列中没有任何容量，甚至一个容量都没有。可以理解一个标记，当一个线程从队列获取数据的时候就会打上一个标记，之后如果有另外一个线程放入数据，数据就会直接传送给获取数据的线程。
如果先向队列中添加元素则会抛出异常`Queue Full`，因为队列是没有容量的。使用put方法的话阻塞，因为一开始队列就是满的。注意取出元素的方法不能用peek，因为peek不会将元素从队列中拿走，只是查看的作用。
优点：方便高效的进行线程间的数据传送，效率高，不会产生队列中数据争抢问题。
~~~
public static void main(String[] args) throws InterruptedException {
SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + ":" + synchronousQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "Thread-take").start();
new Thread(() -> {
try {
synchronousQueue.put("abc");
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "Thread-put").start();
}
~~~
要多个线程配合使用才行。
> 在CachedThreadPool中被使用
## PriorityBlockingQueue
带有优先级的阻塞队列，队列中的元素有不同的优先级；没有界限限制（整型最大值），但是可以指定初始化的长度，不指定时默认为11，底层会进行扩容操作。放入的元素必须实现内部比较器，或者在创建PriorityBlockingQueue时设置外部比较器，元素的优先级根据比较器来决定。
~~~
public static void main(String[] args) throws InterruptedException {
PriorityBlockingQueue<String> blockingQueue = new PriorityBlockingQueue<>(11, new Comparator<String>() {
@Override
public int compare(String o1, String o2) {
return o1.compareTo(o2);
}
});
blockingQueue.put("bbb");
blockingQueue.put("aaa");
blockingQueue.put("ccc");
System.out.println(blockingQueue);
System.out.println("------------------------------------------------------");
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
}
~~~
## DelayQueue
DelayQueue是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，存放在DelayQueue队列中的对象只能在其到期时才能从队列中取走。
当生产者线程调用put方法添加元素时，会触发Delayed接口中的compareTo方法进行排序，也就是队列中的元素的顺序是按到期时间排序的，而非进入队列的顺序。排在队列头部的元素最早到期，越往后到期时间越晚。
消费者线程查看队列头部的元素，并不是取出操作。然后调用元素的getDelay方法，如果此方法的返回值≤0，则消费者线程会从队列中取出此元素处理。如果getDealy方法返回值大于0，则消费者线程wait返回的时间值后，再次从队列头部取出元素，此时元素到期可以取出。
注意：不能将null元素放置到这种队列中。
使用场景：
1. 淘宝订单业务：下单之后如果30分钟之内没有付款就自动取消订单。
2. 饿了么订餐通知：下单成功后60S之内给用户发送短信通知。
3. 关闭空闲连接：服务器中客户端的连接空闲一段时间后会自动关闭。
4. 缓存：缓存中的对象超过空闲时间限制，需要从缓存中移出。
5. 任务超时处理：在tcp协议的滑动窗口中，用来处理超时未响应的请求。
~~~
public class TestDelayQueue {
public static void main(String[] args) throws InterruptedException {
DelayQueue<Student> delayQueue = new DelayQueue<>();
delayQueue.put(new Student("zhangsan", 20, System.currentTimeMillis() + 5000));
delayQueue.put(new Student("lisi", 19, System.currentTimeMillis() + 2000));
delayQueue.put(new Student("wangwu", 18, System.currentTimeMillis() + 1000));
System.out.println(delayQueue);
System.out.println(delayQueue.take());
System.out.println(delayQueue.take());
System.out.println(delayQueue.take());
}
}
// Student.java
public class Student implements Delayed {
private String name;
private int age;
private long endTime;
public Student() {}
public Student(String name, int age, long endTime) {
this.name = name;
this.age = age;
this.endTime = endTime;
}
@Override
public long getDelay(TimeUnit unit) {
return endTime - System.currentTimeMillis();
}
@Override
public int compareTo(Delayed o) {
Student other = (Student) o;
return age - other.getAge();
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public String toString() {
return "Student{" +
"name='" + name + '\'' +
", age=" + age +
", endTime=" + endTime +
'}';
}
}
~~~
实现Delayed接口的实现类需要重写getDelay和compareTo方法。
- 第一章 Java基础
- ThreadLocal
- Java异常体系
- Java集合框架
- List接口及其实现类
- Queue接口及其实现类
- Set接口及其实现类
- Map接口及其实现类
- JDK1.8新特性
- Lambda表达式
- 常用函数式接口
- stream流
- 面试
- 第二章 Java虚拟机
- 第一节、运行时数据区
- 第二节、垃圾回收
- 第三节、类加载机制
- 第四节、类文件与字节码指令
- 第五节、语法糖
- 第六节、运行期优化
- 面试常见问题
- 第三章 并发编程
- 第一节、Java中的线程
- 第二节、Java中的锁
- 第三节、线程池
- 第四节、并发工具类
- AQS
- 第四章 网络编程
- WebSocket协议
- Netty
- Netty入门
- Netty-自定义协议
- 面试题
- IO
- 网络IO模型
- 第五章 操作系统
- IO
- 文件系统的相关概念
- Java几种文件读写方式性能对比
- Socket
- 内存管理
- 进程、线程、协程
- IO模型的演化过程
- 第六章 计算机网络
- 第七章 消息队列
- RabbitMQ
- 第八章 开发框架
- Spring
- Spring事务
- Spring MVC
- Spring Boot
- Mybatis
- Mybatis-Plus
- Shiro
- 第九章 数据库
- Mysql
- Mysql中的索引
- Mysql中的锁
- 面试常见问题
- Mysql中的日志
- InnoDB存储引擎
- 事务
- Redis
- redis的数据类型
- redis数据结构
- Redis主从复制
- 哨兵模式
- 面试题
- Spring Boot整合Lettuce+Redisson实现布隆过滤器
- 集群
- Redis网络IO模型
- 第十章 设计模式
- 设计模式-七大原则
- 设计模式-单例模式
- 设计模式-备忘录模式
- 设计模式-原型模式
- 设计模式-责任链模式
- 设计模式-过滤模式
- 设计模式-观察者模式
- 设计模式-工厂方法模式
- 设计模式-抽象工厂模式
- 设计模式-代理模式
- 第十一章 后端开发常用工具、库
- Docker
- Docker安装Mysql
- 第十二章 中间件
- ZooKeeper