当前位置：首页 > article >正文

【SpringBoot】调度和执行定时任务--DelayQueue (附demo)

article 2024/9/20 15:31:35

DelayQueue 是 Java 中 java.util.concurrent 包的一部分，它是一个无界阻塞队列，其中的元素只能在其到期时才能被取出。DelayQueue 是一个泛型类，要求队列中的元素实现 Delayed 接口。

1. `Delayed` 接口

要使用 DelayQueue，首先需要实现 Delayed 接口。Delayed 接口继承自 Comparable 接口，要求实现以下两个方法：

long getDelay(TimeUnit unit): 返回剩余的延迟时间。
int compareTo(Delayed o): 比较两个延迟对象的顺序。

2. 创建一个 `Delayed` 元素

下面是一个简单的实现 Delayed 接口的类：

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

public class DelayedElement implements Delayed {
    private final long delayTime;
    private final long expireTime;
    private final String message;

    public DelayedElement(long delayTime, String message) {
        this.delayTime = delayTime;
        this.expireTime = System.currentTimeMillis() + delayTime;
        this.message = message;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = expireTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (this.expireTime < ((DelayedElement) o).expireTime) {
            return -1;
        }
        if (this.expireTime > ((DelayedElement) o).expireTime) {
            return 1;
        }
        return 0;
    }

    public String getMessage() {
        return message;
    }
}

3. 使用 `DelayQueue`

创建一个 DelayQueue 并添加 Delayed 元素：

import java.util.concurrent.DelayQueue;

public class DelayQueueExample {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedElement> delayQueue = new DelayQueue<>();

        // 添加延迟元素到队列中
        delayQueue.put(new DelayedElement(5000, "Message 1")); // 5秒延迟
        delayQueue.put(new DelayedElement(10000, "Message 2")); // 10秒延迟
        delayQueue.put(new DelayedElement(15000, "Message 3")); // 15秒延迟

        // 取出元素
        while (!delayQueue.isEmpty()) {
            DelayedElement element = delayQueue.take(); // 阻塞，直到有元素到期
            System.out.println("Received: " + element.getMessage());
        }
    }
}

4. 常用方法

DelayQueue 提供了一些常用的方法：

boolean add(E e): 添加元素到队列中。
boolean offer(E e): 尝试添加元素到队列中。
void put(E e): 添加元素到队列中，如果队列已满则阻塞。
E take(): 取出并移除队列头部的元素，如果没有元素到期则阻塞。
E poll(): 取出并移除队列头部的元素，如果没有元素到期返回 null。
E peek(): 获取队列头部的元素，但不移除，如果没有元素到期返回 null。
int size(): 返回队列中元素的数量。

5. 示例：使用 `DelayQueue` 实现延迟任务调度

下面是一个使用 DelayQueue 实现延迟任务调度的示例：

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DelayQueueScheduler {
    private final DelayQueue<DelayedElement> delayQueue = new DelayQueue<>();
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public DelayQueueScheduler() {
        executor.execute(() -> {
            try {
                while (true) {
                    DelayedElement element = delayQueue.take();
                    System.out.println("Executing task: " + element.getMessage());
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }

    public void scheduleTask(long delay, String message) {
        delayQueue.put(new DelayedElement(delay, message));
    }

    public void shutdown() {
        executor.shutdown();
    }

    public static void main(String[] args) throws InterruptedException {
        DelayQueueScheduler scheduler = new DelayQueueScheduler();
        scheduler.scheduleTask(5000, "Task 1"); // 5秒后执行
        scheduler.scheduleTask(10000, "Task 2"); // 10秒后执行
        scheduler.scheduleTask(15000, "Task 3"); // 15秒后执行

        Thread.sleep(20000); // 主线程等待20秒
        scheduler.shutdown();
    }
}