常见的阻塞队列
Queue接口
public interface Queue<E> extends Collection<E> {
//添加一个元素，添加成功返回true, 如果队列满了，就会抛出异常
boolean add(E e);
//添加一个元素，添加成功返回true, 如果队列满了，返回false
boolean offer(E e);
//返回并删除队首元素，队列为空则抛出异常
E remove();
//返回并删除队首元素，队列为空则返回null
E poll();
//返回队首元素，但不移除，队列为空则抛出异常
E element();
//获取队首元素，但不移除，队列为空则返回null
E peek();
}
BlockingQueue 接口
public interface BlockingQueue<E> extends Queue<E> {
// 入队 add offer put
//添加一个元素，添加成功返回true, 如果队列满了，就会抛出异常
boolean add(E e);
//如果队列没满，返回true，如果队列已满，返回false（不阻塞）
boolean offer(E e);
//队列没满的时候是正常的插入，如果队列已满，则阻塞，直至队列空出位置
void put(E e) throws InterruptedException;
//可以设置阻塞时间，如果队列已满，则进行阻塞。超过阻塞时间，则返回false
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
//出队 take poll remove
//可以设置阻塞时间，如果没有数据，则阻塞，超过阻塞时间，则返回null
E take() throws InterruptedException;
//如果有数据，出队，如果没有数据，返回null （不阻塞）
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
//返回并删除队首元素，队列为空则抛出异常
boolean remove(Object o);
}
ArrayBlockingQueue 队列
-
ArrayBlockingQueue是最典型的有界阻塞队列，其内部是用数组存储元素的，初始化时需要指定容量大小，利用ReentrantLock 实现线程安全。
-
在生产者-消费者模型中使用时，如果生产速度和消费速度基本匹配的情况下，使用ArrayBlockingQueue是个不错选择；当如果生产速度远远大于消费速度，则会导致队列填满，大量生产线程被阻塞。
-
使用独占锁ReentrantLock实现线程安全，入队和出队操作使用同一个锁对象，也就是只能有一个线程可以进行入队或者出队操作；这也就意味着生产者和消费者无法并行操作，在高并发场景下会成为性能瓶颈。
重要方法
这里主要讲解阻塞方法
//队列没满的时候是正常的插入，如果队列已满，则阻塞，直至队列空出位置
void put(E e) throws InterruptedException;
//可以设置阻塞时间，如果队列已满，则进行阻塞。超过阻塞时间，则返回false
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
//出队 take poll remove
//可以设置阻塞时间，如果没有数据，则阻塞，超过阻塞时间，则返回null
E take() throws InterruptedException;
//如果有数据，出队，如果没有数据，返回null （不阻塞）
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
基础属性
//数据元素数组
final Object[] items;
//下一个待取出元素索引
int takeIndex;
//下一个待添加元素索引
int putIndex;
//数组大小
int count;
//内部锁
final ReentrantLock lock;
//消费者 等待
private final Condition notEmpty;
//生产者 等待
private final Condition notFull;
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity]; //创建 capacity 大小的数组
lock = new ReentrantLock(fair);//创建内部锁 fair = false 非公平锁 fair = true 公平锁
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
put 方法
public void put(E e) throws InterruptedException {
checkNotNull(e);//校验元素e 是否为空
final ReentrantLock lock = this.lock; //获取内部锁，创建队列时，已经初始化过内部锁
lock.lockInterruptibly();//加锁，如果线程中断抛出异常
try {
//阻塞队列已满，则将生产者挂起，等待消费者唤醒
//设计注意点： 用while不用if是为了防止虚假唤醒
while (count == items.length)
notFull.await();//生产者线程等待
// 入队
enqueue(e);
} finally {
//释放锁
lock.unlock();
}
}
enqueue 方法
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;//获取数组
//入队 使用的putIndex 入队
items[putIndex] = x;
if (++putIndex == items.length) //如果putIndex 等于数组大小
putIndex = 0; //putIndex 指向下标为0 的位置 这里会构建一个环形数组
count++; 数组大小 +1
notEmpty.signal(); //notEmpty条件队列转同步队列，准备唤醒消费者线程，因为入队了一个元素，肯定不为空了
}
这里入队满了以后为什么要设置成环形数组？
从图片中可以看出环形数组可以让队列出队的复杂度从O(n) 变为 O(1)
task 方法
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock; //获得内部锁
lock.lockInterruptibly();//加锁，如果线程中断抛出异常
try {
while (count == 0)//如果队列大小为0 则阻塞消费者线程
notEmpty.await();
return dequeue();//出队
} finally {
lock.unlock();//解锁 唤醒生产者线程
}
}
dequeue 方法
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items; //获得队列数组
@SuppressWarnings("unchecked")
E x = (E) items[takeIndex];//通过takeIndex 下标 取出任务 这里实现了数组的先进先出
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;//环形数组，takeIndex 指针到数组尽头了，返回头部
count--;//数组大小-1
if (itrs != null)
itrs.elementDequeued();
notFull.signal();//notFull条件队列转同步队列，准备唤醒生产者线程，此时队列有空位
return x;
}
LinkedBlockingQueue 队列
- LinkedBlockingQueue是一个基于链表实现的阻塞队列，默认情况下，该阻塞队列的大小Integer.MAX_VALUE，由于这个数值特别大，所以LinkedBlockingQueue也被称作无界队列，代表它几乎没有界限，队列可以随着元素的添加而动态增长，但是如果没有剩余内存，则队列将抛出OOM错误。所以为了避免队列过大造成机器负载或者内存爆满的情况出现，我们在使用的时候建议手动传一个队列的大小。
- LinkedBlockingQueue内部由单链表实现，只能从head取元素，从tail添加元素。LinkedBlockingQueue采用两把锁的锁分离技术实现入队出队互不阻塞，添加元素和获取元素都有独立的锁，也就是说LinkedBlockingQueue是读写分离的，读写操作可以并行执行。
基础属性
// 容量,指定容量就是有界队列
private final int capacity;
// 元素数量
private final AtomicInteger count = new AtomicInteger();
// 链表头 本身是不存储任何元素的，初始化时item指向null
transient Node<E> head;
// 链表尾
private transient Node<E> last;
// take锁 锁分离，提高效率 读锁
private final ReentrantLock takeLock = new ReentrantLock();
// notEmpty条件
// 当队列无元素时，take锁会阻塞在notEmpty条件上，等待其它线程唤醒
private final Condition notEmpty = takeLock.newCondition();
// put锁 写锁
private final ReentrantLock putLock = new ReentrantLock();
// notFull条件
// 当队列满了时，put锁会会阻塞在notFull上，等待其它线程唤醒
private final Condition notFull = putLock.newCondition();
//典型的单链表结构
static class Node<E> {
E item; //存储元素
Node<E> next; //后继节点 单链表结构
Node(E x) { item = x; }
}
public LinkedBlockingQueue() {
// 如果没传容量，就使用最大int值初始化其容量
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
// 初始化head和last指针为空值节点
last = head = new Node<E>(null);
}
put 方法
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();//入队元素为空 抛异常
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e); //创建node 节点
final ReentrantLock putLock = this.putLock; //获取内部锁 也就是写锁
final AtomicInteger count = this.count; //链表大小，这里AtomicInteger 保证原子性
putLock.lockInterruptibly();
try {
// 如果队列满了，就阻塞在notFull上等待被其它线程唤醒（阻塞生产者线程）
while (count.get() == capacity) {
notFull.await();
}
enqueue(node);//入队
c = count.getAndIncrement();//链表大小+1
// 如果现队列长度小于容量，notFull条件队列转同步队列，准备唤醒一个阻塞在notFull条件上的线程(可以继续入队)
// 因为可能有很多线程阻塞在notFull这个条件上,而取元素时只有取之前队列是满的才会唤醒notFull,此处不用等到取元素时才唤醒
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();//真正唤醒生产者线程
}
// 如果原队列长度为0，现在加了一个元素后立即唤醒阻塞在notEmpty上的线程
if (c == 0)
signalNotEmpty();
}
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node;//添加到链表的尾端 last 的下一个节点
}
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();// 获得 take锁
try {
notEmpty.signal();// notEmpty条件队列转同步队列，准备唤醒阻塞在notEmpty上的线程
} finally {
takeLock.unlock(); // 真正唤醒消费者线程
}
take 方法
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;//链表大小
final ReentrantLock takeLock = this.takeLock;//获得 读锁
takeLock.lockInterruptibly();
try {
//如果队列无元素 则消费者线程 阻塞在notEmpty 上
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();//出队
c = count.getAndDecrement(); //队列元素减1 返回原值
if (c > 1) // 如果取之前队列长度大于1，notEmpty条件队列转同步队列，准备唤醒阻塞在notEmpty上的线程，原因与入队同理
notEmpty.signal();
} finally {
takeLock.unlock();// 真正唤醒消费者线程
}
// 为什么队列是满的还唤醒阻塞在notFull上的线程呢？
// 因为唤醒是需要加putLock的，这是为了减少锁的次数,所以，这里索性在放完元素就检测一下，未满就唤醒其它notFull上的线程,
// 这也是锁分离带来的代价
// 如果取之前队列长度等于容量（已满），则唤醒阻塞在notFull的线程
if (c == capacity)
signalNotFull();
return x;
}
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head; //获取头节点 head节点本身是不存储任何元素的
Node<E> first = h.next;//获取头节点的下一个节点，这里的节点存储元素
h.next = h; // help GC //断开原head jvm 回收
head = first; //将原头节点的下一个节点置为头节点
E x = first.item; //取出节点元素
first.item = null; 节点置空
return x;
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
//获取写锁 只有获得写锁才能 唤醒notFull 等待队列里面的线程
putLock.lock();
try {
notFull.signal(); //将notFull 等待队列同步至同步队列中
} finally {
putLock.unlock();//真正唤醒写线程
}
}
DelayQueue 队列
DelayQueue 是一个支持延时获取元素的阻塞队列， 内部采用优先队列 PriorityQueue 存储元素，同时元素必须实现Delaye 接口；在创建元素时可以指定多久才可以从队列中获取当前元素，只有在延迟期满时才能从队列中提取元素。延迟队列的特点是：不是先进先出，而是会按照延迟时间的长短来排序，下一个即将执行的任务会排到队列的最前面。
DelayQueue 是无界队列，放入的元素必须实现 Delayed 接口，而 Delayed 接口又继承了 Comparable 接口，所以自然就拥有了比较和排序的能力。
public interface Delayed extends Comparable<Delayed> {
//getDelay 方法返回的是“还剩下多长的延迟时间才会被执行”，
//如果返回 0 或者负数则代表任务已过期。
//元素会根据延迟时间的长短被放到队列的不同位置，越靠近队列头代表越早过期。
//使用时需要实现该方法 与当前时间做比较
long getDelay(TimeUnit unit);
}
// ScheduledThreadPoolExecutor 该线程池 就实现该方法起到延时的作用
public long getDelay(TimeUnit unit) {
return unit.convert(time - now(), NANOSECONDS);
}
基础属性
//用于保证队列操作的线程安全
private final transient ReentrantLock lock = new ReentrantLock();
// 优先级队列,存储元素，用于保证延迟低的优先执行
private final PriorityQueue<E> q = new PriorityQueue<E>();
// 用于标记当前是否有线程在排队（仅用于取元素时） leader 指向的是第一个从队列获取元素阻塞的线程
private Thread leader = null;
// 条件，用于表示现在是否有可取的元素 当新元素到达，或新线程可能需要成为leader时被通知
private final Condition available = lock.newCondition();
public DelayQueue() {}
public DelayQueue(Collection<? extends E> c) {
this.addAll(c);
}
put 方法
public void put(E e) {
offer(e);
}
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 入队
q.offer(e);
if (q.peek() == e) {
// 若入队的元素位于队列头部，说明当前元素延迟最小
// 将 leader 置空
leader = null;
// available条件队列转同步队列,准备唤醒阻塞在available上的线程
available.signal();
}
return true;
} finally {
lock.unlock(); // 解锁，真正唤醒阻塞的线程
}
}
take 方法
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();// 取出堆顶元素
if (first == null)// 如果堆顶元素为空，说明队列中还没有元素，直接阻塞等待
available.await();
else {
long delay = first.getDelay(NANOSECONDS);// 堆顶元素的到期时间
if (delay <= 0)// 如果小于0说明已到期，直接调用poll()方法弹出堆顶元素
return q.poll();
// 如果delay大于0 ，则下面要阻塞了
// 将first置为空方便gc
first = null;
// 如果前面有其它线程在等待，直接进入等待
if (leader != null)
available.await();
else {
// 如果leader为null，把当前线程赋值给它
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
// 等待delay时间后自动醒过来
// 醒过来后把leader置空并重新进入循环判断堆顶元素是否到期
// 这里即使醒过来后也不一定能获取到元素
// 因为有可能其它线程先一步获取了锁并弹出了堆顶元素
// 条件锁的唤醒分成两步，先从Condition的队列里出队
// 再入队到AQS的队列中，当其它线程调用LockSupport.unpark(t)的时候才会真正唤醒
available.awaitNanos(delay);
} finally {
// 如果leader还是当前线程就把它置为空，让其它线程有机会获取元素
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 成功出队后，如果leader为空且堆顶还有元素，就唤醒下一个等待的线程
if (leader == null && q.peek() != null)
// available条件队列转同步队列,准备唤醒阻塞在available上的线程
available.signal();
// 解锁，真正唤醒阻塞的线程
lock.unlock();
}
