【Java】通俗易懂的JUC源码剖析-DelayQueue

通俗易懂的JUC源码剖析-DelayQueue

小强大人发布于 今天 14:45

前言

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 国际》许可协议

avatar

小强大人

9 声望

1 粉丝

0 条评论

得票时间

avatar

小强大人

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();

}

}

明天再分析吧,先睡了,晚安全世界!

以上是 【Java】通俗易懂的JUC源码剖析-DelayQueue 的全部内容, 来源链接: www.h5w3.com/114010.html

回到顶部