Java高并发专题之37、如何实现一个通用的延迟队列？

电商大家都用过吧，下单后若未支付，通常都有一段支付倒计时，比如15分钟，若时间到了之后，还未支付的，订单将被关闭，库存将被释放。

这种业务就需要用到延迟队列的功能，将任务丢到延迟队列、设置一个延迟时间、回调函数，到了时间之后，延迟队列将回调指定的函数消费指定的任务。

下面代码是一个通用的延迟队列的实现，大家可以直接拿去用。

代码还是比较简单的，技术要点：

  • 调用addTask方法将任务丢到延迟队列中，主要参数（延迟时间、任务信息、回调【任务到期后会进行回调】）
  • 使用到了java中的延迟队列DelayQueue来存放延迟任务
  • 下面的构造方法会自动调用一个start方法，start方法中会自动启动一个线程，线程轮询从延迟队列中拉取到期的任务，然后丢到线程池executorService.submit中进行处理，会自动调用创建延迟任务中指定的回调函数
  • main方法中有使用步骤
  1. import java.util.concurrent.*;
  2. import java.util.function.Consumer;
  3. import java.util.logging.Logger;
  5. public class DelayQueueService<T> {
  6.     Logger logger = Logger.getLogger(DelayQueueService.class.getName());
  7.     //延迟队列名称
  8.     private String delayQueueName;
  9.     private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
  10.     //处理队列中任务的线程池
  11.     private ExecutorService executorService;
  13.     public DelayQueueService(String delayQueueName) {
  14.         this(delayQueueName, Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 4));
  15.     }
  17.     public DelayQueueService(String delayQueueName, ExecutorService executorService) {
  18.         this.delayQueueName = delayQueueName;
  19.         this.executorService = executorService;
  20.         //启动队列消费
  21.         this.start();
  22.     }
  24.     /**
  25.      * 添加任务
  26.      *
  27.      * @param delayedTimeUnit 延迟时间单位
  28.      * @param delayedTime     延迟时间
  29.      * @param task            任务
  30.      * @param consumer        任务消费者（到期了会回调）
  31.      */
  32.     public void addTask(TimeUnit delayedTimeUnit, long delayedTime, T task, Consumer<T> consumer) {
  33.         this.delayQueue.offer(new DelayedTask(delayedTimeUnit, delayedTime, task, consumer));
  34.     }
  36.     private void start() {
  37.         //轮询从延迟队列中拉取任务，然后调用线程池进行处理
  38.         Thread pollThread = new Thread(() -> {
  39.             while (true) {
  40.                 try {
  41.                     DelayedTask delayedTask = this.delayQueue.poll(100, TimeUnit.MILLISECONDS);
  42.                     if (this.executorService.isShutdown()) {
  43.                         break;
  44.                     }
  45.                     if (delayedTask != null) {
  46.                         executorService.submit(() -> {
  47.                             delayedTask.consumer.accept(delayedTask.task);
  48.                         });
  49.                     }
  50.                 } catch (InterruptedException e) {
  51.                     logger.warning(e.getMessage());
  52.                 }
  53.             }
  54.         });
  55.         pollThread.setDaemon(Thread.currentThread().isDaemon());
  56.         pollThread.setName(this.getClass().getName() + -pollThread- + this.delayQueueName);
  57.         pollThread.start();
  58.     }
  60.     public void close() {
  61.         if (!this.executorService.isShutdown()) {
  62.             this.executorService.shutdown();
  63.         }
  64.     }
  66.     public class DelayedTask implements Delayed {
  67.         //延迟时间单位
  68.         private TimeUnit delayedTimeUnit;
  69.         //延迟时间
  70.         private long delayedTime;
  71.         //到期时间(毫秒)
  72.         private long endTime;
  73.         //延迟任务信息
  74.         private T task;
  75.         //消费者
  76.         private Consumer<T> consumer;
  78.         public DelayedTask(TimeUnit delayedTimeUnit, long delayedTime, T task, Consumer<T> consumer) {
  79.             this.delayedTimeUnit = delayedTimeUnit;
  80.             this.delayedTime = delayedTime;
  81.             this.task = task;
  82.             this.endTime = System.currentTimeMillis() + delayedTimeUnit.toMillis(delayedTime);
  83.             this.consumer = consumer;
  84.         }
  86.         @Override
  87.         public long getDelay(TimeUnit unit) {
  88.             return unit.convert(this.endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
  89.         }
  91.         @Override
  92.         public int compareTo(Delayed o) {
  93.             DelayedTask task = (DelayedTask) o;
  94.             return Long.compare(this.endTime, task.endTime);
  95.         }
  96.     }
  98.     public static void main(String[] args) {
  99.         //创建一个延迟队列：用来对超过支付日期的订单进行关闭
  100.         String delayQueueName = orderCloseDelayQueue;
  101.         //1、创建延迟队列
  102.         DelayQueueService<String> orderCloseDelayQueue = new DelayQueueService<String>(delayQueueName);
  103.         for (int i = 1; i <= 10; i++) {
  104.             //2、调用addTask将延迟任务加入延迟队列
  105.             orderCloseDelayQueue.addTask(TimeUnit.SECONDS, i, 订单 + i, new Consumer<String>() {
  106.                 @Override
  107.                 public void accept(String s) {
  108.                     System.out.println(System.currentTimeMillis() + , + Thread.currentThread() + ,关闭订单: + s);
  109.                 }
  110.             });
  111.         }
  112.         //3、系统关闭的时候，调用延迟队列的close方法
  113.         //orderCloseDelayQueue.close();
  114.     }
  115. }

main方法中模拟了10个延迟任务，运行看看效果，输出

  1. 1614346780438,Thread[pool-1-thread-1,5,main],关闭订单:订单1
  2. 1614346781437,Thread[pool-1-thread-2,5,main],关闭订单:订单2
  3. 1614346782436,Thread[pool-1-thread-3,5,main],关闭订单:订单3
  4. 1614346783437,Thread[pool-1-thread-4,5,main],关闭订单:订单4
  5. 1614346784437,Thread[pool-1-thread-5,5,main],关闭订单:订单5
  6. 1614346785437,Thread[pool-1-thread-6,5,main],关闭订单:订单6
  7. 1614346786437,Thread[pool-1-thread-7,5,main],关闭订单:订单7
  8. 1614346787436,Thread[pool-1-thread-8,5,main],关闭订单:订单8
  9. 1614346788437,Thread[pool-1-thread-9,5,main],关闭订单:订单9
  10. 1614346789437,Thread[pool-1-thread-10,5,main],关闭订单:订单10
来源：http://itsoku.com/course/1/205