java中延时队列的使用

我就是我 2024-02-19 18:24 3阅读 0赞

最近遇到这么一个需求，程序中有一个功能需要发送短信，当满足某些条件后，如果上一步的短信还没有发送出去，那么应该取消这个短信的发送。在翻阅java的api后，发现java中有一个延时队列可以解决这个问题。

**实现思路**：

1、需要延时发送消息的实体类实现 **Delayed** 接口,在getDelay方法中，返回这个对象还有多上时间需要执行。

2、将短信类加入到延时队列**DelayQueye**中

**实现效果：**

将短信Sms加入到队列中，到了invokedTime后从队列中直接取出执行。

主线程中启动一个子线程，打印当前时间，查看队列中的数据执行的时机是否正确。

delayqueue 的take方法，获取并移除此队列的头部，在可从此队列获得到期延迟的元素之前一直等待（如有必要）

**代码如下**：

@Slf4j
    public class DelayTest {
    	@Data
    	@Builder
    	static class Sms implements Delayed {
    		// 执行时间
    		private long invokedTime;
    		// 需要发出的消息
    		private String smg;
    
    		public static Sms build(String msg, int expireTimeSecond) {
    			return Sms.builder().smg(msg).invokedTime(System.nanoTime() + TimeUnit.NANOSECONDS.convert(expireTimeSecond, TimeUnit.SECONDS)).build();
    		}
    
    		/**
    		 * 返回这个对象剩余的延时时间
    		 *
    		 * @param unit
    		 * @return
    		 */
    		@Override
    		public long getDelay(TimeUnit unit) {
    			return unit.convert(invokedTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    		}
    
    		@Override
    		public int compareTo(Delayed o) {
    			if (o == null || !(o instanceof Sms))
    				return 1;
    			if (o == this)
    				return 0;
    			Sms sms = (Sms) o;
    			if (this.invokedTime > sms.invokedTime) {
    				return 1;
    			} else if (this.invokedTime == sms.invokedTime) {
    				return 0;
    			} else {
    				return -1;
    			}
    		}
    	}
    
    	public static void main(String[] args) throws InterruptedException {
    		DelayQueue<Sms> queue = new DelayQueue<>();
    		queue.add(Sms.build("3-hello3", 3));
    		queue.add(Sms.build("5-hello5", 5));
    		queue.add(Sms.build("2-hello2", 2));
    		queue.add(Sms.build("4-hello4", 1));
    		queue.add(Sms.build("1-hello1", 1));
    		new Thread(() -> {
    			Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
    				log.info(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
    			}, 0, 1, TimeUnit.SECONDS);
    		}).start();
    
    		Sms sms;
    		while (null != (sms = queue.take())) {
    			System.out.println(sms);
    		}
    	}
    }

执行结果：  
![aHR0cDovL2RsMi5pdGV5ZS5jb20vdXBsb2FkL2F0dGFjaG1lbnQvMDEyNy83MTc2L2EwNzM4MjAyLTI5MzEtM2U3OC1iMTM4LTQ2Nzg0YTY5ZWNkMi5wbmc][]

[aHR0cDovL2RsMi5pdGV5ZS5jb20vdXBsb2FkL2F0dGFjaG1lbnQvMDEyNy83MTc2L2EwNzM4MjAyLTI5MzEtM2U3OC1iMTM4LTQ2Nzg0YTY5ZWNkMi5wbmc]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/ed5e7800357d4f2dade8fea432d453c6.png