延时队列DelayQueue实现方案

淡淡的烟草味﹌ 2023-02-24 07:53 78阅读 0赞

实现了一版过期时效的需求,

真实的需求是,给某个客户发一条消息,如果在指定的时间(x小时)内,客户没有响应,则再给该客户发送一条消息;

考虑到java本身有delayQueue的实现方案,所以在实现的时候,想的直接用,就可以...但是这种方式有两个问题存在

1,delayQueue是存储在内存中.所以重启或者系统down掉后,所有的消息都丢失了...(这个的话,后期使用了redis的存储,利用redis的key值失效策略,可以参考[Redis单机及集群实现key值过期监听机制][Redis_key]).

2,delayQueue对于分布式的应用(或者讲多应用实现的情况)是不适用的,(后期一想,其实这个也是适用的..不管有多少个应用实例,给客户发消息的时候,只是调用一个应用服务器,然后也就只有该应用服务器会存储或者执行该delayQueue,其他应用服务器对此无关联)....这个有大佬熟悉的话,还请多加指点指点...

下面就是测试delayQueue实现的简单demo实现

实体操作类

package com.young.springdemo.controller.delay;
    
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.ToString;
    
    /**
     * @program: springdemo
     * @description 操作实体类,被延迟队队列维护的实体
     * @author: Young
     * @create: 2020-08-08 14:05
     **/
    @Data
    @AllArgsConstructor
    @ToString
    public class Stuedent {
    
    
        private String name;
        private int age;
        private String address;
    
    
    }

延时队列

package com.young.springdemo.controller.delay;
 
 import lombok.Data;
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.concurrent.Delayed;
 import java.util.concurrent.TimeUnit;
 
 /**
 * @program: springdemo
 * @description
 * @author: Young
 * @create: 2020-08-08 14:02
 **/
 @Data
 @ToString
 @Slf4j
 public class DelayEntity implements Delayed {
 
 private Stuedent student;
 
 private long time;//到期时间
 
 public DelayEntity(Stuedent student, long time, TimeUnit unit) {
 this.student = student;
 //计算到期时间,即time.unit后到期
 this.time = unit.toMillis(time) + System.currentTimeMillis();
 }
 
 /**
 * 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
 */
 //重写延迟的方法
 @Override
 public long getDelay(TimeUnit unit) {
 //根据给的时间单位返回剩余到期时间
 return unit.convert(time - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
 
 }
 
 /**
 * Compares this object with the specified object for order. Returns a
 * negative integer, zero, or a positive integer as this object is less
 * than, equal to, or greater than the specified object.
 * 
 * The implementor must ensure <tt>sgn(x.compareTo(y)) ==
 * -sgn(y.compareTo(x))</tt> for all <tt>x</tt> and <tt>y</tt>. (This
 * implies that <tt>x.compareTo(y)</tt> must throw an exception iff
 * <tt>y.compareTo(x)</tt> throws an exception.)
 * 
 * The implementor must also ensure that the relation is transitive:
 * <tt>(x.compareTo(y)>0 && y.compareTo(z)>0)</tt> implies
 * <tt>x.compareTo(z)>0</tt>.
 * 
 * Finally, the implementor must ensure that <tt>x.compareTo(y)==0</tt>
 * implies that <tt>sgn(x.compareTo(z)) == sgn(y.compareTo(z))</tt>, for
 * all <tt>z</tt>.
 * 
 * It is strongly recommended, but not strictly required that
 * <tt>(x.compareTo(y)==0) == (x.equals(y))</tt>. Generally speaking, any
 * class that implements the <tt>Comparable</tt> interface and violates
 * this condition should clearly indicate this fact. The recommended
 * language is "Note: this class has a natural ordering that is
 * inconsistent with equals."
 * 
 * In the foregoing description, the notation
 * <tt>sgn(</tt>expression<tt>)</tt> designates the mathematical
 * signum function, which is defined to return one of <tt>-1</tt>,
 * <tt>0</tt>, or <tt>1</tt> according to whether the value of
 * expression is negative, zero or positive.
 *
 * @param o the object to be compared.
 * @return a negative integer, zero, or a positive integer as this object
 * is less than, equal to, or greater than the specified object.
 * @throws NullPointerException if the specified object is null
 * @throws ClassCastException if the specified object's type prevents it
 * from being compared to this object.
 */
 //重写比较的方法
 @Override
 public int compareTo(Delayed o) {
 DelayEntity delayEntity = (DelayEntity) o;
 //重新排序
 return this.time - delayEntity.time <= 0 ? -1 : 1;
 }
 }

//测试类

package com.young.springdemo.controller.delay;
 
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.concurrent.DelayQueue;
 import java.util.concurrent.TimeUnit;
 
 /**
 * @program: springdemo
 * @description
 * @author: Young
 * @create: 2020-08-08 15:44
 **/
 @Slf4j
 public class TestMain {
 public static void main(String[] args) throws Exception{
 Stuedent xiaoming = new Stuedent("小明",12,"南京");
 Stuedent xiaohong = new Stuedent("小红",15,"北京");
 DelayEntity xiaomingD = new DelayEntity(xiaoming,5000, TimeUnit.MILLISECONDS);
 DelayEntity xiaoHongD = new DelayEntity(xiaohong,1000,TimeUnit.MILLISECONDS);
 DelayQueue<DelayEntity> queue = new DelayQueue<>();
 queue.put(xiaoHongD);
 queue.put(xiaomingD);
 while (true){
 if(queue.size()>0){
 DelayEntity poll = queue.take();
 log.info("poll : [{}]",poll.getStudent().toString());
 }else {
 break;
 }
 }
 }
 }

console:页面

![20200808155926774.png][]

[Redis_key]: https://blog.csdn.net/YoungLee16/article/details/107206360
[20200808155926774.png]: /images/20230209/af85b69765404918b44ef8bb63e03631.png