Java multithreaded queue, BlockingQueue and method parsing of production consumer model using BlockingQueue

What is queue

Queue is a data structure. Except for the priority queue and LIFO queue, the queue sorts the elements in FIFO (first in, first out). No matter which sorting method is used, the head of the queue calls remove() or poll() to remove the elements. In the FIFO queue, all new elements are inserted at the end of the queue.

Methods in queue

The methods in the queue are not difficult to understand. There are 6, one for every 2 pairs, that is, a total of 3 pairs. Just look at jdkapi:

Just note that nulls are usually not allowed to be inserted into queues. Although some implementations (such as LinkedList) are allowed, they are not recommended.

BlockingQueue

1. BlockingQueue overview

BlockingQueue is also Java util. The tool under concurrent is mainly used to control thread synchronization.

BlockingQueue has four specific implementation classes. Different implementation classes are selected according to different needs

1. Arrayblockingqueue: a bounded blocking queue supported by an array. The constructor of a specified size BlockingQueue must take an int parameter to indicate its size The objects contained are sorted in FIFO (first in, first out) order.

2. Linkedblockingqueue: an indefinite BlockingQueue. If its constructor takes a specified size parameter, the generated BlockingQueue has a size limit. If it does not take a size parameter, the size of the generated BlockingQueue is determined by integer MAX_ Value The objects contained are sorted in FIFO (first in, first out) order.

3. Priorityblockingqueue: similar to linkedblockqueue, but the sorting of its objects is not FIFO, but according to the natural sorting order of objects or the order determined by the comparator of the constructor.

4. Synchronousqueue: a special BlockingQueue. Operations on it must be done alternately by putting and taking.

Linkedblockingqueue can specify the capacity or not. If it is not specified, the default maximum is integer MAX_ Value, which mainly uses the put and take methods. The put method will block when the queue is full until a queue member is consumed, and the take method will block when the queue is empty until a queue member is put in.

Let's talk about BlockingQueue, because BlockingQueue is a key point in the queue, and through BlockingQueue, we can deepen our understanding of the producer / consumer model again. Other queues are not difficult. You can fully understand their role by viewing the JDK API and simply reading the source code.

BlockingQueue, as the name suggests, blocks a queue. BlockingQueue is in Java util. Therefore, it is not difficult to understand that BlockingQueue is proposed to solve the efficient and safe data transmission in multithreading.

In multithreading, many scenarios can be implemented using queues, such as the classic producer / consumer model. Data sharing between the two can be easily realized through queues. Define a producer thread and a consumer thread, and share data through queues.

Of course, the reality cannot be ideal. For example, when consumers consume faster than producers produce, when consumers consume to a certain extent, The BlockingQueue is proposed to solve this problem. It does not need programmers to control these details, but also take into account efficiency and thread safety.

The so-called "blocking" of blocking queue refers to that in some cases, the thread will hang (i.e. blocking), and once the conditions are met, the suspended thread will wake up automatically. Using BlockingQueue, you don't need to care about when to block the thread and wake up the thread. BlockingQueue has done all these things

2. Methods in BlockingQueue

Since BlockingQueue is a sub interface of queue, there must be methods in queue, as listed above. Take a look at the unique methods in BlockingQueue:

（1）voidput(Ee)throwsInterruptedException

Add e to the BlockingQueue. If there is no space in the BlockingQueue, the calling thread will be blocked and enter the waiting state until there is space in the BlockingQueue

（2）voidtake()throwsInterruptedException

Take the first object in the BlockingQueue. If the BlockingQueue is empty, the calling thread will be blocked and enter the waiting state until new data is added to the BlockingQueue

（3）intdrainTo(Collectionc,intmaxElements)

Take the data in BlockingQueue to C at one time, and you can specify the number of fetches. This method can improve the efficiency of obtaining data without locking or releasing locks in batches

3、ArrayBlockingQueue

For array based blocking queues, the queue size must be specified. It's simple. There is only one reentrantlock object in arrayblockingqueue, which means that producers and consumers cannot run in parallel (see the code below). In addition, when creating arrayblockingqueue, you can specify whether reentrantlock is a fair lock, and non fair lock is adopted by default.

4、LinkedBlockingQueue

The blocking queue based on linked list is similar to arrayblockingqueue. However, if the linkedblockingqueue does not specify the queue capacity, it will default to a capacity similar to the infinite size. The reason is that the infinite size is integer MAX_ Value, so it's easy to understand why the arrayblockingqueue must be sized. If the arrayblockingqueue does not specify a size, use integer MAX_ Value, which will cause a lot of space waste, but the implementation based on linked list is different. It is only connected one node by one. In addition, linkedblockingqueue producers and consumers have their own locks (see the code below), which means that producers and consumers can run "at the same time".

5、SynchronousQueue

It is a special kind of waiting queue without buffer. What is called no buffer? In arrayblocking:

The array is used to store the queue. The linkedblockingqueue contains:

Connect queues as linked lists.

Producer / consumer operation data are actually operated through these two "intermediaries", However, synchronousqueue means that the producer directly gives the data to the consumer (the consumer directly takes the data from the producer), which seems to be back to the old method without the producer / consumer model. In other words, each insertion operation must wait for the removal operation corresponding to a thread. Synchronousqueue has two modes:

1. Fair model

Fair lock is adopted, and a FIFO queue is used to manage redundant producers and consumers

2. Unfair model

Non fair locks and a LIFO stack are used to manage redundant producers and consumers, which is also the default mode of synchronous queue

Implementing producer consumer model with BlockingQueue

The producer consumer model we wrote in the previous article has limitations, which are reflected in:

Only one data can be stored in the buffer. In the actual producer / consumer model, a large number of data produced by producers can be stored in the buffer. The data processing speed of producers and consumers is almost the same. OK, let's simply write an example with BlockingQueue, and let producers and consumers process data at different speeds. The subclass is arrayblockingqueue, with a size of 10:

The code is used to make the production speed of producers faster than the consumption speed of consumers. Take a look at the operation results:

The output results are divided into two parts:

1. Lines 1 to 23. This BlockingQueue is not full, so producers produce freely and consumers consume freely. Basically, three are produced and one is consumed, and consumers consume slowly

2. From lines 24 to 27, we can see from the above that when the production reaches 16 and the consumption reaches 6, it means that the limit of arrayblockingqueue is 10. At this time, there is no way. The producer will be full when producing an arrayblockingqueue, so it can't continue production. It can't continue production until the consumer has consumed it. Therefore, the printed content must be a producer and a consumer

This is what the previous chapter said at the beginning of "improving the overall data processing speed by balancing the processing capacity of producers and consumers", which should be clearly reflected in the example. In addition, don't worry about system fake death in non single producer / consumer scenarios. BlockingQueue in scenarios with empty buffer and full buffer defines different conditions, so it won't wake up its own peers.

summary

The above is all about Java multithreaded queue, BlockingQueue and using BlockingQueue to realize the method analysis of production consumer model. I hope it will be helpful to you. Interested friends can refer to this website:

On the advantages of Java multithreading and code examples

On the wonderful use of future in Java multithreading (with source code)

Java uses future to get multithreading results in time