通俗易懂的JUC源码剖析-DelayQueue
前言
DelayQueue是BlockingQueue接口的实现类,它是带有延时功能的无界阻塞队列,意思就是每个元素都有过期时间,当从队列获取元素时,只有过期元素才会出队列。队列的头部元素是最快要过期的元素。
实现原理
先来看看它的类结构:
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
}
可以看到,存放的元素必须实现了Delayed接口,它的定义如下:
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}
再来看关键属性:
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;
private final Condition available = lock.newCondition();
再来看重要方法:
offer()
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
poll()
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return q.poll();
} 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)
return q.poll();
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
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();
}
}
明天再分析吧,先睡了,晚安全世界!
java
阅读 44发布于 今天 14:45
本作品系原创,采用《署名-非商业性使用-禁止演绎 4.0 国际》许可协议
小强大人
9 声望
1 粉丝
0 条评论
得票时间
小强大人
9 声望
1 粉丝
宣传栏
目录
前言
DelayQueue是BlockingQueue接口的实现类,它是带有延时功能的无界阻塞队列,意思就是每个元素都有过期时间,当从队列获取元素时,只有过期元素才会出队列。队列的头部元素是最快要过期的元素。
实现原理
先来看看它的类结构:
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
implements BlockingQueue<E> {
}
可以看到,存放的元素必须实现了Delayed接口,它的定义如下:
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}
再来看关键属性:
private final transient ReentrantLock lock = new ReentrantLock();
private final PriorityQueue<E> q = new PriorityQueue<E>();
private Thread leader = null;
private final Condition available = lock.newCondition();
再来看重要方法:
offer()
public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}
poll()
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
if (first == null || first.getDelay(NANOSECONDS) > 0)
return null;
else
return q.poll();
} 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)
return q.poll();
first = null; // don't retain ref while waiting
if (leader != null)
available.await();
else {
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();
}
}
明天再分析吧,先睡了,晚安全世界!