欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

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

程序员文章站 2024-02-29 14:29:22
一、了解delayqueue delayqueue是什么? delayqueue是一个*的blockingqueue,用于放置实现了delayed接口的对象,其中的对...

一、了解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的冲突发生.

总结

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