java利用delayedQueue实现本地的延迟队列

一、了解delayqueue

delayqueue是什么？

delayqueue是一个*的blockingqueue，用于放置实现了delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是有序的，即队头对象的延迟到期时间最长。

注意：不能将null元素放置到这种队列中。

delayqueue能做什么？

在我们的业务中通常会有一些需求是这样的：

• 淘宝订单业务:下单之后如果三十分钟之内没有付款就自动取消订单。
• 饿了吗订餐通知:下单成功后60s之后给用户发送短信通知。
  

那么这类业务我们可以总结出一个特点:需要延迟工作。
由此的情况，就是我们的delayqueue应用需求的产生。

二、怎么用delayqueue来解决这类的问题

先声明一个delayed的对象

import java.util.concurrent.delayed;
import java.util.concurrent.timeunit;
import java.util.concurrent.atomic.atomiclong;

/**
 * <p>
 * [任务调度系统]
 * <br>
 * [队列中要执行的任务]
 * </p>
 *
 * @author wangguangdong
 * @version 1.0
 * @date 2015年11月22日19:46:39
 */
public class task<t extends runnable> implements delayed {
 /**
  * 到期时间
  */
 private final long time;

 /**
  * 问题对象
  */
 private final t task;
 private static final atomiclong atomic = new atomiclong(0);

 private final long n;

 public task(long timeout, t t) {
  this.time = system.nanotime() + timeout;
  this.task = t;
  this.n = atomic.getandincrement();
 }

 /**
  * 返回与此对象相关的剩余延迟时间，以给定的时间单位表示
  */
 @override
 public long getdelay(timeunit unit) {
  return unit.convert(this.time - system.nanotime(), timeunit.nanoseconds);
 }

 @override
 public int compareto(delayed other) {
  // todo auto-generated method stub
  if (other == this) // compare zero only if same object
   return 0;
  if (other instanceof task) {
   task x = (task) other;
   long diff = time - x.time;
   if (diff < 0)
    return -1;
   else if (diff > 0)
    return 1;
   else if (n < x.n)
    return -1;
   else
    return 1;
  }
  long d = (getdelay(timeunit.nanoseconds) - other.getdelay(timeunit.nanoseconds));
  return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
 }

 public t gettask() {
  return this.task;
 }

 @override
 public int hashcode() {
  return task.hashcode();
 }

 @override
 public boolean equals(object object) {
  if (object instanceof task) {
   return object.hashcode() == hashcode() ? true : false;
  }
  return false;
 }


}

再实现一个管理延迟任务的类

import org.apache.log4j.logger;

import java.util.concurrent.delayqueue;
import java.util.concurrent.executor;
import java.util.concurrent.executors;
import java.util.concurrent.timeunit;

/**
 * <p>
 * [任务调度系统]
 * <br>
 * [后台守护线程不断的执行检测工作]
 * </p>
 *
 * @author wangguangdong
 * @version 1.0
 * @date 2015年11月23日14:19:40
 */
public class taskqueuedaemonthread {

 private static final logger log = logger.getlogger(taskqueuedaemonthread.class);

 private taskqueuedaemonthread() {
 }

 private static class lazyholder {
  private static taskqueuedaemonthread taskqueuedaemonthread = new taskqueuedaemonthread();
 }

 public static taskqueuedaemonthread getinstance() {
  return lazyholder.taskqueuedaemonthread;
 }

 executor executor = executors.newfixedthreadpool(20);
 /**
  * 守护线程
  */
 private thread daemonthread;

 /**
  * 初始化守护线程
  */
 public void init() {
  daemonthread = new thread(() -> execute());
  daemonthread.setdaemon(true);
  daemonthread.setname("task queue daemon thread");
  daemonthread.start();
 }

 private void execute() {
  system.out.println("start:" + system.currenttimemillis());
  while (true) {
   try {
    //从延迟队列中取值,如果没有对象过期则队列一直等待，
    task t1 = t.take();
    if (t1 != null) {
     //修改问题的状态
     runnable task = t1.gettask();
     if (task == null) {
      continue;
     }
     executor.execute(task);
     log.info("[at task:" + task + "] [time:" + system.currenttimemillis() + "]");
    }
   } catch (exception e) {
    e.printstacktrace();
    break;
   }
  }
 }

 /**
  * 创建一个最初为空的新 delayqueue
  */
 private delayqueue<task> t = new delayqueue<>();

 /**
  * 添加任务，
  * time 延迟时间
  * task 任务
  * 用户为问题设置延迟时间
  */
 public void put(long time, runnable task) {
  //转换成ns
  long nanotime = timeunit.nanoseconds.convert(time, timeunit.milliseconds);
  //创建一个任务
  task k = new task(nanotime, task);
  //将任务放在延迟的队列中
  t.put(k);
 }

 /**
  * 结束订单
  * @param task
  */
 public boolean endtask(task<runnable> task){
  return t.remove(task);
 }
}

使用方法

• 在容器初始化的时候调用init方法.
• 实现一个runnable接口的类，调用taskqueuedaemonthread的put方法传入进去.
• 如果需要实现动态的取消任务的话，需要task任务的类重新hashcode方法，最好用业务限制hashcode的冲突发生.

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作能带来一定的帮助，如果有疑问大家可以留言交流，谢谢大家对的支持。