How to use the delay queue DelayQueue in the Java thread pool queue

In the blocking queue, in addition to adding and deleting elements, we can delay the deletion of elements, that is, using the DelayQueue method. This article will talk to you about DelayQueue in Java thread pool queue - Delay Queue

public enum QueueTypeEnum {
    ARRAY_BLOCKING_QUEUE(1, "ArrayBlockingQueue"),
    LINKED_BLOCKING_QUEUE(2, "LinkedBlockingQueue"),
    DELAY_QUEUE(3, "DelayQueue"),
    PRIORITY_BLOCKING_QUEUE(4, "PriorityBlockingQueue"),
    SYNCHRONOUS_QUEUE(5, "SynchronousQueue"),
    LINKED_TRANSFER_QUEUE(6, "LinkedTransferQueue"),
    LINKED_BLOCKING_DEQUE(7, "LinkedBlockingDeque"),
    VARIABLE_LINKED_BLOCKING_QUEUE(8, "VariableLinkedBlockingQueue"),
    MEMORY_SAFE_LINKED_BLOCKING_QUEUE(9, "MemorySafeLinkedBlockingQueue");
}

DelayQueue Delay Queue

is similar to PriorityBlockingQueue and is an unbounded priority blocking implementation implemented by the binary heap queue. All elements are required to implement the Delayed interface, and tasks are extracted from the queue by executing the delay. Elements can only be extracted from the queue after the delay expires. The generic parameters of DelayQueue need to implement the Delayed interface. The Delayed interface inherits the Comparable interface. DelayQueue internally uses a non-thread-safe priority queue (PriorityQueue) and uses the Leader/Followers mode to minimize unnecessary waiting time. . DelayQueue is not allowed to contain null elements.

public interface Delayed extends Comparable<Delayed> {

    /**
     * 返回与此对象关联的剩余延迟（给定的时间单位）。
     * @param unit 时间单位
     * @返回剩余延迟；零值或负值表示 延迟已过期
     */
    long getDelay(TimeUnit unit);
}

DelayQueue usage scenarios

Cache system design: You can use DelayQueue to save the validity period of cache elements, and use a thread to query DelayQueue in a loop. Once the element can be obtained from DelayQueue , it means that the cache validity period has expired.

Timing task scheduling: Use DelayQueue to save the tasks and execution time that will be executed on the day. Once the task is obtained from DelayQueue, it will be executed. For example, TimerQueue is implemented using DelayQueue.

DelayQueue attribute

//可重入同步锁
private final transient ReentrantLock lock = new ReentrantLock();

//DelayQueue的实现依赖于PriorityQueue(优先队列)
private final PriorityQueue<E> q = new PriorityQueue<E>();

//第一个等待某个延时对象的线程，在延时对象还没有到期时其他线程看到这个leader不为null，那么就直接wait
//主要是为了避免大量线程在同一时间点唤醒，导致大量的竞争，反而影响性能
private Thread leader = null;

//条件队列，用于wait线程
private final Condition available = lock.newCondition();

DelayQueue construction method

//从上面属性就可以看出，DelayQueue采用了饿汉模式，调用构造方法即创建了队列实例
public DelayQueue() {}

/**
 * 创建一个DelayQueue，最初包含给定的Collection实例集合。
 * @param c 最初包含的元素集合
 */
public DelayQueue(Collection<? extends E> c) {
    this.addAll(c);
}

Implementation of Delayed interface usage example

class MyDelay<T> implements Delayed {

    long delayTime; // 延迟时间
    long expire; // 过期时间
    T data;

    public MyDelay(long delayTime, T t) {
        this.delayTime = delayTime;
        // 过期时间 = 当前时间 + 延迟时间
        this.expire = System.currentTimeMillis() + delayTime;
        data = t;
    }

    /**
     * 剩余时间 = 到期时间 - 当前时间
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    /**
     * 优先级规则：两个任务比较，时间短的优先执行
     */
    @Override
    public int compareTo(Delayed o) {
        long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        return (int) f;
    }


    @Override
    public String toString() {
        return "delayTime=" + delayTime +
                ", expire=" + expire +
                ", data=" + data;
    }
}

public class DelayQueueDemo {
    static BlockingQueue<Delayed> queue = new DelayQueue();
    public static void main(String[] args) throws InterruptedException {
        queue.add(new MyDelay(8, "第一次添加任务"));
        queue.add(new MyDelay(3, "第二次添加任务"));
        queue.add(new MyDelay(5, "第三次添加任务"));

        while (!queue.isEmpty()) {
            Delayed delayed = queue.take();
            System.out.println(delayed);
        }
    }
}

DelayQueue summary

DelayQueue actually uses

Decorator mode, adds delay time acquisition element function under the packaging of PriorityQueue. Its main features are summarized as follows:

• DelayQueue It is an unbounded blocking queue. PriorityQueue is used inside the queue to implement

• The elements entering the queue must implement the Delayed interface. When creating an element, you can specify how long it takes to get the current element from the queue. Only when Elements can only be extracted from it when the delay expires

• The head of the queue is the Delayed element that has the longest storage time after the delay expires

• If not Delay unexpired elements, and the queue has no head, and poll will return null

• When the getDelay(TimeUnit.NANOSECONDS) method of an element returns a value less than or equal to 0, it means that the The element has expired

• Cannot use poll or take to remove unexpired elements, nor will these elements be treated as normal elements; for example: the size method returns expired and unexpired The sum of the counts of elements

• This queue does not allow null elements

The above is the detailed content of How to use the delay queue DelayQueue in the Java thread pool queue. For more information, please follow other related articles on the PHP Chinese website!