java併發之DelayQueue實際運用示例
阿新 • • 發佈：2018-12-10
在學習Java 多執行緒併發開發過程中，瞭解到DelayQueue類的主要作用：是一個無界的BlockingQueue，用於放置實現了Delayed介面的物件，其中的物件只能在其到期時才能從佇列中取走。這種佇列是有序的，即隊頭物件的延遲到期時間最長。注意：不能將null元素放置到這種佇列中。
Delayed，一種混合風格的介面，用來標記那些應該在給定延遲時間之後執行的物件。此介面的實現必須定義一個 compareTo 方法，該方法提供與此介面的 getDelay 方法一致的排序。
在網上也看到兩個示例，但這兩個示例個人在實際執行時均沒有達到滿足業務場景的效果，因而對其進行了修改，供大家參考討論。
業務場景一：多考生考試
這個場景中幾個點需要注意：
- 考試時間為120分鐘，30分鐘後才可交卷，初始化考生完成試卷時間最小應為30分鐘
- 對於能夠在120分鐘內交卷的考生，如何實現這些考生交卷
- 對於120分鐘內沒有完成考試的考生，在120分鐘考試時間到後需要讓他們強制交卷
- 在所有的考生都交完卷後，需要將控制執行緒關閉
實現思想：用DelayQueue儲存考生（Student類），每一個考生都有自己的名字和完成試卷的時間，Teacher執行緒對DelayQueue進行監控，收取完成試卷小於120分鐘的學生的試卷。當考試時間120分鐘到時，先關閉Teacher執行緒，然後強制DelayQueue中還存在的考生交卷。每一個考生交卷都會進行一次countDownLatch.countDown()，當countDownLatch.await()不再阻塞說明所有考生都交完捲了，而後結束考試。
package com.my.base.concurrent.delayQueue; import java.util.Iterator; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; /** *this project is created for my partactice. *In the project I will write the mybatis by myself * *2014-1-10 下午9:43:48 *@author 孫振超[email protected] */ public class Exam { /** * *2014-1-10 下午9:43:48 by 孫振超 * *@param args *void * @throws InterruptedException */ public static void main(String[] args) throws InterruptedException { // TODO Auto-generated method stub int studentNumber = 20; CountDownLatch countDownLatch = new CountDownLatch(studentNumber+1); DelayQueue< Student> students = new DelayQueue<Student>(); Random random = new Random(); for (int i = 0; i < studentNumber; i++) { students.put(new Student("student"+(i+1), 30+random.nextInt(120),countDownLatch)); } Thread teacherThread =new Thread(new Teacher(students)); students.put(new EndExam(students, 120,countDownLatch,teacherThread)); teacherThread.start(); countDownLatch.await(); System.out.println(" 考試時間到，全部交卷！"); } } class Student implements Runnable,Delayed{ private String name; private long workTime; private long submitTime; private boolean isForce = false; private CountDownLatch countDownLatch; public Student(){} public Student(String name,long workTime,CountDownLatch countDownLatch){ this.name = name; this.workTime = workTime; this.submitTime = TimeUnit.NANOSECONDS.convert(workTime, TimeUnit.NANOSECONDS)+System.nanoTime(); this.countDownLatch = countDownLatch; } @Override public int compareTo(Delayed o) { // TODO Auto-generated method stub if(o == null || ! (o instanceof Student)) return 1; if(o == this) return 0; Student s = (Student)o; if (this.workTime > s.workTime) { return 1; }else if (this.workTime == s.workTime) { return 0; }else { return -1; } } @Override public long getDelay(TimeUnit unit) { // TODO Auto-generated method stub return unit.convert(submitTime - System.nanoTime(), TimeUnit.NANOSECONDS); } @Override public void run() { // TODO Auto-generated method stub if (isForce) { System.out.println(name + " 交卷, 希望用時" + workTime + "分鐘"+" ,實際用時 120分鐘" ); }else { System.out.println(name + " 交卷, 希望用時" + workTime + "分鐘"+" ,實際用時 "+workTime +" 分鐘"); } countDownLatch.countDown(); } public boolean isForce() { return isForce; } public void setForce(boolean isForce) { this.isForce = isForce; } } class EndExam extends Student{ private DelayQueue<Student> students; private CountDownLatch countDownLatch; private Thread teacherThread; public EndExam(DelayQueue<Student> students, long workTime, CountDownLatch countDownLatch,Thread teacherThread) { super("強制收卷", workTime,countDownLatch); this.students = students; this.countDownLatch = countDownLatch; this.teacherThread = teacherThread; } @Override public void run() { // TODO Auto-generated method stub teacherThread.interrupt(); Student tmpStudent; for (Iterator<Student> iterator2 = students.iterator(); iterator2.hasNext();) { tmpStudent = iterator2.next(); tmpStudent.setForce(true); tmpStudent.run(); } countDownLatch.countDown(); } } class Teacher implements Runnable{ private DelayQueue<Student> students; public Teacher(DelayQueue<Student> students){ this.students = students; } @Override public void run() { // TODO Auto-generated method stub try { System.out.println(" test start"); while(!Thread.interrupted()){ students.take().run(); } } catch (Exception e) { // TODO: handle exception e.printStackTrace(); } } }
業務場景二：具有過期時間的快取
這個場景中幾個點需要注意：
- 當向快取中新增key-value對時，如果這個key在快取中存在並且還沒有過期，需要用這個key對應的新過期時間
- 為了能夠讓DelayQueue將其已儲存的key刪除，需要重寫實現Delayed介面新增到DelayQueue的DelayedItem的hashCode函式和equals函式
- 當快取關閉，監控程式也應關閉，因而監控執行緒應當用守護執行緒
具體實現如下：
package com.my.base.concurrent.delayQueue; import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; /** *Cache.java * * Created on 2014-1-11 上午11:30:36 by sunzhenchao [email protected] */ public class Cache<K, V> { public ConcurrentHashMap<K, V> map = new ConcurrentHashMap<K, V>(); public DelayQueue<DelayedItem<K>> queue = new DelayQueue<DelayedItem<K>>(); public void put(K k,V v,long liveTime){ V v2 = map.put(k, v); DelayedItem<K> tmpItem = new DelayedItem<K>(k, liveTime); if (v2 != null) { queue.remove(tmpItem); } queue.put(tmpItem); } public Cache(){ Thread t = new Thread(){ @Override public void run(){ dameonCheckOverdueKey(); } }; t.setDaemon(true); t.start(); } public void dameonCheckOverdueKey(){ while (true) { DelayedItem<K> delayedItem = queue.poll(); if (delayedItem != null) { map.remove(delayedItem.getT()); System.out.println(System.nanoTime()+" remove "+delayedItem.getT() +" from cache"); } try { Thread.sleep(300); } catch (Exception e) { // TODO: handle exception } } } /** * TODO * @param args * 2014-1-11 上午11:30:36 * @author:孫振超 * @throws InterruptedException */ public static void main(String[] args) throws InterruptedException { Random random = new Random(); int cacheNumber = 10; int liveTime = 0; Cache<String, Integer> cache = new Cache<String, Integer>(); for (int i = 0; i < cacheNumber; i++) { liveTime = random.nextInt(3000); System.out.println(i+" "+liveTime); cache.put(i+"", i, random.nextInt(liveTime)); if (random.nextInt(cacheNumber) > 7) { liveTime = random.nextInt(3000); System.out.println(i+" "+liveTime); cache.put(i+"", i, random.nextInt(liveTime)); } } Thread.sleep(3000); System.out.println(); } } class DelayedItem<T> implements Delayed{ private T t; private long liveTime ; private long removeTime; public DelayedItem(T t,long liveTime){ this.setT(t); this.liveTime = liveTime; this.removeTime = TimeUnit.NANOSECONDS.convert(liveTime, TimeUnit.NANOSECONDS) + System.nanoTime(); } @Override public int compareTo(Delayed o) { if (o == null) return 1; if (o == this) return 0; if (o instanceof DelayedItem){ DelayedItem<T> tmpDelayedItem = (DelayedItem<T>)o; if (liveTime > tmpDelayedItem.liveTime ) { return 1; }else if (liveTime == tmpDelayedItem.liveTime) { return 0; }else { return -1; } } long diff = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS); return diff > 0 ? 1:diff == 0? 0:-1; } @Override public long getDelay(TimeUnit unit) { return unit.convert(removeTime - System.nanoTime(), unit); } public T getT() { return t; } public void setT(T t) { this.t = t; } @Override public int hashCode(){ return t.hashCode(); } @Override public boolean equals(Object object){ if (object instanceof DelayedItem) { return object.hashCode() == hashCode() ?true:false; } return false; } }