原创Java并发编程系列32 | 阻塞队列（下）
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了原创Java并发编程系列32 | 阻塞队列（下）相关的知识，希望对你有一定的参考价值。
【原创】Java并发编程系列32 | 阻塞队列（下）
收录于话题
#java 976 #程序员 2286 #并发编程 240 #进阶架构师 | 并发编程专题 12
★★★建议星标我们★★★
公众号改版后文章乱序推荐，希望你可以点击上方“Java进阶架构师”，点击右上角，将我们设为★“星标”！这样才不会错过每日进阶架构文章呀。
2020年Java原创面试题库连载中
【000期】Java最全面试题库思维导图
【001期】JavaSE面试题（一）：面向对象
【002期】JavaSE面试题（二）：基本数据类型与访问修饰符
【003期】JavaSE面试题（三）：JavaSE语法（1）
【004期】JavaSE面试题（四）：JavaSE语法（3）
【005期】JavaSE面试题（五）：String类
【006期】JavaSE面试题（六）：泛型
【007期】JavaSE面试题（七）：异常
【008期】JavaSE面试题（八）：集合之List
【009期】JavaSE面试题（九）：集合之Set
【010期】JavaSE面试题（十）：集合之Map
【011期】JavaSE面试题（十一）：多线程（1）
【012期】JavaSE面试题（十二）：多线程（2）
【013期】JavaSE面试题（十三）：多线程（3）
【014期】JavaSE面试题（十四）：基本IO流
【015期】JavaSE面试题（十五）：网络IO流
【016期】JavaSE面试题（十六）：反射
【017期】JavaSE面试题（十七）：JVM之内存模型
【018期】JavaSE面试题（十八）：JVM之垃圾回收
【020期】JavaSE系列面试题汇总（共18篇）
【019期】JavaWeb面试题（一）：JDBC
【021期】JavaWeb面试题（二）：HTTP协议
【022期】JavaWeb面试题（三）：Cookie和Session
【023期】JavaWeb面试题（四）：JSP
【024期】JavaWeb面试题（五）：Filter和Listener
【025期】Java工具面试题（一）：版本控制工具
【026期】Java工具面试题（二）：项目管理工具
【027期】Java设计模式面试题
【028期】JavaWeb系列面试题汇总（共10篇）
【029期】JavaEE面试题（一）Web应用服务器
【030期】JavaEE面试题（二）SpringMVC
【031期】JavaEE面试题（三）Spring（1）
【032期】JavaEE面试题（四）Spring（2）
【033期】JaveEE面试题（五）MyBatis
【034期】JavaEE面试题（六）Hibernate
【035期】JavaEE面试题（七）SpringBoot（1）
更多内容，点击上面蓝字查看
Java并发编程系列32 | 阻塞队列（下）
阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。本文是阻塞队列下篇。
4.3 SynchronousQueue
SynchronousQueue的同步指的是读线程和写线程需要同步，一个读线程匹配一个写线程。当一个线程往队列中写入一个元素时，写入操作不会立即返回，需要等待另一个线程来将这个元素拿走；当一个读线程做读操作的时候，同样需要一个相匹配的写线程的写操作。
SynchronousQueue 实际不存储元素，数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。
SynchronousQueue 执行put/take操作时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程），则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
使用
生产者线程每5秒put一个数据，消费者线程每1秒take一个数据。不管put和take时间如何调整，put和take总是成对出现，SynchronousQueue保证一个读线程匹配一个写线程。
public class SynchronousQueueDemo {
public static void main(String[] args) {
BlockingQueue<String> queue = new SynchronousQueue<String>();
new Thread("生产者") {
public void run() {
while (true) {
String data = UUID.randomUUID().toString();
try {
System.out.println("生产者 put: " + data);
queue.put(data);
Thread.sleep(5000);// 可修改时间测试
} catch (Exception e) {
e.printStackTrace();
}
}
};
}.start();
new Thread("消费者") {
public void run() {
while (true) {
try {
String data = queue.take();
System.out.println("消费者 take: " + data);
Thread.sleep(1000);// 可修改时间测试
} catch (Exception e) {
e.printStackTrace();
}
}
};
}.start();
}
}输出结果：
生产者 put: 890cf163-7c3e-4190-b45e-656cb5757cb5
消费者 take: 890cf163-7c3e-4190-b45e-656cb5757cb5
生产者 put: a858b31c-8bc1-4dce-b5b8-5f1cd318827f
消费者 take: a858b31c-8bc1-4dce-b5b8-5f1cd318827f
生产者 put: 75e6bdd0-a29a-4b70-9d0a-fd2e13fbe479
消费者 take: 75e6bdd0-a29a-4b70-9d0a-fd2e13fbe479
生产者 put: 8db3693e-fe24-4f4e-8f01-eef34542f3ec
消费者 take: 8db3693e-fe24-4f4e-8f01-eef34542f3ec
生产者 put: 233960ce-9ed0-40dd-b450-dd48055191c0
消费者 take: 233960ce-9ed0-40dd-b450-dd48055191c0类结构：
abstract static class Transferer {
// 用于转移元素
abstract Object transfer(Object e, boolean timed, long nanos);
}
// 公平模式
static final class TransferQueue<E> extends Transferer<E> {
// 等待队列节点
static final class QNode {
volatile QNode next;
volatile Object item;
volatile Thread waiter;
final boolean isData;
}
}
// 非公平模式
static final class TransferStack<E> extends Transferer<E> {}put()/take()
public void put(E o) throws InterruptedException {
if (o == null) throw new NullPointerException();
if (transferer.transfer(o, false, 0) == null) { // 1
Thread.interrupted();
throw new InterruptedException();
}
}
public E take() throws InterruptedException {
Object e = transferer.transfer(null, false, 0); // 2
if (e != null)
return (E)e;
Thread.interrupted();
throw new InterruptedException();
}可以看到，都是调用Transferer.transfer(E, boolean, long)方法，transfer()就是核心方法了。
transfer()用于转移元素，从生产者手上转到消费者手上，或者消费者调用这个方法来从生产者手上取元素。
第一个参数 e!=null，表示将元素从生产者转移给消费者；如果e==null，表示消费者等待生产者提供元素，然后返回生产者提供的元素。
当调用这个方法时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程，则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
Object transfer(Object e, boolean timed, long nanos) {
QNode s = null;
boolean isData = (e != null);
for (;;) {
QNode t = tail;
QNode h = head;
if (t == null || h == null)
continue;
if (h == t || t.isData == isData) {
/*
* 队列为空或队列中节点类型和当前节点一致，节点直接入队
*/
QNode tn = t.next;
if (t != tail)// 有其他节点入队
continue;
// 有其他节点入队，但是 tail 还是指向原来的，此时设置 tail 即可
if (tn != null) {
advanceTail(t, tn);// 如果 tail==t 的话，设置tail=tn
continue;
}
//
if (timed && nanos <= 0) // can‘t wait
return null;
if (s == null)
s = new QNode(e, isData);
// 将当前节点，插入到 tail 的后面
if (!t.casNext(null, s)) // failed to link in
continue;
// 将当前节点设置为新的 tail
advanceTail(t, s); // swing tail and wait
// 自旋阻塞，直到匹配到节点，返回节点
Object x = awaitFulfill(s, e, timed, nanos);
// 到这里，说明之前入队的线程被唤醒了
if (x == s) { // wait was cancelled
clean(t, s);
return null;
}
if (!s.isOffList()) { // not already unlinked
advanceHead(t, s); // unlink if head
if (x != null) // and forget fields
s.item = s;
s.waiter = null;
}
return (x != null) ? x : e;
} else { // complementary-mode
/*
* 如果队列中有等待节点，而且与当前操作可以匹配，则匹配等待队列的队头，出队，返回相应数据。
*/
QNode m = h.next; // node to fulfill
if (t != tail || m == null || h != head)
continue; // inconsistent read
Object x = m.item;
if (isData == (x != null) || // m already fulfilled
x == m || // m cancelled
!m.casItem(x, e)) { // lost CAS
advanceHead(h, m); // dequeue and retry
continue;
}
advanceHead(h, m); // successfully fulfilled
LockSupport.unpark(m.waiter);
return (x != null) ? x : e;
}
}
}
void advanceTail(QNode t, QNode nt) {
if (tail == t)
UNSAFE.compareAndSwapObject(this, tailOffset, t, nt);
}
/**
* 自旋阻塞，直到匹配到节点，返回节点
*/
Object awaitFulfill(QNode s, Object e, boolean timed, long nanos) {
long lastTime = timed ? System.nanoTime() : 0;
Thread w = Thread.currentThread();
// 判断需要自旋的次数，
int spins = ((head.next == s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
// 如果被中断了，那么取消这个节点
if (w.isInterrupted())
// 就是将当前节点 s 中的 item 属性设置为 this
s.tryCancel(e);
Object x = s.item;
// 这里是这个方法的唯一的出口
if (x != e)
return x;
// 如果需要，检测是否超时
if (timed) {
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) {
s.tryCancel(e);
continue;
}
}
if (spins > 0)
--spins;
// 如果自旋达到了最大的次数，那么检测
else if (s.waiter == null)
s.waiter = w;
// 如果自旋到了最大的次数，那么线程挂起，等待唤醒
else if (!timed)
LockSupport.park(this);
// spinForTimeoutThreshold 这个之前讲 AQS 的时候其实也说过，剩余时间小于这个阈值的时候，就
// 不要进行挂起了，自旋的性能会比较好
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}4.4 PriorityBlockingQueue
PriorityBlockingQueue队列为***队列，只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容。
PriorityBlockingQueue其实是 PriorityQueue 的线程安全版本，插入队列的对象必须是可比较大小的（comparable）。PriorityBlockingQueue/PriorityQueue 通过堆实现，这里不再详细介绍数据结构，重点讲解阻塞原理。
PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序，插入元素是根据排序规则找到新元素在堆中位置插入。
PriorityBlockingQueue put 方法不会 block，因为它是***队列；take 方法在队列为空的时候会阻塞。
简单看一下put和take方法的阻塞操作，很容易理解。put()：
public void put(E e) {
offer(e); // never need to block
}
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();// 获取锁
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();// 插入元素成功后，唤醒因队列为空而阻塞的读操作线程
} finally {
lock.unlock();// 释放锁
}
return true;
}take()：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();// 获取锁
E result;
try {
/*
* 队列空时，将当前线程加入notEmpty条件队列阻塞；
* 当有元素入队时，队列不为空了就可以take出元素，
* 此时会唤醒notEmpty条件队列中的线程，加入AQS阻塞队列等锁或者直接抢锁，然后执行出队操作。
*/
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();// 释放锁
}
return result;
}4.5 DelayQueue
DelayQueue是一个支持延时获取元素的***阻塞队列。
DelayQueue中的元素都是可延期的，因为必须实现Delayed接口。
插入元素时，会根据延期时间对元素排序，队头的元素是最先到期的；取出元素时，只有在队头元素到期时才能够从队列中取元素。如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。
使用
DelayQueue中元素都要实现Delayed接口，getDelay()方法获取延时时间，compareTo()方法比较延时时间用于排序。
将5s、10s、15s后执行的三个item加入DelayQueue队列，从打印结果来看，都是在预期的延时时间从DelayQueue中取出并执行的。
public class DelayQueueTest {
public static void main(String[] args) throws InterruptedException {
long curTime = System.currentTimeMillis();
Item item_5 = new Item("5S后执行的item", curTime + 5000);
Item item_10 = new Item("10S后执行的item", curTime + 10000);
Item item_15 = new Item("15S后执行的item", curTime + 15000);
DelayQueue<Item> queue = new DelayQueue<Item>();
queue.put(item_10);
queue.put(item_15);
queue.put(item_5);
System.out.println("开始!!! time=" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
for (int i = 0; i < 3; i++) {
Item take = queue.take();
System.out.println("执行 name=" + take.name + " time=" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
}
}
}
class Item implements Delayed {
String name;
private long time;
public Item(String name, long time) {
this.name = name;
this.time = time;
}
@Override
public long getDelay(TimeUnit unit) {
return time - System.currentTimeMillis();
}
@Override
public int compareTo(Delayed o) {
if (!(o instanceof Item)) {
return -1;
}
return (int)(this.time - ((Item)o).time);
};
}执行结果：
开始!!! time=2019-12-31T12:18:12.361
执行 name=5S后执行的item time=2019-12-31T12:18:17.306
执行 name=10S后执行的item time=2019-12-31T12:18:22.306
执行 name=15S后执行的item time=2019-12-31T12:18:27.306类结构
DelayQueue使用优先级队列PriorityQueue存储元素；使用ReentrantLock锁，保证队列数据并发环境下的安全性；通过lock的Condition实现阻塞。
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
/** 优先级队列，保存元素 */
private final PriorityQueue<E> q = new PriorityQueue<E>();
/** 锁，保证队列数据并发环境下的安全性 */
private final transient ReentrantLock lock = new ReentrantLock();
/** Condition */
private final Condition available = lock.newCondition();
/** 用于优化阻塞 */
private Thread leader = null;
}put()：
获取锁lock
添加元素
插入元素为队首元素时，唤醒take线程尝试take元素。因为更新了队首元素，所以要重新检查队首元素是否到期。
释放锁lock
public void put(E e) {
offer(e);
}
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();// 获取锁
try {
q.offer(e);// 添加元素
/*
* 插入元素为队首元素时，唤醒take线程尝试take元素
* 因为更新了队首元素，所以要重新检查队首元素是否到期
*/
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();// 释放锁
}
}
/**
* 获取队首元素
*/
public E peek() {
return (size == 0) ? null : (E) queue[0];
}take()：
获取锁
如果队列为空，阻塞take线程；插入元素后会take唤醒去获取队首元素。
如果队首元素到期，出队。
如果队首元素未到期，阻塞take线程t时间（t时间就是队首元素的到期剩余时间），时间到后唤醒take线程，尝试获取队首元素出队。
释放锁
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();// 获取锁
try {
for (;;) {// 注意是循环
E first = q.peek();// 获取队首元素
// 队列为空，take线程阻塞，等待被唤醒后再循环尝试take元素
if (first == null)
available.await();
// 队列不为空
else {
long delay = first.getDelay(NANOSECONDS);
// 队首元素执行时间到了，出队
if (delay <= 0)
return q.poll();// 出队
// 到这里，队列不为空，队首元素执行时间还没到，设置leader
first = null; // don‘t retain ref while waiting
// 在此之前，其他线程已经调用过take()设置过leader了
if (leader != null)
available.await();
else {
/*
* 设置当前take()线程为leader，并阻塞delay时间
* 阻塞唤醒之后，继续循环，尝试take元素
*/
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();// 释放锁
}
}5. 总结
阻塞队列是一个比普通队列多出两个附加操作的队列。两个操作分别是：
- 在队列为空时，获取元素的线程会等待队列变为非空。
- 当队列满时，存储元素的线程会等待队列可用。
ArrayBlockingQueue
ArrayBlockingQueue是由数组实现的有界队列，通过ReentrantLock锁保证队列数据的安全性，通过ReentrantLock的条件Condition是实现阻塞。
添加元素时，如果队列满了不能添加元素，就将添加元素的线程阻塞并加入notFull条件队列；当成功删除元素后，队列就可以添加元素了，唤醒notFull条件队列中阻塞的线程，添加元素。
删除元素时，如果队列空了不能删除元素，就将删除元素的线程阻塞并加入notEmpty条件队列；当成功添加元素后，队列就可以删除元素了，唤醒notEmpty条件队列中阻塞的线程，删除元素。
LinkedBlockingQueue
LinkedBlockingQueue用链表实现的有界阻塞队列。（不设置容量，默认为Integer.MAX_VALUE） - 锁takeLock保证删除数据的安全性，队列为空时读操作线程阻塞并加入takeLock锁的notEmpty条件等待队列。
- 锁putLock保证添加数据的安全性，队列满时写操作线程阻塞并加入putLock锁的notFull条件等待队列。
ArrayBlockingQueue的读写使用同一个锁来保证数据安全。LinkedBlockingQueue的读写分别用不同的锁来保证数据安全，采用不同的锁可以使读线程和写线程并发执行，提高了吞吐量，但也增加了编程的复杂度。
SynchronousQueue
SynchronousQueue的同步指的是读线程和写线程需要同步，一个读线程匹配一个写线程。当一个线程往队列中写入一个元素时，写入操作不会立即返回，需要等待另一个线程来将这个元素拿走；当一个读线程做读操作的时候，同样需要一个相匹配的写线程的写操作。
SynchronousQueue 实际不存储元素，数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。
SynchronousQueue 执行put/take操作时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程），则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
PriorityBlockingQueue
PriorityBlockingQueue队列为***队列，只能指定初始的队列大小，后面插入元素的时候，如果空间不够的话会自动扩容。
PriorityBlockingQueue其实是 PriorityQueue 的线程安全版本，插入队列的对象必须是可比较大小的（comparable）。PriorityBlockingQueue/PriorityQueue 通过堆实现，这里不再详细介绍数据结构，重点讲解阻塞原理。PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序，插入元素是根据排序规则找到新元素在堆中位置插入。
PriorityBlockingQueue put 方法不会 block，因为它是***队列；take 方法在队列为空的时候会阻塞。
DelayQueue
DelayQueue是一个支持延时获取元素的***阻塞队列。
DelayQueue中的元素都是可延期的，因为必须实现Delayed接口。
插入元素时，会根据延期时间对元素排序，队头的元素是最先到期的；取出元素时，只有在队头元素到期时才能够从队列中取元素。如果队头元素还有t时间到期，则将取出元素线程阻塞t时间，t时间到后再次尝试取出队头元素。
并发系列文章汇总
【原创】01|开篇获奖感言
【原创】02|并发编程三大核心问题
【原创】03|重排序-可见性和有序性问题根源
【原创】04|Java 内存模型详解
【原创】05|深入理解 volatile
【原创】06|你不知道的 final
【原创】07|synchronized 原理
【原创】08|synchronized 锁优化
【原创】09|基础干货
【原创】10|线程状态
【原创】11|线程调度
【原创】12|揭秘 CAS
【原创】13|LockSupport
【原创】14|AQS 源码分析
【原创】15|重入锁 ReentrantLock
【原创】16|公平锁与非公平锁
【原创】17|读写锁八讲（上）
【原创】18|读写锁八讲（下）
【原创】19|JDK8新增锁StampedLock
【原创】20|StampedLock源码解析
【原创】21|Condition-Lock的等待通知
【原创】22|倒计时器CountDownLatch
【原创】22|倒计时器CountDownLatch
【原创】23|循环屏障CyclicBarrier
【原创】24|信号量Semaphore
【原创】25|交换器Exchangere
【原创】26|ConcurrentHashMap（上）
【原创】27|ConcurrentHashMap（下）
【原创】28|Copy-On-Write容器
【原创】29|ConcurrentLinkedQueue
之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。
获取方式：点“在看”，V信关注上述Java最全面试题库号并回复 【面试】即可领取，更多精彩陆续奉上。
看到这里，证明有所收获
必须点个在看支持呀，喵
以上是关于原创Java并发编程系列32 | 阻塞队列（下）的主要内容，如果未能解决你的问题，请参考以下文章
Java Review - 并发编程_ConcurrentLinkedQueue原理&源码剖析