java延时队列DelayQueue

Posted 2021-06-11 沟渠映明月

对于一些活动相关的业务场景，如果设置了到时间开启，可以通过自己实现延时队列的方式来完成类似业务。如添加完活动用活动开启时间减去当前时间，去创建延时任务，再将其添加到延时队列中，当到时间后，可以通过spring的发布订阅，做一个异步活动状态修改，以此来完成到时间活动自动开启关闭的功能，当然这种方式只适用于单体架构，如果是集群或分布式，需要自行加分布式锁，更好的方式是利用redis或其他中间件如mq完成

实体类

public class DelayTaskEntity {
    /** 1活动开启 2活动关闭 */
    private Integer type;


    private String data;


    public DelayTaskEntity(Integer type, String data) {
        this.type = type;
        this.data = data;
    }


    public Integer getType() {
        return type;
    }


    public void setType(Integer type) {
        this.type = type;
    }


    public String getData() {
        return data;
    }


    public void setData(String data) {
        this.data = data;
    }


    @Override
    public int hashCode() {
        return super.hashCode();
    }


    @Override
    public boolean equals(Object obj) {
        if(obj instanceof DelayTaskEntity){
            DelayTaskEntity entity = (DelayTaskEntity)obj;
            return this.getData().equals(entity.getData()) && this.getType().equals(entity.getType());
        }


        return false;
    }
}

延时队列

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


/**
* @author wjc
*/
public class DelayTask implements Delayed {


    private long time;
    private DelayTaskEntity param;


    public DelayTask(DelayTaskEntity param, long time) {
        this.param = param;
        this.time = time;
    }


    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(time,TimeUnit.NANOSECONDS) - unit.convert(System.currentTimeMillis(),TimeUnit.NANOSECONDS);
    }


    @Override
    public int compareTo(Delayed o) {
        DelayTask item = (DelayTask) o;
        long diff = this.time - item.time;
        if (diff <= 0) {
            return -1;
        }else {
            return 1;
        }
    }


    public long getTime() {
        return time;
    }


    public void setTime(long time) {
        this.time = time;
    }


    public DelayTaskEntity getParam() {
        return param;
    }


    public void setParam(DelayTaskEntity param) {
        this.param = param;
    }


    @Override
    public boolean equals(Object obj) {
        if(obj instanceof DelayTask){
            DelayTask task = (DelayTask) obj;
            return this.getParam().equals(task.getParam());
        }
        return false;
    }


    @Override
    public int hashCode() {
        return super.hashCode();
    }
}

工具类

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


import javax.annotation.PostConstruct;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executors;


/**
* @author wjc
*/
@Component
public class DelayQueueUtil {


    @Autowired
    private DelayQueueService service;


    public static DelayQueue<DelayTask> queue = new DelayQueue<>();


    public static void add(DelayTask delayTask){
        queue.add(delayTask);
    }


    public static void remove(DelayTask delayTask){
        queue.remove(delayTask);
    }


    @PostConstruct
    public void init(){
        Executors.newFixedThreadPool(1).execute(new Thread(this::run));
    }


    private void run(){
        while (true){
            try{
                DelayTask task = queue.take();
                service.work(task.getParam());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }


}