java 延迟阻塞队列详解

待我称王封你为后i 2022-09-25 11:18 195阅读 0赞

# [java 延迟阻塞队列详解][java] #

标签： [数据结构][Link 1][延迟队列][Link 2][DelayQueue][][缓存设计][Link 3][java][java 1]

2016-02-19 14:59  183人阅读  [评论][Link 4](0)  收藏  [举报][Link 5]

![category_icon.jpg][]  分类：

Technology（22） ![arrow_triangle_20_down.jpg][]

目录[(?)][Link 6][\[+\]][Link 6]

## DelayQueue 官方给的解释： ##

## Delayed 元素的一个无界[阻塞队列][Link 7]，只有在延迟期满时才能从中提取元素。该队列的头部 是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。即使无法使用 take 或 poll 移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size 方法同时返回到期和未到期元素的计数。此队列不允许使用 null 元素。 ##

A 也就是说过期的元素会按过期的时间顺序在队列里面排列

B 此队列不允许有null元素

那么为什么会有 DelayQueue呢.

典型场景:

一个缓存系统 ,每个缓存元素都有固定的缓存时间.也就是过期时间.如果自己做缓存设计.粗暴的就要一直循环缓存队列去清理过期的缓存.缓存数据多时候即造成性能问题,也会赵成过期时间不太准确.(循环导致的)

DelayQueue也就是用来解决这类问题的.很多开源的缓存底层都用它实现.用守护线程去DelayQueue里面不断获取过期的缓存.清理即可.这里先不去深究DelayQueue的实现原理 有机会补上.先看看怎么应用DelayQueue 去解决实际的问题.

引用了一片比较好的文章 写的很好 给大家分享(http://www.cnblogs.com/sunzhenchao/p/3515085.html)

**Delayed，**一种混合风格的接口，用来标记那些应该在给定延迟时间之后执行的对象。此接口的实现必须定义一个 compareTo 方法，该方法提供与此接口的 getDelay 方法一致的排序。

在网上也看到两个示例，但这两个示例个人在实际运行时均没有达到满足业务场景的效果，因而对其进行了修改，供大家参考讨论。

### 业务场景一：多考生考试 ###

该场景来自于[http://ideasforjava.iteye.com/blog/657384][http_ideasforjava.iteye.com_blog_657384]，模拟一个考试的日子，考试时间为120分钟，30分钟后才可交卷，当时间到了，或学生都交完卷了考试结束。

这个场景中几个点需要注意：

1.  考试时间为120分钟，30分钟后才可交卷，初始化考生完成试卷时间最小应为30分钟
2.  对于能够在120分钟内交卷的考生，如何实现这些考生交卷
3.  对于120分钟内没有完成考试的考生，在120分钟考试时间到后需要让他们强制交卷
4.  在所有的考生都交完卷后，需要将控制线程关闭

实现思想：用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  
 \*  
 \*/  
  
  
public class Exam \{  
  
  
    /\*\*  
     \*  
     \*  
     \*@param args  
     \* @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();  
        \}  
    \}  
      
\}

### 业务场景二：具有过期时间的缓存 ###

该场景来自于[http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html][http_www.cnblogs.com_jobs_archive_2007_04_27_730255.html]，向缓存添加内容时，给每一个key设定过期时间，系统自动将超过过期时间的key清除。

这个场景中几个点需要注意：

1.  当向缓存中添加key-value对时，如果这个key在缓存中存在并且还没有过期，需要用这个key对应的新过期时间
2.  为了能够让DelayQueue将其已保存的key删除，需要重写实现Delayed接口添加到DelayQueue的DelayedItem的hashCode函数和equals函数
3.  当缓存关闭，监控程序也应关闭，因而监控线程应当用守护线程

具体实现如下：

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  
 \*  
 \*/  
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  
     \* @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;  
    \}  
      
\}

总结：

元素继承Delayed 实现getDelay接口来确定是否放入到延迟队列中. 实现compareTo接口 来确定自己在延迟队列中的顺序.

在外部就可以从DelayQueue去不断去 过期的元素.取不到就阻塞.取到后逻辑处理或者清理即可.复杂的可以配合线程栅栏 让多个过期元素时来触发特定逻辑.（这玩意真是好东西 虽然现在做移动端.不过学习还是要学习的,如果有问题或者想交流的 欢迎拍砖）

[java]: http://blog.csdn.net/maomaolingyu/article/details/50697848
[Link 1]: http://www.csdn.net/tag/%e6%95%b0%e6%8d%ae%e7%bb%93%e6%9e%84
[Link 2]: http://www.csdn.net/tag/%e5%bb%b6%e8%bf%9f%e9%98%9f%e5%88%97
[DelayQueue]: http://www.csdn.net/tag/DelayQueue
[Link 3]: http://www.csdn.net/tag/%e7%bc%93%e5%ad%98%e8%ae%be%e8%ae%a1
[java 1]: http://www.csdn.net/tag/java
[Link 4]: http://blog.csdn.net/maomaolingyu/article/details/50697848#comments
[Link 5]: http://blog.csdn.net/maomaolingyu/article/details/50697848#report
[category_icon.jpg]: /images/20220718/b91ce2e1975042a1918fae92ce3cd2a0.png
[arrow_triangle_20_down.jpg]: /images/20220718/bfedd69a4cdf4ed4b65441b1dcbfc866.png
[Link 6]: http://blog.csdn.net/maomaolingyu/article/details/50697848#
[Link 7]: http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/BlockingQueue.html
[http_ideasforjava.iteye.com_blog_657384]: http://ideasforjava.iteye.com/blog/657384
[Java]: http://lib.csdn.net/base/17
[http_www.cnblogs.com_jobs_archive_2007_04_27_730255.html]: http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html