1、 创建延时消息的实体类

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/** * 延时消息实体类 * @program: ReservedInstance * @description: DelayMessage * @author: sunyuhua * @create: 2021-09-18 11:53 **/
public class DelayMessage  implements Delayed {

	//延迟任务中的任务数据
	private  String  message;
	//延迟任务到期时间
	private long ttl;

	public DelayMessage(String message,long ttl) {
		setMessage(message);
		this.ttl=System.currentTimeMillis()+ttl;
	}

	/** * 计算任务距离过期还有多长时间(单位：MILLISECONDS) * @param unit * @return */
	@Override
	public long getDelay(TimeUnit unit) {
		long remaining=ttl-System.currentTimeMillis();
		return unit.convert(remaining,TimeUnit.MILLISECONDS);
	}

	/** * 比较，排序；对任务的延时大小进行排序，将延时时间最小的任务放在队列头部 * @param o * @return */
	@Override
	public int compareTo(Delayed o) {
		return (int)(this.getDelay(TimeUnit.MILLISECONDS)-o.getDelay(TimeUnit.MILLISECONDS));
	}

	public String getMessage() {
		return message;
	}

	public void setMessage(String message) {
		this.message = message;
	}
}

2、延迟队列的消费者定义类

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;

/** * 延迟队列的消费者定义类 * @program: ReservedInstance * @description: DelayQueueConsumer * @author: sunyuhua * @create: 2021-09-18 14:10 **/
@Slf4j
public class DelayQueueConsumer implements Runnable{

 private final DelayQueue<DelayMessage> delayQueue;

 public DelayQueueConsumer(DelayQueue<DelayMessage> delayQueue){
  this.delayQueue=delayQueue;
 }

 @Override
 public void run() {
  while(true){
   try {
    DelayMessage delayMessage=delayQueue.take();
    log.info("消费延迟队列中的消息:{}",delayMessage.getMessage());
   }catch (Exception e){
    e.printStackTrace();
   }
  }
 }
}

3、延迟队列的生产者消息对象

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.DelayQueue;

/** * 延迟队列消息的生产者 * @program: ReservedInstance * @description: DelayQueueProducer * @author: sunyuhua * @create: 2021-09-18 14:31 **/
@Slf4j
public class DelayQueueProducer {

 public static void main(String[] args) {
  //创建延时消息队列
  DelayQueue<DelayMessage> delayQueue=new DelayQueue<>();
  //创建并启动延迟队列的消费者线程
  new Thread(new DelayQueueConsumer(delayQueue)).start();


  /** * 生成5条TTL时间，依次增大的延时消息：1,2,3,4,5秒 */
  for (int i = 1; i <= 5; i++) {
   DelayMessage delayMessage = new DelayMessage(String.valueOf(i), i*2000L);
   log.info("生产延迟队列中的消息: {}", String.valueOf(i));
   delayQueue.offer(delayMessage);
  }

  /** * 生成5条TTL时间依次减小的延迟消息：5秒，4秒，3秒，2秒，1秒 */
  for (int i = 5; i > 0; i--) {
   String message = String.valueOf(i);
   DelayMessage delayMessage = new DelayMessage(message, i*1000L);
   log.info("生产延迟队列中的消息: {}", message);
   delayQueue.offer(delayMessage);
  }

  /** * 生成5个延迟时间随机的延迟消息 */
  Random random = new Random();
  for (int i = 0; i < 5; i++) {
   // 生成1~10的随机数：作为1秒-10秒的延迟时间
   int ttl = 1 + random.nextInt(10);
   String message = String.valueOf(ttl);
   DelayMessage delayMessage = new DelayMessage(message, ttl*1000L);
   log.info("生产延迟队列中的消息: {}", message);
   delayQueue.offer(delayMessage);
  }

 }
}

查看结果