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

Today, I will talk to you about how to use blocking queues to implement a concurrent container in JAVA. Many people may not know much about it. In order to make you understand better, the editor has summarized the following for you. I hope you can get something from this article.

In concurrent programming, you sometimes need to use thread-safe queues. There are two ways to implement a thread-safe queue: one is to use a blocking algorithm, and the other is to use a non-blocking algorithm. Queues using blocking algorithms can be implemented in such ways as one lock (the same lock for queuing and demarcation) or two locks (queuing and demarcation with different locks). The non-blocking implementation can be implemented using a circular CAS.

Blocking queue

The blocking queue (BlockingQueue) is a queue that supports two additional operations. These two additional operations support blocking insertion and removal methods.

Plug-in method that supports blocking: this means that when the queue is full, the queue blocks the thread that inserts the element until the queue is dissatisfied.

Support for blocking removal: this means that when the queue is empty, the thread that gets the element waits for the queue to become non-empty.

Application scenario

Blocking queues are often used in the scenarios of producers and consumers, where producers are threads that add elements to the queue, and consumers are threads that fetch elements from the queue. Blocking queues are containers that producers use to store elements and consumers use to get elements.

4 handling of insert and remove operations

Throw an exception: when the queue is full, if you insert elements into the queue again, an IllegalStateException ("Queue full") exception will be thrown. Getting an element from the queue throws a NoSuchElementException exception when the queue is empty.

Returns a special value: when an element is inserted into the queue, it returns whether the element was inserted successfully, and true is returned successfully. If it is a remove method, it takes an element from the queue and returns null if not.

Blocking: when the blocking queue is full, if the producer thread put the queue, the queue blocks the producer thread until the queue is available or exits in response to an interrupt. When the queue is empty, if the consumer thread take elements from the queue, the queue blocks the consumer thread until the queue is not empty.

Timeout exit: when the blocking queue is full, if the producer thread inserts elements into the queue, the queue will block the producer thread for a period of time, and if the specified time is exceeded, the producer thread will exit.

Note: if it is an unbounded blocking queue, the queue is unlikely to be full, so using the put or offer method will never be blocked, and when using the offer method, this method will always return true.

Blocking queue in Java

ArrayBlockingQueue: a bounded blocking queue consisting of array structures.

LinkedBlockingQueue: a bounded blocking queue consisting of linked list structures.

PriorityBlockingQueue: an unbounded blocking queue that supports prioritization.

DelayQueue: an unbounded blocking queue implemented using priority queues.

SynchronousQueue: a blocking queue that does not store elements.

LinkedTransferQueue: an unbounded blocking queue consisting of linked list structures.

LinkedBlockingDeque: a bidirectional blocking queue consisting of linked list structures.

ArrayBlockingQueue

ArrayBlockingQueue is a bounded blocking queue implemented with an array, and the length of the queue needs to be specified during initialization. This queue sorts elements on a first-in, first-out (FIFO) basis. By default, threads are not guaranteed fair access to the queue, but specify the fairness of the blocking queue when initializing the queue, such as: ArrayBlockingQueue fairQueue = new ArrayBlockingQueue. It uses ReentrantLock to achieve thread safety in queues.

Core attributes / * * The queued items * / final Object [] items; / * * items index for next take, poll, peek or remove * / int takeIndex; / * * items index for next put, offer, or add * / int putIndex; / * * Number of elements in the queue * / int count; / * * Main lock guarding all access * / final ReentrantLock lock; / * * Condition for waiting takes * / private final Condition notEmpty / * * Condition for waiting puts * / private final Condition notFull; Core method public boolean offer (E, long timeout, TimeUnit unit) throws InterruptedException {checkNotNull (e); long nanos = unit.toNanos (timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly (); try {while (count = = items.length) {if (nanos x = (ScheduledThreadPoolExecutor.ScheduledFutureTask) other) / / the row with small expiration time is at the front of the row, and the row with large expiration time is at the back. If it is the same, use sequenceNumber to sort. Long diff = time-x.time; if (diff

< 0) return -1; else if (diff >

0) return 1; else if (sequenceNumber

< x.sequenceNumber) return -1; else return 1; } // 快要过期的排在前面 long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS); return (diff < 0) ? -1 : (diff >

0)? 1: 0;} how to implement a delay blocking queue

The implementation of the delay blocking queue is simple: when the consumer gets the element from the queue, if the element does not reach the delay, the current thread is blocked.

Public E poll (long timeout, TimeUnit unit) throws InterruptedException {long nanos = unit.toNanos (timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly (); try {for (;;) {E first = q.peek (); if (first = = null) {if (nanos)

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.