DelayQueue 學習和應用
阿新 • • 發佈：2019-02-16
DelayQueue的實現原理
DelayQueue的本質是一個實現了針對元素為Delayed的PriorityQueue， 它裡面比較出彩的地方就是使用了Leader/Follower 模式，來減少執行緒為是否過期輪詢，這樣提高了系統效率。
public interface Delayed extends Comparable<Delayed> { /** * Returns the remaining delay associated with this object, in the * given time unit. * * @param unit the time unit * @return the remaining delay; zero or negative values indicate * that the delay has already elapsed */ long getDelay(TimeUnit unit); }
主要程式碼剖析：
public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> { private final transient ReentrantLock lock = new ReentrantLock(); private final PriorityQueue<E> q = new PriorityQueue<E>(); /** * Thread designated to wait for the element at the head of * the queue. This variant of the Leader-Follower pattern * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to * minimize unnecessary timed waiting. When a thread becomes * the leader, it waits only for the next delay to elapse, but * other threads await indefinitely. The leader thread must * signal some other thread before returning from take() or * poll(...), unless some other thread becomes leader in the * interim. Whenever the head of the queue is replaced with * an element with an earlier expiration time, the leader * field is invalidated by being reset to null, and some * waiting thread, but not necessarily the current leader, is * signalled. So waiting threads must be prepared to acquire * and lose leadership while waiting. */ private Thread leader = null;
上面程式碼指出了內部實現，一個優先順序佇列，一個併發鎖，還有個LEADER/FOLLOWER的head；
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(); } }
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();
}
}最重要的是這裡：
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;
}
}為什麼不是直接
available.awaitNanos(delay);所以能理解這裡為什麼使用leader / follower 模式提高效率了。
DelayQueue的常用方法
public void put(E e)
public E take() throws InterruptedException
public boolean offer(E e)
public E poll()
public int drainTo(Collection<? super E> c)
DelayQueue的應用案例
這個場景中幾個點需要注意：
- 考試時間為120分鐘，30分鐘後才可交卷，初始化考生完成試卷時間最小應為30分鐘
- 對於能夠在120分鐘內交卷的考生，如何實現這些考生交卷
- 對於120分鐘內沒有完成考試的考生，在120分鐘考試時間到後需要讓他們強制交卷
package test.time;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class Examinee implements Delayed {
private int id;
private String name;
private long submitTime;
public Examinee(int i){
this.id = i;
this.name = "考生-" + i;
this.submitTime = Exam.EXAM_DEADLINE;
}
public int getId(){
return id;
}
/**
* 交卷了哦。
*/
public void submit(){
long current = System.currentTimeMillis();
if(current >= Exam.EXAM_MIN_TIME){
submitTime = current;
}else{
long cost = TimeUnit.SECONDS.convert(current - Exam.EXAM_START, TimeUnit.MILLISECONDS);
System.err.println("考試時間沒有超過30分鐘， ["+name+"] 不能交卷, 考試用時: "+cost);
}
}
@Override
public int compareTo(Delayed o) {
return (int)(getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
@Override
public long getDelay(TimeUnit unit) {
long delayMills = submitTime - System.currentTimeMillis();
return unit.convert(delayMills, TimeUnit.MILLISECONDS);
}
public String toString(){
long current = System.currentTimeMillis();
long cost = TimeUnit.SECONDS.convert(current - Exam.EXAM_START, TimeUnit.MILLISECONDS);
return name+" 在已經提交試卷, 考試耗時： "+cost;
}
}
package test.time;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
public class Exam {
// 考試開始時間
static long EXAM_START = System.currentTimeMillis();
// 考試時間 120 秒 （為了方便模擬）
static long EXAM_DEADLINE = EXAM_START + 120*1000L;
// 最早交卷時間 30 秒
static long EXAM_MIN_TIME = EXAM_START + 30*1000L;
// 考場總人數
private static final int EXAMINEE_NUM = 20;
private DelayQueue<Examinee> examinees = new DelayQueue<>();
private Map<Integer, Examinee> map = new ConcurrentHashMap<>(EXAMINEE_NUM*3/2);
private int submitCount;
public Exam(){
for(int i=1; i<=EXAMINEE_NUM; i++){
Examinee e = new Examinee(i);
examinees.offer(e);
map.put(i, e);
}
}
public void examining() throws InterruptedException{
DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date start = new Date(EXAM_START);
Date end = new Date(EXAM_DEADLINE);
System.out.println("====> 考試開始 ["+format.format(start)+"] ...");
// 25 分鐘以後
TimeUnit.SECONDS.sleep(25);
while(System.currentTimeMillis() <= EXAM_DEADLINE ){
someoneSubmit();
TimeUnit.SECONDS.sleep(2);
System.out.println("--------------- time past 2 seconds -------");
List<Examinee> submitted = new LinkedList<>();
examinees.drainTo(submitted);
for(Examinee e: submitted){
System.out.println(e);
}
submitCount += submitted.size();
if(submitCount >= EXAMINEE_NUM){
break;
}
}
System.out.println("====> 考試結束 ["+format.format(end)+"] ...");
}
void someoneSubmit(){
int num = new Random().nextInt(EXAMINEE_NUM)+1;
map.get(num).submit();
}
public static void main(String[] args){
try {
new Exam().examining();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}