Java并发编程系列32 | 阻塞队列（下）
阻塞队列在并发编程非常常用，被广泛使用在“生产者-消费者”问题中。本文是阻塞队列下篇。
4.3 SynchronousQueue
- SynchronousQueue的同步指的是读线程和写线程需要同步，一个读线程匹配一个写线程。当一个线程往队列中写入一个元素时，写入操作不会立即返回，需要等待另一个线程来将这个元素拿走；当一个读线程做读操作的时候，同样需要一个相匹配的写线程的写操作。
- SynchronousQueue 实际不存储元素，数据必须从某个写线程交给某个读线程，而不是写到某个队列中等待被消费。
- SynchronousQueue 执行put/take操作时，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是写线程），则将当前线程加入到等待队列。如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是读操作线程，当前线程是写操作线程，反之亦然），则匹配等待队列的队头，出队，返回相应数据。
使用
生产者线程每5秒put一个数据，消费者线程每1秒take一个数据。不管put和take时间如何调整，put和take总是成对出现，SynchronousQueue保证一个读线程匹配一个写线程。
代码语言：javascript复制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();
}
}输出结果：
代码语言：javascript复制生产者 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类结构：
代码语言：javascript复制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()
代码语言：javascript复制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()：
代码语言：javascript复制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()：
代码语言：javascript复制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);
};
}执行结果：
代码语言：javascript复制开始!!! 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实现阻塞。
代码语言：javascript复制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时间到后再次尝试取出队头元素。
