ArrayBlockingQueue for concurrent packet blocking queue

##jdk1.7.0_79

## In the previous section, a brief analysis was made on the entry and exit of the non-blocking queue

ConcurrentLinkedQueue in the concurrent package. This article will do a brief analysis of the blocking queue in the concurrent package. A brief analysis.

## Java

The total number of blocking queues in the concurrent package is

7, of course they are all thread-safe.

# ArrayBlockingQueue

: A bounded blocking consisting of an array structure queue.

LinkedBlockingQueue

: A bounded blocking queue composed of a linked list structure.

PriorityBlockingQueue

: An unbounded blocking queue that supports priority sorting.

DealyQueue

: An unbounded blocking queue implemented using a priority queue.

SynchronousQueue

: A blocking queue that does not store elements.

LinkedTransferQueue

: An unbounded blocking queue composed of a linked list structure.

LinkedBlockingDeque

: A two-way blocking queue composed of a linked list structure. (Excerpted from "

JavaThe Art of Concurrent Programming") ## In this article

make a brief analysis of

ArrayBlockingQueue

blocking queue For

ArrayLinkedQueue

, looking at the past, its security is guaranteed by

ReentrantLock, related# For the analysis of ##ReentrantLock, please refer to "5.LockInterface and its ImplementationReentrantLock", which I will also mention appropriately below. #

First let’s look at its constructor:

#Construct a bounded queue of the specified size, designated as fair or unfair lock#public

##Construction method

#public

ArrayBlockingQueue( int capacity)

##Construct a bounded queue of specified size

##public

ArrayBlockingQueue

(int capacity, boolean fair )

ArrayBlockingQueue

(

int capacity, boolean fair, Collection c)

##Construct a bounded queue of the specified size, specify it as a fair or unfair lock, and specify to join a set during initialization

 1 public ArrayBlockingQueue(int capacity) { 
 2 　　this(capacity, false);//默认构造非公平锁的阻塞队列  3 } 
 4 public ArrayBlockingQueue(int capacity, boolean fair) { 
 5 　　if (capacity <= 0)  
 6 　　　　throw new IllegalArgumentException(); 
 7 　　this.items = new Object[capacity]; 
 8 　　lock = new ReentrantLock(fair);//初始化ReentrantLock重入锁，出队入队拥有这同一个锁  9 　　notEmpty = lock.newCondition;//初始化非空等待队列，有关Condition可参考《6.类似Object监视器方法的Condition接口》10 　　notFull = lock.newCondition;//初始化非满等待队列 11 } 
12 public ArrayBlockingQueue(int capacity, boolean fair, Collecation<? extends E> c) { 
13 　　this(capacity, fair); 
14 　　final ReentrantLock lock = this.lock; 
15 　　lock.lock();//注意在这个地方需要获得锁，这为什么需要获取锁的操作呢？ 16 　　try { 
17 　　　　int i = 0; 
18 　　　　try { 
19 　　　　　　for (E e : c) { 
20 　　　　　　　　checkNotNull(e); 
21 　　　　　　　　item[i++] = e;//将集合添加进数组构成的队列中 22 　　　　　　} 
23 　　　　} catch (ArrayIndexOutOfBoundsException ex) { 
24 　　　　　　throw new IllegalArgumentException(); 
25 　　　　} 
26 　　　　count = i;//队列中的实际数据数量 27 　　　　putIndex = (i == capacity) ? 0 : i; 
28 　　} finally { 
29 　　　　lock.unlock(); 
30 　　} 
31 }

In line 15, there is a comment in the source code: Lock only for visibility, not mutual exclusion. The meaning of this sentence is to give that the operation of this lock is not for mutual exclusion operations, but to ensure its visibility. Thread T1 is the instantiated ArrayBlockingQueue object, and T2 is the enqueuing operation for the instantiated ArrayBlockingQueue object. (Of course, the execution order of T1 and T2 must be ensured), if it is not locked (locking will ensure its visibility, that is, writing it back to the main memory), ## The collection c of #T1 may only exist in the cache maintained by the T1 thread, and is not written back to the main memory, ## instantiated in T2 #ArrayBlockingQueueThere is no collection c in the maintained cache and main memory. At this time, data inconsistency is caused by visibility, causing thread safety issues .

## The following are some dequeue and inqueue operations of

ArrayBlockingQueue

.

#QueueElement Insert

##put(e)//When the queue is not full, insert directly without Return value; when the queue is full, it will block and wait until the queue is not full before inserting.

#Throw exception

##Return value (non-blocking)

Return value within a certain period

Return value (blocking)

##INSERT

##add(e)//When the queue is not full, return true; if the queue is full, an

IllegalStateException("Queue full") exception will be thrown——AbstractQueue

# #offer(e)//When the queue is not full, return true; when the queue is full, return false. Non-blocking returns immediately.

##offer(e, time, unit)//Set the waiting Time. If data cannot be inserted into the queue within the specified time, it returns false. If the insertion is successful, it returns true.

//ArrayBlockingQueue#add public boolean add(E e) { 
　　return super.add(e); 
}

//AbstractQueue#add，这是一个模板方法，只定义add入队算法骨架，成功时返回true，失败时抛出IllegalStateException异常，具体offer实现交给子类实现。 public boolean add(E e) { 
　　if (offer(e))//offer方法由Queue接口定义 　　　　return true; 
　　else 　　　　throw new IllegalStateException(); 
}

//ArrayBlockingQueue#offer，队列未满时返回true，满时返回false public boolean offer(E e) { 
　　checkNotNull(e);//检查入队元素是否为空 　　final ReentrantLock lock = this.lock; 
　　lock.lock();//获得锁，线程安全 　　try { 
　　　　if (count == items.length)//队列满时，不阻塞等待，直接返回false 　　　　　　return false; 
　　　　else { 
　　　　　　insert(e);//队列未满，直接插入 　　　　　　return true; 
　　　　} 
　　} finally {
　　　　lock.unlock();
　　}
}

//ArrayBlockingQueue#insert private void insert(E e) { 
　　items[putIndex] = x; 
　　putIndex = inc(putIndex); 
　　++count; 
　　notEmpty.signal();//唤醒非空等待队列中的线程，有关Condition可参考《6.类似Object监视器方法的Condition接口》

 }

　　在这里有几个ArrayBlockingQueue成员变量。items即队列的数组引用，putIndex表示等待插入的数组下标位置。当items[putIndex] = x将新元素插入队列中后，调用inc将数组下标向后移动，如果队列满则将putIndex置为0:

//ArrayBlockingQueue#inc private int inc(int i) { 
　　return (++i == items.length) ? 0 : i; 
}

　　接着解析下put方法，阻塞插入队列，当队列满时不会返回false，也不会抛出异常，而是一直阻塞等待，直到有空位可插入，但它可被中断返回。

//ArrayBlockingQueue#put public void put(E e) throws InterruptedException { 
　　checkNotNull(e);//同样检查插入元素是否为空 　　final ReentrantLock lock = this.lock; 
　　lock.lockInterruptibly();//这里并没有调用lock方法，而是调用了可被中断的lockInterruptibly，该方法可被线程中断返回，lock不能被中断返回。 　　try { 
　　　　while (count == items.length) 
　　　　　　notFull.await();//当队列满时，使非满等待队列休眠 　　　　insert(e);//此时表示队列非满，故插入元素，同时在该方法里唤醒非空等待队列 　　} finally { 
　　　　lock.unlock(); 
　　} 
}

队列元素的删除 

抛出异常	返回值（非阻塞）	一定时间内返回值	返回值（阻塞）
remove()//队列不为空时，返回队首值并移除；队列为空时抛出NoSuchElementException()异常——AbstractQueue	poll()//队列不为空时返回队首值并移除；队列为空时返回null。非阻塞立即返回。	poll(time, unit)//设定等待的时间，如果在指定时间内队列还未孔则返回null，不为空则返回队首值	take(e)//队列不为空返回队首值并移除；当队列为空时会阻塞等待，一直等到队列不为空时再返回队首值。

//AbstractQueue#remove，这也是一个模板方法，定义删除队列元素的算法骨架，队列中元素时返回具体元素，元素为空时抛出异常，具体实现poll由子类实现， public E remove() { 
　　E x = poll();//poll方法由Queue接口定义 　　if (x != null) 
　　　　return x; 
　　else 　　　　throw new NoSuchElementException(); 
}

//ArrayBlockingQueue#poll，队列中有元素时返回元素，不为空时返回null public E poll() { 
　　final ReentrantLock lock = this.lock; 
　　lock.lock(); 
　　try { 
　　　　return (count == 0) ? null : extract(); 
　　} finally { 
　　　　lock.unlock(); 
　　} 
}

//ArrayBlockingQueue#extract private E extract() { 
　　final Object[] items = this.items; 
　　E x = this.<E>cast(items[takeIndex]);//移除队首元素 　　items[takeIndex] = null;//将队列数组中的第一个元素置为null，便于GC回收 　　takeIndex = inc(takeIndex); 
　　--count; 
　　notFull.signal();//唤醒非满等待队列线程 　　return x; 
}

　　对比add和offer方法，理解了上两个方法后remove和poll实际不难理解，同理在理解了put阻塞插入队列后，对比take阻塞删除队列元素同样也很好理解。

//ArrayBlockQueue#take public E take() throws InterruptedException { 
　　final ReentrantLock lock = this.lock(); 
　　lock.lockInterrupted();//这里并没有调用lock方法，而是调用了可被中断的lockInterruptibly，该方法可被线程中断返回，lock不能被中断返回。 　　try { 
　　　　while (count == 0)//队列元素为空 　　　　　　notEmpty.await();//非空等待队列休眠 　　　　return extract();//此时表示队列非空，故删除元素，同时在里唤醒非满等待队列 　　} finally { 
　　　　lock.unlock(); 
　　} 
}

　　最后一个方法size。

public int size() { 
　　final ReentrantLock lock = this.lock; 
　　lock.lock(); 
　　try { 
　　　　return count; 
　　} finally { 
　　　　lock.unlock(); 
　　} 
}

　　可以看到ArrayBlockingQueue队列的size方法，是直接返回的count变量，它不像ConcurrentLinkedQueue，ConcurrentLinkedQueue的size则是每次会遍历这个队列，故ArrayBlockingQueue的size方法比ConcurrentLinkedQueue的size方法效率高。而且ConcurrentLinkedQueue的size方法并没有加锁！也就是说很有可能其size并不准确，这在它的注释中说明了ConcurrentLinkedQueue的size并没有多大的用处。

 

 

The above is the detailed content of ArrayBlockingQueue for concurrent packet blocking queue. For more information, please follow other related articles on the PHP Chinese website!