记录Delay延时队列的使用

Posted 2021-05-13 猫与梧桐

记录Delay延时队列的使用

• 场景
• • 延时队列DelayQueue
  • 源码
  • 结合springboot简单应用

场景

在公司业务场景中，偶然出现了一个量较小的需求：在某个业务系统处理完成之后，需要做一个临时操作，例如：用户申请某web网页账号后给用户指定账号发送一份邮件提示用户账号已开通… 等等，如果跟业务系统耦合，会影响到业务处理的速度，本来想用MQ做一个队列处理消息，但是系统消息量不大，尝试使用原生API DelayQueue延时队列实现。

延时队列DelayQueue

java.util包中提供了现成的轮子，先扒一个main示例跑一下效果。

 

最直观的表现是用random随机创建时间，出队列的时间却根据设置时间顺序执行，扒一下源码看一下。

1 // An highlighted block
2 public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
3     implements BlockingQueue<E>

 

入列元素需要实现Delayed接口 ，接口提供了以下几个方法：

 1  @Override
 2     public long getDelay(TimeUnit unit) {
 3 //        return unit.convert(this.expire - System.currentTimeMillis(),unit);
 4         return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
 5     }
 6 
 7     @Override
 8     public int compareTo(Delayed o) {
 9 //        long delta = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
10 //        return (int) delta;
11         return (int)(this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
12     }
13 
14     @Override
15     public boolean equals(Object o) {
16         if (o instanceof DelayTask){
17             return String.valueOf(this.data.getIdentifier()).equals(String.valueOf(((DelayTask) o).getData().getIdentifier()));
18         }
19         return false;
20     }

 

其中getDelay是其中最核心的方法，在上面main示例中，delay是一个延时队列，它在初始化init时是一个阻塞等待的线程，getDelay调用时，队列中元素会判断当前任务是否已经达到过期时间，如果过期我们才可以通过take取出元素。
我们在自定义Delay元素时，需要指定一个过期时间戳

 1 private final long delayTime; //延迟时间
 2     private final long expire;  //到期时间
 3     private String data;   //数据
 4 
 5     //通过构造，设置过期时间 ，默认单位为毫秒 如果修改时间单位，使用TimeUnit.MILLISECONDS 去设置时间类型
 6     public DelayTask(long delay, String data) {
 7         delayTime = delay;
 8         this.data = data;
 9         expire = System.currentTimeMillis() + delay; 
10     }

 

在我们添加元素时，根据重写compareTo方法，队列会根据返回值进行排序。

 

源码

 1     private final transient ReentrantLock lock = new ReentrantLock();//获取锁
 2     private final PriorityQueue<E> q = new PriorityQueue<E>();//队列
 3     private Thread leader = null;//最快出队列的阻塞元素
 4 
 5     public boolean add(E e) {
 6             return offer(e);
 7         }
 8     public boolean offer(E e) {
 9         final ReentrantLock lock = this.lock;
10         lock.lock();
11         try {
12             q.offer(e);
13             if (q.peek() == e) {
14                 leader = null;
15                 available.signal();
16             }
17             return true;
18         } finally {
19             lock.unlock();
20         }
21     }
22     public void put(E e) {
23         offer(e);
24     }

能看出，所有提供的入列方法，其实最后都调用了offer()，先获取锁，给线程上锁，加入元素时判断加入的元素是否为最头元素，leader设置为空并唤醒线程。

 1 public E take() throws InterruptedException {
 2         final ReentrantLock lock = this.lock;
 3         lock.lockInterruptibly();
 4         try {
 5             for (;;) {
 6                 E first = q.peek();
 7                 if (first == null)
 8                     available.await();
 9                 else {
10                     long delay = first.getDelay(NANOSECONDS);
11                     if (delay <= 0)
12                         return q.poll();
13                     first = null; // don\'t retain ref while waiting
14                     if (leader != null)
15                         available.await();
16                     else {
17                         Thread thisThread = Thread.currentThread();
18                         leader = thisThread;
19                         try {
20                             available.awaitNanos(delay);
21                         } finally {
22                             if (leader == thisThread)
23                                 leader = null;
24                         }
25                     }
26                 }
27             }
28         } finally {
29             if (leader == null && q.peek() != null)
30                 available.signal();
31             lock.unlock();
32         }
33     }
34     
35     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
36         long nanos = unit.toNanos(timeout);
37         final ReentrantLock lock = this.lock;
38         lock.lockInterruptibly();
39         try {
40             for (;;) {
41                 E first = q.peek();
42                 if (first == null) {
43                     if (nanos <= 0)
44                         return null;
45                     else
46                         nanos = available.awaitNanos(nanos);
47                 } else {
48                     long delay = first.getDelay(NANOSECONDS);
49                     if (delay <= 0)
50                         return q.poll();
51                     if (nanos <= 0)
52                         return null;
53                     first = null; // don\'t retain ref while waiting
54                     if (nanos < delay || leader != null)
55                         nanos = available.awaitNanos(nanos);
56                     else {
57                         Thread thisThread = Thread.currentThread();
58                         leader = thisThread;
59                         try {
60                             long timeLeft = available.awaitNanos(delay);
61                             nanos -= delay - timeLeft;
62                         } finally {
63                             if (leader == thisThread)
64                                 leader = null;
65                         }
66                     }
67                 }
68             }
69         } finally {
70             if (leader == null && q.peek() != null)
71                 available.signal();
72             lock.unlock();
73         }
74     }

从队列中获取元素，如果获取到最前的元素，线程被阻塞，如果获取元素为null，则锁被释放。然后根据元素中getDelay方法获取一个int的返回值，这里就是我们构造中设置的过期时间，如果当前时间小于0，说明当前任务已经过期，我们可以取出过期元素；如果元素还在等待中，并且leader不为空，释放锁。
finally如果队列中还有元素且leader中没有阻塞元素时，全局释放锁。

结合springboot简单应用

 1 public class DelayManager implements CommandLineRunner{} 
 2 //创建一个外部管理类实现CommandLineRunner，重写run方法，这里实现runner是为了在springboot加载完成调用run方法时会启动队列
 3 
 4 /*
 5 **省略定义的add remove 等方法
 6 */
 7 ExecutorService executorService = Executors.newSingleThreadExecutor();
 8         executorService.execute(new Runnable() {
 9             @Override
10             public void run() {
11                 excuteThread();
12             }
13         });
14 //从队列中获取元素，一旦元素到期拿出元素处理
15      while(true){
16             try {
17                 DelayTask task = delayQueue.take();
18                 System.out.println(task.toString());
19                 processTask(task);//获取到任务后的具体业务处理
20             } catch (Exception e) {
21                 e.printStackTrace();
22                 continue;
23             }
24         }

记录Delay延时队列的使用_猫与梧桐的博客-CSDN博客