Java blocking queue thread synchronous cooperation

At the same level as the list set, the queue interface inherits the set interface LinkedList and implements the queue interface. The Queue interface narrows the access rights to the methods of LinkedList (that is, if the parameter type in the method is Queue, then only the methods defined by the Queue interface can be accessed, but not the LinkedList directly). Non-queued methods), so that only appropriate methods can use BlockingQueue to inherit the queue interface.

A queue is a data structure. It has two basic operations: adding an element at the end of the queue and deleting an element from the head of the queue. In other words, queues manage data in a first-in-first-out manner. If you try to add an element to a full blocking queue or delete a meta line from an empty blocking queue, the thread will be blocked. Blocking queues is a very useful tool when multiple threads cooperate. Worker threads can periodically send. If the first thread group runs slower than the second, the second thread group will block while waiting for the result. If the first thread group runs faster, it will wait for the second thread group to catch up. The following table shows the operation of blocking queues in jdk.

Add adds a meta-string, and if the queue is full, it will throw an IIIegaISlabEepeplian exception.

Remove? Removes and returns the element at the head of the queue, and throws a NoSuchElementException if the queue is empty.

Elements? Returns the element at the head of the queue, and throws a NoSuchElementException if the queue is empty.

Quote? Add an element and return true? Returns false if the queue is full.

Poll removes and queries the element at the head of the queue, and returns null if the queue is empty.

Peep? Returns the element of the queue head, or null if the queue is empty.

Put added an element? If the queue is full, it is blocked.

Take removes and returns the element at the head of the queue. If the queue is empty, it will block.

Remove element offer poll peek actually belongs to the queue interface.

Operations that block queues can be divided into the following three categories according to their response patterns: aadremove and element operations will throw an exception when you try to add elements to a full queue or get elements from an empty queue. Of course, in multithreaded programs, the queue may become full or empty at any time, so you may want to use the offer poll peek method. These methods only give an error indication and will not throw an exception when the task cannot be completed.

Note that the poll and peek methods incorrectly return null, so it is illegal to insert null values into the queue.

There are also offer and poll method variants with timeouts, such as the following call.

Boolean success = q offer(x TimeUnit milliseconds);

Attempt to insert an element into the end of the queue within milliseconds. If successful, return true immediately; otherwise, return false when timeout arrives. Call in the same way.

Object head = q poll (time unit milliseconds);

If the queue header element is successfully removed within milliseconds, the header element is returned immediately, otherwise null is returned when the timeout arrives.

Finally, we have blocking operations: when the queue is full, the put and take put methods block, and when the queue is empty, the take method blocks.

The ncurrent package provides three kinds of blocking queues. There is no upper limit for the capacity of LinkedBlockingQueue by default (it is inaccurate to say that the capacity is an integer MAX_VALUE when it is not specified, but how can it be blocked when it is put? But you can also choose to specify its maximum capacity. It is a queue based on linked list. This queue sorts the elements according to FIFO.

ArrayBlockingQueue needs to specify the capacity when it is constructed, and you can choose whether fairness is needed. If the fairness parameter is set to true, the thread with the longest waiting time will be processed first (in fact, this fairness is achieved by setting ReentrantLock to true, that is, the thread with the longest waiting time will run first). Fairness usually comes at the expense of your performance and is only used when it is really needed. It is an array-based blocking circular queue. The queue sorts the elements according to FIFO (first in first out) principle.

PriorityBlockingQueue is a PriorityQueue, not a first-in first-out queue. Queue elements are deleted in priority order, and there is no upper limit. (Look at the source code. PriorityBlockingQueue is a repackaging of PriorityQueue, which is based on heap data structure, but Priority Queue has no capacity limit of Arra. YList is the same, so when put is placed on the priority blocking queue, it will not be blocked. Although the queue is logically infinite, attempting to perform an add operation may result in an OutOfMemoryError because resources are exhausted. But if the queue is empty, the operation of fetching elements will be blocked, so its retrieval operation will also be blocked. In addition, the elements entering this queue should have comparative ability.

Finally, DelayQueue (based on PriorityQueue) is an unbounded blocking queue for storing delay elements. Only when the delay expires can elements be extracted from it. The head of the queue is the delay element with the longest retention time after the delay expires. If the delay has not expired and the queue has no head, poll will return null as the getDe of the element. When the lay(TimeUnit nanosecond) method returns a value less than or equal to zero, an expired poll will appear to delete the element. Empty elements are not allowed in this queue. Below is the delay interface.

Java code

Common interface delay spread can be compared with < delay > {

Long getDelay(TimeUnit unit);

}

Common interface delay spread can be compared with < delay > {

Long getDelay(TimeUnit unit);

}

Elements placed in the DelayQueue also implement the pareTo method. DelayQueue uses this to sort elements.

The following example shows how to use blocking queue to control the thread set program to search all files in the directory and all its subdirectories, and print out a list of files containing the specified keywords. As can be seen from the following examples, using blocking queues has two obvious advantages, that is, multithreading does not need extra synchronization when processing the same queue, and the queue will automatically balance the load, that is, if the processing on the other side (production side and consumption side) is fast, it will be blocked, thus narrowing the processing speed gap between the two sides. The following is the concrete implementation.

Java code

Public class BlockingQueueTest {

Public static void main(String[] args) {

Scanner input = new scanner (system input);

System output printing (input basic directory (for example,/usr/local/JDK/src):);

The string directory = is in nextLine ();

System output printing (enter keywords (such as volatile):)；:);

String keyword = innextline ();

final int FILE _ QUEUE _ SIZE =； //Blocking queue size

final int SEARCH _ THREADS =； //Number of keyword search threads

//Blocking queue based on ArrayBlockingQueue

BlockingQueue & lt file & gtQueue = New ArrayBlockingQueue < File & gt (

File queue size);

//Start only one thread to search the directory.

File enumeration task enumerator = new file enumeration task (queue

New file (directory));

New thread (enumerator) start ();

//Start the thread to search the file for the specified keyword.

for(int I =； I < = search _ thread; i++)

New Thread(new SearchTask(queue keyword)) start ();

}

}

The FileEnumerationTask class implements Runnable {

//The virtual file object is placed in the blocking queue to indicate that the file has been traversed.

Public static file dummy = newfile ();

Private blocking queue & lt file & gt queue;

Private file startup directory;

Common file enumeration task (BlockingQueue & lt file & gt queue file startup directory) (

This queue = queue

this starting directory = starting directory；

}

Public invalid operation () {

Try {

Enumeration (starting directory);

Queue placement (virtual); //The execution here means that the files in the specified directory have been traversed.

} catch (InterruptedException e) {

}

}

//Put all files in the specified directory into the blocking queue as file objects.

The public void enumeration (file directory) threw an InterruptedException {

File[] files = directory list file ();

For (file file: file)

If (file is directory ())

Enumeration (files);

other

//Put the element at the end of the queue, and block if the queue is full.

Queue placement (file);

}

}

}

Class SearchTask implements Runnable {

Private blocking queue & lt file & gt queue;

Private string keyword;

public search task(blocking queue & lt； File & gt queue string keyword) (

This queue = queue

This keyword = keyword;

}

Public invalid operation () {

Try {

Boolean done = false

And (! Done) {

//Take out the element at the head of the queue and block it if the queue is empty.

file file = queue take()；

if(file = = file enumeration task DUMMY){

//Take it out and put it back so that other threads can read it, and it will be over soon.

Queue placement (file);

done = true

} Otherwise,

Search (files);

}

} catch (IOException e) {

e printStackTrace()；

} catch (InterruptedException e) {

}

}

The public void search (File file) threw IOException {

Scanner in = new Scanner(new file inputstream(file))；

int line number =；

While (in hasNextLine ())

line number++；

string line = in next line()；

If (ntains))

System output printf (%s:% d:% s% nFile acquisition path () line number.

Line);

}

in close()；

}

Lishi Xinzhi/Article/program/Java/hx/20 13 1 1/26657