阻塞队列在并发编程非常常用,被广泛使用在“生产者-消费者”问题中。本文是阻塞队列下篇。
生产者线程每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> {}
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()就是核心方法了。
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);
}
}
PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序,插入元素是根据排序规则找到新元素在堆中位置插入。
简单看一下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;
}
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;
}
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];
}
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();// 释放锁
}
}
阻塞队列是一个比普通队列多出两个附加操作的队列。两个操作分别是:
PriorityQueue 优先级队列的元素按照其自然顺序进行排序或者构造队列时提供的 Comparator 进行排序,插入元素是根据排序规则找到新元素在堆中位置插入。