JDK自带的延迟队列——DelayQueue

发布时间:2024年01月16日

DelayQueue是基于阻塞队列实现的,是JDK自带延迟队列。这种方案使用成本最低,而且不依赖任何第三方服务,减少了网络交互。但缺点也很明显,就是需要占用JVM内存,在数据量非常大的情况下可能会有问题。如果数据量非常大,DelayQueue不能满足业务需求,也可以替换为其它延迟队列方式,例如Redisson、MQ等。

1. DelayQueue的原理

首先来看一下DelayQueue的源码:

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>?
    implements BlockingQueue<E> {?
?
    private final transient ReentrantLock lock = new ReentrantLock();?
    private final PriorityQueue<E> q = new PriorityQueue<E>();?
    ?
    // ... 略?
}

可以看到DelayQueue实现了BlockingQueue接口,是一个阻塞队列。队列就是容器,用来存储东西的。DelayQueue叫做延迟队列,其中存储的就是延迟执行的任务

我们可以看到DelayQueue的泛型定义:

DelayQueue<E extends Delayed>

这说明存入 DelayQueue 内部的元素必须是 Delayed 类型,这其实就是一个延迟任务的规范接口。来看一下:

public interface Delayed extends Comparable<Delayed> {

    /**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     *
     * @param unit the time unit
     * @return the remaining delay; zero or negative values indicate
     * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

从源码中可以看出,Delayed类型必须具备两个方法:

  • getDelay() : 获取延迟任务的剩余延迟时间
  • compareTo( T t ) : 比较两个延迟任务的延迟时间,判断执行顺序

可见,Delayed类型的延迟任务具备两个功能:获取剩余延迟时间、比较执行顺序。当然,我们可以对Delayed做实现和功能扩展,比如添加延迟任务的数据。

将来每一次提交数据,就可以将数据保存在这样的一个 Delayed 类型的延迟任务里并设定延迟时间。然后交给 DelayQueue 队列。 DelayQueue 会调用compareTo方法,根据剩余延迟时间对任务排序。剩余延迟时间越短的越靠近队首,这样就会被优先执行。

2. DelayQueue 的基本使用

2.1 创建 Delayed 的实现类

import lombok.Data;

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

@Data
public class DelayedTask implements Delayed {
    // 消息体
    private Object data;
    // 过期时间
    private Long deadTime;

    /**
     * 到期时间,毫秒
     */
    public DelayedTask(Object data, Long expireTime) {
        this.data = data;
        this.deadTime = System.currentTimeMillis() + expireTime * 1000;
    }

    /**
     * 获取元素的剩余的有效期
     */
    @Override
    public long getDelay(TimeUnit unit) {
        long remain = this.deadTime - System.currentTimeMillis();
        if (remain < 0) {
            remain = 0;
        }
        return unit.convert(remain, TimeUnit.MILLISECONDS);
    }

    /**
     * 比较元素的剩余,看谁先到期
     */
    @Override
    public int compareTo(Delayed o) {
        long l = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        if (l > 0) {
            return 1;
        } else if (l < 0) {
            return -1;
        } else {
            return 0;
        }
    }
}

2.2 向列队发送消息并接收

import com.example.delayedqueuedemo.task.DelayedTask;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.DelayQueue;

@Slf4j
public class DelayedQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1.初始化延迟队列
        DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
        DelayedTask task1 = new DelayedTask("data 5 seconds", 5L);
        DelayedTask task2 = new DelayedTask("data 1 seconds", 1L);
        DelayedTask task3 = new DelayedTask("data 3 seconds", 3L);
        // 2.向队列中添加延迟执行的任务
        delayQueue.put(task1);
        delayQueue.put(task2);
        delayQueue.put(task3);
        // 3.尝试执行任务
        log.info("消息发送完成");
        // 如果队列中没有到期的元素,take会阻塞等待
        while (true) {
            DelayedTask task = delayQueue.take();
            log.info("获取延迟任务:task:{}", task.getData());
        }
    }
}

3. 实际项目应用

3.1 将发送端和接收端分离

在实际项目中,需要将发送端和接收端分离,避免死循环的队列接收阻塞主程序运行,故将列队接收代码抽离到 Spring 在生命周期回调期间执行,具体代码如下:

import com.example.delayedqueuedemo.task.DelayedTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class DelayedService {
    private ThreadPoolExecutor takeTaskPool = new ThreadPoolExecutor(1, 1, 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
    private ThreadPoolExecutor executeTaskPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors()), new ThreadPoolExecutor.CallerRunsPolicy());
    private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();

    /**
     * 生命周期回调
     */
    @PostConstruct
    public void postConstruct() {
        // 开启线程池,避免死循环阻塞主程序
        takeTaskPool.execute(new Runnable() {
            @Override
            public void run() {
                // 如果队列中没有到期的元素,take会阻塞等待
                while (true) {
                    try {
                        DelayedTask task = delayQueue.take();
                        // 为了防止业务逻辑执行太长,阻塞消息的读取
                        executeTaskPool.execute(new Runnable() {
                            @Override
                            public void run() {
                                // 执行业务代码
                                log.info("获取延迟任务:task:{}", task.getData());
                            }
                        });
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }

    public void delayed(String data, Long expireTime) {
        // 1.初始化延迟队列
        DelayedTask task = new DelayedTask(data, expireTime);
        // 2.向队列中添加延迟执行的任务
        delayQueue.put(task);
        // 3.尝试执行任务
        log.info("消息发送完成");
    }
}

3.2 开启事务

若需要开启事务支持,直接用 @Transactional 注解会出现事务失效的情况(@Transactional 失效场景可参考该文章: 一口气怼完12种@Transactional的失效场景 - 知乎?),故有两种方法解决:

  1. 将业务代码写到新的 service 服务中,让 Spring 代理;
  2. 编程式事务处理

以下代码以编程式事务处理方案演示如何开启事务:

import com.example.delayedqueuedemo.task.DelayedTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import javax.annotation.PostConstruct;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class DelayedService {
    private ThreadPoolExecutor takeTaskPool = new ThreadPoolExecutor(1, 1, 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
    private ThreadPoolExecutor executeTaskPool = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 120, TimeUnit.SECONDS, new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors()), new ThreadPoolExecutor.CallerRunsPolicy());
    private DelayQueue<DelayedTask> delayQueue = new DelayQueue<>();
    @Autowired
    private PlatformTransactionManager transactionManager;

    /**
     * 生命周期回调
     */
    @PostConstruct
    public void postConstruct() {
        // 开启线程池,避免死循环阻塞主程序
        takeTaskPool.execute(new Runnable() {
            @Override
            public void run() {
                // 如果队列中没有到期的元素,take会阻塞等待
                while (true) {
                    try {
                        DelayedTask task = delayQueue.take();
                        // 开启线程池,避免业务执行阻塞队列获取消息,核心线程数根据电脑cpu核数而定
                        executeTaskPool.execute(new Runnable() {
                            @Override
                            public void run() {
                                // 开启事务
                                TransactionDefinition definition = new DefaultTransactionDefinition();
                                TransactionStatus transaction = transactionManager.getTransaction(definition);
                                try {
                                    // 业务逻辑
                                    log.info("获取延迟任务:task:{}", task.getData());
                                    // 提交事务
                                    transactionManager.commit(transaction);
                                } catch (TransactionException e) {
                                    // 回滚事务
                                    transactionManager.rollback(transaction);
                                }
                            }
                        });
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }

    @Transactional
    public void delayed(String data, Long expireTime) {
        // 1.初始化延迟队列
        DelayedTask task = new DelayedTask(data, expireTime);
        // 2.向队列中添加延迟执行的任务
        delayQueue.put(task);
        // 3.尝试执行任务
        log.info("消息发送完成");
    }
}

文章来源:https://blog.csdn.net/kersixy/article/details/135562005
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。