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

What are the commonly used BlockingQueue? I believe many inexperienced people are at a loss about this. Therefore, this paper summarizes the causes and solutions of the problem. Through this article, I hope you can solve this problem.

Why use blocking queues

Earlier, I introduced the meaning of the parameters of ThreadPoolExecutor (the thread pool of concurrent programming ThreadPoolExecutor), where there is a BlockingQueue, which is a blocking queue. So, have you guys ever wondered why the thread pool here uses a blocking queue?

We know the queue is first-in, first-out. When an element is placed, it is placed at the end of the queue, and when the element is taken out, it is taken from the head of the queue. So what to do when the queue is empty or full.

At this point, the blocking queue will automatically help us deal with this situation.

When the blocking queue is empty, the operation of fetching elements from the queue is blocked. When the blocking queue is full, the operation of putting elements into the queue is blocked.

Then, once the empty queue has data, or there is free space in the full queue, the blocked thread will be automatically awakened.

This is the advantage of blocking queues. You don't need to care about when the thread is blocked or when the thread is awakened. Everything is done automatically by the blocking queue. We only need to focus on the specific business logic.

This blocking queue is often used in the producer-consumer model. (see: the interviewer asked me to write a producer-consumer model by hand.)

Commonly used blocking queues

So, what are the blocking queues we usually use? Next, through the idea class diagram, list the commonly used blocking queues, and then explain one by one (do not know how to use, you can refer to this article: how to use IDEA to quickly view class diagram relationships).

In the blocking queue, all commonly used methods are defined in the BlockingQueue interface. Such as

The method to insert the element is put,offer,add. The method to remove an element is remove,poll,take.

They can be handled in four different ways, the first is to throw an exception on failure, the second is to return a special value on failure, the third is to block the current thread all the time, and the last is to block within a specified time, otherwise a special value is returned. (the above special values mean that failure returns false when inserting an element, and null when fetching an element)

Throw exception special value blocking timeout insert add (e) offer (e) put (e) offer (eMagneTime unit) remove remove () poll () take () poll (time,unit)

1) ArrayBlockingQueue

This is a bounded blocking queue consisting of array structures. First of all, take a look at its construction methods, there are three.

The first can specify the size of the queue, and the second can also specify whether the queue is fair or not. If not, the default is unfair. It is implemented using ReentrantLock's fair and unfair locks (more on AQS later).

The simple understanding is that a sequential waiting queue is maintained within ReentrantLock, and if five tasks come together, they are all blocked. If it is fair, the longest waiting task in the waiting queue enters the blocking queue first. If it is unfair, then the five threads need to grab the lock, and whoever gets it first enters the blocking queue.

The third construction method is to initialize the elements of a collection to the blocking queue.

In addition, ArrayBlockingQueue does not achieve read-write separation, that is, reading and writing cannot be done at the same time. Because it reads and writes with the same lock, as shown in the following figure:

File

2) LinkedBlockingQueue

This is a bounded blocking queue consisting of linked list structures. It can be constructed in three ways.

You can see that the construction method is much the same as that of ArrayBlockingQueue, except that LinkedBlockingQueue can not specify the size of the queue, and the default value is Integer.MAX_VALUE.

However, it is best not to do this, and it is recommended that you specify a fixed size. Because, if the speed of the producer is much faster than that of the consumer, this will cause the blocking queue to swell until the system memory is exhausted (at this point, the maximum queue capacity has not been reached).

In addition, LinkedBlockingQueue achieves the separation of read and write, so that data reading and writing do not affect each other, which is undoubtedly a great improvement in efficiency in high concurrency scenarios.

3) SynchronousQueue

This is an unbounded queue without buffers. What do you mean, take a look at its size method:

Always returns 0 because it is a queue with no capacity.

When you perform an insert element operation, you must wait for a fetch operation. That is, when you put the element, you must wait for the take operation.

So, some students are curious, there is no capacity, what is called a queue, what is the significance of this?

My understanding is that this applies to a series of operations where concurrent tasks are small and the speeds of producers and consumers are not much different, directly connecting producers and consumers without going through the queue. Therefore, it will be more efficient.

This queue has two constructors that are used to pass in whether it is fair or unfair, and the default is unfair.

4) PriorityBlockingQueue

This is an unbounded queue that supports prioritization. There are four construction methods:

You can specify the initial capacity size (note that the initial capacity does not represent the maximum capacity), or if not, the default size is 11. You can also pass in a comparator to sort the elements according to certain rules. If you do not specify a comparator, the default is the natural order.

PriorityBlockingQueue is implemented based on the minimum heap of the binary tree, and whenever an element is fetched, the element with the highest priority is taken out. Let's test it:

Public class Person {

Private int id

Private String name

Public int getId () {

Return id

}

Public void setId (int id) {

This.id = id

}

Public String getName () {

Return name

}

Public void setName (String name) {

This.name = name

}

@ Override

Public String toString () {

Return "Person {" +

"id=" + id +

", name='" + name +'\'+

'}'

}

Public Person (int id, String name) {

This.id = id

This.name = name

}

Public Person () {

}

}

Public class QueueTest {

Public static void main (String [] args) throws InterruptedException {

PriorityBlockingQueue priorityBlockingQueue = new PriorityBlockingQueue (1, new Comparator () {

@ Override

Public int compare (Person o1, Person O2) {

Return o1.getId ()-o2.getId ()

}

});

Person p2 = new Person (7, "Li Si")

Person p1 = new Person (9, "Zhang San")

Person p3 = new Person (6, Wang Wu)

Person p4 = new Person (2, Zhao Liu)

PriorityBlockingQueue.add (p1)

PriorityBlockingQueue.add (p2)

PriorityBlockingQueue.add (p3)

PriorityBlockingQueue.add (p4)

/ / because of the minimum heap implementation of the binary tree, it is not possible to print elements directly in this way.

System.out.println (priorityBlockingQueue)

System.out.println (priorityBlockingQueue.take ())

System.out.println (priorityBlockingQueue)

System.out.println (priorityBlockingQueue.take ())

System.out.println (priorityBlockingQueue)

}

}

Print the results:

[Person {id=2, name=' Zhao Liu'}, Person {id=6, name=' Wang Wu'}, Person {id=7, name=' Li Si'}, Person {id=9, name=' Zhang San'}]

Person {id=2, name=' Zhao Liu'}

[Person {id=6, name=' Wang Wu'}, Person {id=9, name=' Zhang San'}, Person {id=7, name=' Li Si'}]

Person {id=6, name=' Wang Wu'}

[Person {id=7, name=' Li Si'}, Person {id=9, name=' Zhang San'}]

As you can see, the minimum value of id is 2 for the first time and 6 for the second time.

5) DelayQueue

This is an unbounded blocking queue with delay time. The elements in the queue can only be taken out when the delay time is up. This queue is generally used for the deletion of expired data or for task scheduling. Below, simulate the deletion of data for a fixed period of time.

First define the data elements, you need to implement the Delayed interface, implement the getDelay method for calculating the remaining time, and the CompareTo method for prioritization.

Public class DelayData implements Delayed {

Private int id

Private String name

/ / data expiration time

Private long endTime

Private TimeUnit timeUnit = TimeUnit.MILLISECONDS

Public int getId () {

Return id

}

Public void setId (int id) {

This.id = id

}

Public String getName () {

Return name

}

Public void setName (String name) {

This.name = name

}

Public long getEndTime () {

Return endTime

}

Public void setEndTime (long endTime) {

This.endTime = endTime

}

Public DelayData (int id, String name, long endTime) {

This.id = id

This.name = name

/ / you need to add the incoming time endTime plus the current system time as the expiration time of the data.

This.endTime = endTime + System.currentTimeMillis ()

}

Public DelayData () {

}

@ Override

Public long getDelay (TimeUnit unit) {

Return this.endTime-System.currentTimeMillis ()

}

@ Override

Public int compareTo (Delayed o) {

Return o.getDelay (this.timeUnit)-this.getDelay (this.timeUnit) < 0? 1:-1

}

}

Simulate three pieces of data and set different expiration times:

Public class ProcessData {

Public static void main (String [] args) throws InterruptedException {

DelayQueue delayQueue = new DelayQueue ()

DelayData a = new DelayData (5, "A", 5000)

DelayData b = new DelayData (8, "B", 8000)

DelayData c = new DelayData (2, "C", 2000)

DelayQueue.add (a)

DelayQueue.add (b)

DelayQueue.add (c)

System.out.println ("start timing:" + System.currentTimeMillis ())

For (int I = 0; I < 3; iTunes +) {

DelayData data = delayQueue.take ()

System.out.println ("id:" + data.getId () + ", data:" + data.getName () + "is removed, current time:" + System.currentTimeMillis ())

}

}

}

The final result:

Start time: 1583333583216

Id:2, data: C has been removed, current time: 1583333585216

Id:5, data: a has been removed, current time: 1583333588216

Id:8, data: B is removed, current time: 1583333591216

As you can see, the data is removed sequentially according to the expiration time. The time of C is at least 2 seconds, then An expires after 3 seconds, and B expires after 3 seconds.

After reading the above, have you mastered the common methods of BlockingQueue? If you want to learn more skills or want to know more about it, you are welcome to follow the industry information channel, thank you for reading!

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.