网站建设项目运作的可行性顾问
基于java中延时队列的实现该文章,我们这次主要是来实现基于DelayQueue
实现的延时队列。
使用DelayQueue
实现的延时队列的步骤:
- 定义一个继承了Delayed的类,定义其中的属性,并重写
compareTo
和getDelay
两个方法 - 创建一个
Delayqueue
用于创建队列 - 创建一个生产者,用于将信息添加到队列中
- 创建一个消费者,用来从队列中取出信息进行消费
接下来是一个简单的demo :
定义一个元素类
import lombok.Data;import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;@Data
public class DelayTesk implements Delayed {//标签Idprivate String uid;//到期时间private Long timestamp;//延时信息private String data;@Overridepublic long getDelay(TimeUnit unit) {long delayTime = timestamp - System.currentTimeMillis();//将时间转换成毫秒(这边可转可不转,影响不大)return unit.convert(delayTime, TimeUnit.MILLISECONDS);}@Overridepublic int compareTo(Delayed o) {//针对任务的延时时间长短进行排序,把延时时间最短的放在前面long differenceTime = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);return (int)differenceTime;}
}
定义一个延时队列
import java.util.concurrent.DelayQueue;public class DelayTaskQueue {/*** 这边使用单例模式进行创建,保证全局队列的唯一性* 我这边使用的是双检索,双校验模式*/private volatile static DelayQueue<DelayTesk> delayTaskQueue;private DelayTaskQueue(){}public static DelayQueue<DelayTesk> getDelayTaskQueue() {if (delayTaskQueue == null) {synchronized (DelayTaskQueue.class) {if (delayTaskQueue == null) {delayTaskQueue = new DelayQueue<>();}}}return delayTaskQueue;}
}
创建一个延时队列的生产者
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.DelayQueue;//消息生产者
@Slf4j
public class DelayTeskQueueProducer {/*** 往延时队列中插入数据* @param uid* @param time* @param data*/public static void setDelayQueue(String uid, Long time, String data) {//创建队列DelayQueue<DelayTesk> delayTaskQueue = DelayTaskQueue.getDelayTaskQueue();//创建任务DelayTesk delayTesk = new DelayTesk();delayTesk.setUid(uid);delayTesk.setTimestamp(time);delayTesk.setData(data);log.info("====================消息入队:{}===============", uid);boolean res = delayTaskQueue.offer(delayTesk);if (res) {log.info("====================消息入队成功:{}===============", uid);} else {//如果消息入队失败这边可以写一个失败的回调函数//例如将失败的消息存入数据库,写个定时任务对消息进行重写投递……log.info("====================消息入队失败:{}===============", uid);}}
}
定义一个延时队列的消费者
import cn.hutool.core.util.IdUtil;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.DelayQueue;@Slf4j
public class DelayTeskQueueConsumer {public static void main(String[] args) {for (int i = 0; i < 10 ; i++) {DelayTeskQueueProducer.setDelayQueue(IdUtil.fastUUID(), System.currentTimeMillis() + i * 1000, "hello world" + i);}int index = 0;DelayQueue<DelayTesk> delayTaskQueue = DelayTaskQueue.getDelayTaskQueue();while (index < 10) {try {DelayTesk delayTesk = delayTaskQueue.take();System.out.println(delayTesk.getData());} catch (InterruptedException e) {log.error("延时队列消费异常:{}", e.getMessage());}}}
}
结果
在控控制台中每隔1秒打印一行数据
到这差不多我们的Demo就要结束了,不过可能有些同学会问,你这个消费者不是是写在mian方法里的,每次消费的时候都需要手动去调用这跟我直接用sleep函数实现的延时队列有啥区别呀
别急 这个只是个Demo嘛,如果需要使用在项目中可以写一个监听器去实时监听该延时队列
我这边暂时就只讲3种
Timer
通过timer定时定频率去获取DelayTaskQueue
中的消息
import com.study.project.delay.DelayTaskQueue;
import com.study.project.delay.DelayTesk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.DelayQueue;/*** 添加@Configuration 注解,自动注入实例对象,并由springboot 启动 定时器,执行任务。*/@Configuration
@Slf4j
public class DelayTeskQueueTimer {@Beanpublic void DelayTeskQueueTimer() {log.info("====================监听开始====================");final Timer timer = new Timer();DelayQueue<DelayTesk> delayTaskQueue = DelayTaskQueue.getDelayTaskQueue();timer.schedule(new TimerTask() {@Overridepublic void run() {try {DelayTesk delayTesk = delayTaskQueue.take();System.out.println(delayTesk.getData());} catch (Exception e) {log.error("延时队列消费异常:{}", e.getMessage());}}//第一次执行是在当前时间的一秒之后,之后每隔一秒钟执行一次},1000, 1000);}
}
ConmandlineRunner
import com.study.project.delay.DelayTaskQueue;
import com.study.project.delay.DelayTesk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;import java.util.concurrent.DelayQueue;
/*** Spring Boot应用程序在启动后,程序从容器中遍历实现了CommandLineRunner接口的实例并运行它们的run方法*/
@Slf4j
@Configuration
public class DelayTeskQueueTimerCommandLineRunner implements CommandLineRunner {@Overridepublic void run(String... args) {log.info("====================CommandLineRunner监听开始====================");DelayQueue<DelayTesk> delayTaskQueue = DelayTaskQueue.getDelayTaskQueue();new Thread(() ->{while (true) {try {DelayTesk delayTesk = delayTaskQueue.take();System.out.println(delayTesk.getData());} catch (Exception e) {log.error("延时队列消费异常:{}", e.getMessage());}}}).start();}
}
ApplicationListener
该方法和ConmandlineRunner
方法一样 都是在Spring Boot应用程序在启动后,对DelayQueue
进行监听
import com.study.project.delay.DelayTaskQueue;
import com.study.project.delay.DelayTesk;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;import java.util.concurrent.DelayQueue;@Slf4j
@Configuration
public class DelayTeskQueueApplicationListener implements ApplicationListener<ContextRefreshedEvent> {@Overridepublic void onApplicationEvent(ContextRefreshedEvent event) {log.info("====================ApplicationListener监听开始====================");DelayQueue<DelayTesk> delayTaskQueue = DelayTaskQueue.getDelayTaskQueue();new Thread(() -> {while (true) {try {DelayTesk delayTesk = delayTaskQueue.take();System.out.println(delayTesk.getData());} catch (Exception e) {log.error("延时队列消费异常:{}", e.getMessage());}}}).start();}
}
当然监听的方法其实还有很多,不过同学们在实现队列的时候不要觉得实现了就好了,要去思考如何去保证数据的持久化,保证数据不会不会丢失