延时任务通知服务的设计及实现（三）-- JDK的延迟队列DelayQueue

本文主要是介绍延时任务通知服务的设计及实现（三）-- JDK的延迟队列DelayQueue，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、接着上文

上文我们讲述了使用redisson的RDelayedQueue实现分布式延迟队列，本文我们将自己JDK的延迟队列DelayQueue实现。

相比前者的实现，作为进程内的延迟队列，它会遇到许多技术难点：

• 如何支持分布式的多个节点部署场景
• 应用重启会恢复延时队列
• 冷数据如何转换为热数据
• 如何删除延迟队列中的任务

随后，我们也将提及：

• 保存任务至延迟队列（生产者）
• 读取延迟队列中的任务（消费者）

二、设计概要

• 冷数据：mysql表中的任务数据

• 热数据：jdk 延迟队列中的任务

• 广播事件：删除延迟队列中的任务，发布的是广播事件，可以使用redis topic实现。

• 本地事件：分布式多节点部署的时候，每个任务只保存在其中一个节点的延迟队列中，可以使用spring事件驱动实现。

• 延迟队列 DelayQueueJob， 它实现了接口Delayed

包括任务的交易流水号和过期时间（即任务的回调时间）

import lombok.Builder;
import lombok.Data;import java.util.Date;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;/*** @author xxx*/
@Builder
@Data
public class DelayQueueJob implements Delayed {/*** 交易流水号*/private String transNo;/*** 到期时间*/private Date expireDate;public DelayQueueJob(String transNo, Date expireDate) {super();this.transNo = transNo;this.expireDate = expireDate;}/*** 用于队列中排序过期时间** @param o* @return*/@Overridepublic int compareTo(Delayed o) {return Long.valueOf(this.expireDate.getTime()).compareTo(Long.valueOf(((DelayQueueJob) o).expireDate.getTime()));}/*** 用于获取过期时间* 延迟关闭时间 = 过期时间 - 当前时间** @param unit* @return*/@Overridepublic long getDelay(TimeUnit unit) {return this.expireDate.getTime() - System.currentTimeMillis();}
}

三、应用启动流程

解决恢复延迟队列的问题。因为DelayQueue是进程内的，一旦重启，将被销毁。

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;import java.util.List;
import java.util.concurrent.TimeUnit;@Slf4j
@Service
@RequiredArgsConstructor
public class ApplicationStartupListener implements ApplicationListener<ApplicationReadyEvent> {@Overridepublic void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {// 实现代码参考上面的流程图}
}

四、定时任务流程

解决冷数据如何转换为热数据的问题，防止延时任务过多导致消耗过多的jvm内存，所以只有回调时间将近的任务才放入延迟队列。

五、如何删除延迟队列中的任务

删除延迟队列的任务：发送广播消息通知所有的节点，当不是当前节点的时候，执行删除。

if (!NetUtil.getLocalhostStr().equals(ipAddress)) {DelayQueueSingleton.getDelayQueue().remove(transNo);
}

DelayQueueSingletons是一个单例类，详见下：

public class DelayQueueSingleton {private static volatile CustomDelayQueue<DelayQueueJob> delayQueue;private DelayQueueSingleton() {}public static CustomDelayQueue<DelayQueueJob> getDelayQueue() {if (delayQueue == null) {synchronized (DelayQueueSingleton.class) {if (delayQueue == null) {delayQueue = new CustomDelayQueue<>();}}}return delayQueue;}}

这里为了删除延迟队列的任务，我们对DelayQueue进行了重写。

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;public class CustomDelayQueue<T extends Delayed> {private final DelayQueue<T> queue = new DelayQueue<>();private final Map<String, T> map = new ConcurrentHashMap<>();public boolean put(T task, String taskId) {// 如果任务已存在，则删除旧任务，防止重复添加this.remove(taskId);map.put(taskId, task);return queue.add(task);}public boolean remove(String taskId) {// 先删除map，再删除queueT task = map.remove(taskId);if (task != null) {return queue.remove(task);}return false;}public T take() throws InterruptedException {return queue.take();}
}

六、保存任务至延迟队列（生产者）

// 如果通知时间在一定时间范围内
if (DateUtil.offsetMinute(new DateTime(), commonConfig.getHotDataTimeLine()).after(event.getNotifyDate())) {DelayQueueSingleton.getDelayQueue().put(DelayQueueJob.builder().transNo(event.getTransNo()).expireDate(event.getNotifyDate()).build(), event.getTransNo());}

七、读取延迟队列中的任务（消费者）

作为延迟队列的消费者，它的实现和上一篇文章实现类似。不同的是take()获取任务不一样。

String transNo = null;
Date notifyDate = null;DelayQueueJob job = DelayQueueSingleton.getDelayQueue().take();
if (null != job) {transNo = job.getTransNo();notifyDate = job.getExpireDate();
}if (null == transNo) {return;
}if (log.isInfoEnabled()) {log.info("开始执行延迟队列中的任务,transNo={},notifyDate={}", transNo, notifyDate);
}// 异步执行你的操作
notifyTaskService.handleTask(transNo, notifyDate);

八、总结

作为进程内的延迟队列，在多点部署的分布式集群环境下， 代码明显比上一篇要复杂得多。

它们都需要的步骤是：

• 任务的生产
• 任务的消费
• 移除任务

DelayQueue额外多出来的步骤是：

• 应用启动的时候拉取回调时间将近的未完成任务（更新marked标记为true，防止重复拉取冷数据）
• 定时拉取未标记且回调时间将近的未完成任务（和上面必须是互斥，等待上一步执行完成，否则会导致重复拉取）
• 删除延迟队列DelayQueue的任务，必须发布广播消息给全部节点。（引入广播消息机制）

由此可见，任务表的字段marked仅供DelayQueue使用，防止重复拉取数据库的任务到热数据区。

    @Column(name = "marked", nullable = false, columnDefinition = "TINYINT(1) default 0 COMMENT '是否已标记为热数据'")private Boolean marked;

附：相关系列文章链接

延时任务通知服务的设计及实现（一）-- 设计方案

延时任务通知服务的设计及实现（二）-- redisson的延迟队列RDelayedQueue

延时任务通知服务的设计及实现（三）-- JDK的延迟队列DelayQueue

延时任务通知服务的设计及实现（四）-- webhook执行任务

延时任务通知服务的设计及实现（五）-- Netty时间轮HashedWheelTimer

这篇关于延时任务通知服务的设计及实现（三）-- JDK的延迟队列DelayQueue的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---