Shulou(Shulou.com)06/01 Report--

This article introduces the knowledge of "how to use the DelayQueue of JUC". In the operation of actual cases, many people will encounter such a dilemma, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

Delay queue DelayQueue is used to store elements with expiration attributes, and elements added to DelayQueue will not be queued until the expiration time is reached, which is often used to delay task scheduling. DelayQueue is essentially an unbounded blocking queue, the underlying layer relies on the priority queue PriorityQueue as the storage structure, and uses ReentrantLock locks to ensure thread safety.

The fields of DelayQueue are defined as follows:

Public class DelayQueue extends AbstractQueue implements BlockingQueue {/ * * reentrant exclusive lock that ensures thread safety * / private final transient ReentrantLock lock = new ReentrantLock (); / * * underlying storage structure, priority queue * / private final PriorityQueue q = new PriorityQueue (); / * Leader-Follower mode, which is used to record thread objects with role leader * / private Thread leader = null / * * Threads blocked because the queue is empty or the element has not expired * / private final Condition available = lock.newCondition (); / /. Omit method definition}

Focus on the intent of setting the DelayQueue#leader field, which is used to record thread objects whose current role is leader. When performing queue operations (DelayQueue#take and DelayQueue#poll (long, TimeUnit) methods), if the DelayQueue#leader field is null, that is, there is no leader thread, and there is an unexpired delay element, the current thread will be set to the leader role and wait for the element to expire to reduce unnecessary waiting time and ensure that the delay element can be responded in a timely manner when it expires.

Imagine that if it is not designed in this way, what should the thread do when it encounters a delay element that does not expire in the queue? There are three strategies that can be adopted:

Enter busy cycle polling.

Wait in the conditional queue and be awakened later by other threads.

Exit the current method first and wait for the queue operation to be performed again.

As you can see, these policies either consume CPU resources or fail to get elements in the queue out of the queue in time when they expire, and the introduction of leader roles can avoid these problems. When a thread waits as leader for the highest priority delay element to expire, other threads wait indefinitely as follower roles when they find that there are no expired elements in the queue. When the leader thread exits from the waiting state, it actively abandons its leader role and wakes up a waiting follower thread that will have a chance to be promoted to the new leader.

While a leader thread is waiting, it is possible to insert a higher priority element, which needs to be stripped of the thread's leader role to give other threads a chance to become leader, thus ensuring that the newly inserted element will be responded in a timely manner when it expires. Otherwise, you will not have a chance to get out of the queue until the current leader thread exits the waiting state or when a new thread requests to get the element, which may have a large delay.

Core method implementation

DelayQueue is implemented from the BlockingQueue interface. The following analyzes the implementation of the core methods one by one. But before we start the analysis, let's take a look at the java.util.concurrent.Delayed interface, which DelayQueue requires that elements added to it must implement. The Delayed API is defined as follows:

Public interface Delayed extends Comparable {long getDelay (TimeUnit unit);}

This interface inherits from the Comparable interface, so the elements added to the DelayQueue are comparable. The method Delayed#getDelay receives a unit value of type TimeUnit that returns the remaining expiration time of the current delay element. If it is less than or equal to 0, the element has expired.

Add elements: offer & add & put

For the operation of adding elements, DelayQueue implements the DelayQueue#offer, DelayQueue#add, and DelayQueue#put methods, but the latter two call the DelayQueue#offer method directly.

In addition, the timeout version of the method DelayQueue#offer (E, long, TimeUnit) is also directly delegated to the DelayQueue#offer method for execution, and does not really implement the timeout waiting mechanism. This is mainly because DelayQueue is unbounded and all additions can be responded immediately without blocking.

Let's analyze the implementation of the DelayQueue#offer method as follows:

Public boolean offer (E e) {final ReentrantLock lock = this.lock; / / locked lock.lock (); try {/ / add element q.offer (e) to the queue / / the element currently added is if (q.peek () = = e) {/ / clear leader, which is the first to expire in the queue, to ensure that the element can leave the queue in time leader = null; / / wake up the thread waiting because the element is expired or the queue is empty;} return true } finally {/ / release lock lock.unlock ();}}

DelayQueue does not allow you to add elements with a value of null, which is mainly guaranteed by the priority queue PriorityQueue. If the value of the element to be added is legal, execute:

Lock to ensure that only one thread is in the operation queue at a time

Insert the element to be added into the DelayQueue

Check whether the highest priority element in the DelayQueue is the newly added element

If so, deprive the current leader thread of its leader role and wake up a thread that had previously waited because none of the elements in the queue had expired or the queue was empty

Release the lock and return.

Why do you need to deprive the current leader thread of its leader role and wake up a waiting follower thread when the highest priority element is inserted into the queue?

In fact, as mentioned earlier, if the element that expires first in DelayQueue now has 10 seconds to expire, the leader thread will wait 10 seconds and then try to get out of the queue again. Other follower threads will wait when they find that there are no expired elements because they detect that a thread has become leader. Suppose there is now an element with 5 seconds to expire, and if there is no new thread to request out of the queue after the element expires, the element will not be responded in time until the leader thread exits and waits, even if there are a considerable number of follower threads waiting for the element to expire.

Get element: poll & peek & take

DelayQueue implements DelayQueue#poll, DelayQueue#peek, and DelayQueue#take methods for operations that fetch elements. The DelayQueue#peek method calls the PriorityQueue#peek method directly on the basis of acquiring the lock, and only gets the element that expires first in the DelayQueue (which may not expire at the time of acquisition), but does not remove this element, which is relatively simple in implementation.

The difference between the method DelayQueue#take and DelayQueue#poll is that when the queue is empty or there is no expired element, the method blocks indefinitely until an element expires or the thread is interrupted, while the DelayQueue#poll method immediately returns null in the same scenario.

Let's analyze the implementation of these two methods. First, let's take a look at the DelayQueue#poll method. The implementation is as follows:

Public E poll () {final ReentrantLock lock = this.lock; / / acquire lock lock.lock (); try {/ / get the element E first = q.peek () that expires first in the queue; / / if the queue is empty or the element has not yet expired, immediately return null if (first = = null | | first.getDelay (NANOSECONDS) > 0) {return null } / / the current element has expired. Remove and return else {return q.poll ();}} finally {/ / release lock lock.unlock ();}}

The above method checks to see if there is an expired element in the DelayQueue, and if so, takes the element out of the queue and returns, otherwise, if the queue is empty or there are no expired elements, it immediately returns null. For the DelayQueue#poll method, DelayQueue also provides a timeout version of DelayQueue#poll (long, TimeUnit) that waits for a specified time when the queue is empty or there are no expired elements.

Let's take a look at the implementation of the DelayQueue#take method, as follows:

Public E take () throws InterruptedException {final ReentrantLock lock = this.lock; / / acquire lock, support response interrupt lock.lockInterruptibly (); try {for (;;) {/ / get the element E first = q.peek () that expires first / / if the queue is empty, wait for if (first = = null) {available.await ();} / queue is not empty else {/ / if the current element has expired, then dequeue long delay = first.getDelay (NANOSECONDS); if (delay

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.