文章目录
- 前言
- 一、Queue接口的定义
- 二、AbstractQueue实现Queue的基本操作
- 1.AbstractQueue源码注释解析
- 2.方法add、remove、element、clear、addAll的实现原理
- 三、BlockingQueue接口定义解析
- 1.入列操作
- 2.出列操作
- 3.其他操作
- 四、LinkedBlockingQueue源码解析
- 1.LinkedBlockingQueue初步介绍
- 2.链表节点Node介绍
- 3.LinkedBlockingQueue基本属性介绍
- (3.1).capacity队列总容量
- (3.2).count队列节点计数器
- (3.3).head队列头结点
- (3.4).last尾部节点
- (3.5).入队锁putLock、notFull
- (3.6).出队锁takeLock、notEmpty
- 4.LinkedBlockingQueue核心方法源码解析
- (4.1).signalNotEmpty唤醒在notEmpty条件上等待的线程
- (4.2).signalNotFull唤醒在notFull条件上等待的线程
- (4.3).fullyLock锁住入列、出列操作
- (4.4).fullyUnlock解锁入列、出列操作
- (4.5).LinkedBlockingQueue构造函数
- (4.6).enqueue入列函数
- (4.7).dequeue出列函数
- (4.8).size函数统计当前队列节点个数
- (4.9).remainingCapacity函数计算当前队列剩余空间容量
- (5.0).阻塞入列put函数
- (5.1).入列offer函数
- (5.3).阻塞出列take函数
- (5.4).出列poll函数
- (5.5).检索peek函数
前言
队列(Queue)是一种很常见的数据结构,在JAVA中与List、Map、Set并称四大集合,本文将以最常见的阻塞队列LinkedBlockingQueue为例,讲解LinkedBlockingQueue在JAVA中的实现原理。
一、Queue接口的定义
在JAVA中Queue被定义为一个次顶层接口,它的父接口是Collection,Collection这个接口是List、Set、Map、Queue的公共父接口。Collection中定义了一些基本的集合操作方法,比如添加一个元素到集合中add,合并两个集合addAll,清除集合中的所有元素clear,集合中是否包含某个元素contains等操作。Queue接口继承了Collection,那么也就意味着在Collection中定义的这个方法,肯定会在Queue的实现类有具体的实现逻辑。Queue除了继承了Collection定义的基本方法以外,另外新定义了六个方法,它们分别是add、offer、remove、poll、element、peek。
在源码注释中,已经说明了每个方法的使用意义:
add
:这个方法其实是来源于Collection接口中,当需要将一个元素插入列队中时,可以使用该方法。如果插入成功,那么返回true,否则抛出异常。
offer
:方法作用与add一致,也是将一个元素插入队列,但是如果插入失败(队列满了的情况),返回false,并不会抛出异常。
remove
:这个方法也是来源于Collection接口中,用于从队列中移除一个元素。如果移除成功,返回移除的元素,否则抛出异常。
poll
:方法与remove作用一致,从队列中移除一个元素(头元素),如果移除成功,则返回移除的元素,否则返回null。
element
:用于检索队列元素,返回队列中的头元素,但是并不会移除头元素。如果检索失败(队列为空的情况),那么将抛出异常NoSuchElementException
peek
:方法与element一致,也是用于检索元素,如果检索成功返回头元素,检索失败则返回null值。
PS:由此可见poll和peek方法当操作失败时都是返回NULL,那么我们应该禁止将NULL作为元素值插入队列,不然在使用这两个方法时,将混淆拿出来的是值为NULL的元素还是操作返回的NULL
在Queue的源码中整理了一个HTML格式的表格,标识着哪些方法将抛出异常,哪些方法将返回特殊值。
将源码整理复制到HTML文件中打开:
此时我们已经将Queue中定义的六个方法大体的了解了一遍,以下将继续探讨这些方法的具体实现。
二、AbstractQueue实现Queue的基本操作
1.AbstractQueue源码注释解析
在AbstractQueue的源码中,开头有这样一段英文注释:
This class provides skeletal(原始) implementations of some {@link Queue} operations. The implementations in this class are appropriate when the base implementation does not allow null elements. Methods {@link #add add}, {@link #remove remove}, and {@link #element element} are based on {@link #offer offer}, {@link#poll poll}, and {@link #peek peek}, respectively(分别的), but throw exceptions instead of indicating failure via false or null returns.
大概的意思就是:AbstractQueue这个类提供了Queue接口定义的最基本的方法操作的实现。要求插入队列的元素不能为NULL值(至于为什么不能为NULL,在Queue的接口定义我已经阐述过
),add方法、remove方法、element方法都是分别依赖于offer、poll、peek方法来实现的。也就是说add方法底层就是调用的offer方法,只是在offer上进行了封装而已。remove底层调用poll,element底层调用peek。它们之间的差距只是在于一个是抛出异常,一个是返回false或者null值。
2.方法add、remove、element、clear、addAll的实现原理
通过查看add方法可以看到其底层就是调用了offer方法,而offer方法将元素插入对队列成功则返回true,失败则返回false,add判断如果offer如果返回false,则直接抛出IllegalStateException异常。
/**
* Inserts the specified element into this queue if it is possible to do so
* immediately without violating capacity restrictions, returning
* <tt>true</tt> upon success and throwing an <tt>IllegalStateException</tt>
* if no space is currently available.
* (当为到达队列容器限制时,插入指定的元素应该马上返回true表示成功,
* 如果已经到达队列容器上线,那么抛出IllegalStateException)
* 在offer之上再加了一层判断而已
* */
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
通过查看remove方法可以看到其底层就是调用了poll方法,而我们知道调用poll方法时,如果队列为空,返回null值,如果不为空,则移除并返回队列第一个元素。remove就是将poll方法的返回结果再进行判断一次,如果为null,那么就抛出NoSuchElementException异常。
/**
* Retrieves and removes the head of this queue(检索并删除队列头元素).
* This method differs
* from {@link #poll poll} only in that it throws an exception if this
* queue is empty.(不同于poll方法在于当队列为空时,抛出异常)
*
* <p>This implementation returns the result of <tt>poll</tt>
* unless the queue is empty.
*
* @return the head of this queue
* @throws NoSuchElementException if this queue is empty
*/
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
通过查看element方法可以看到其底层就是调用了peek方法,调用peek方法时,如果没有检索到队列元素(队列为空),那么就返回null,element只是在peek返回的值基础上加了一个判断,如果返回为null,那么就抛出NoSuchElementException异常。
/**
* Retrieves, but does not remove, the head of this queue(检索队列头元素但是不会移除该元素). This method
* differs from {@link #peek peek} only in that it throws an exception if
* this queue is empty.(该方法不同于peek方法在于当队列为空的情况下抛出异常)
*
* <p>This implementation returns the result of <tt>peek</tt>
* unless the queue is empty.
*
* @return the head of this queue
* @throws NoSuchElementException if this queue is empty
*/
public E element() {
E x = peek();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
clear方法也相当简单,当poll函数返回的值不为null时,则一直调用poll函数将元素出列。如果队列中元素存在null值时(这种情况不会出现,除非你自己写了一个Queue并允许插入null值
)
/**
* Removes all of the elements from this queue.
* The queue will be empty after this call returns.
*
* <p>This implementation repeatedly invokes {@link #poll poll} until it
* returns <tt>null</tt>.
*/
public void clear() {
while (poll() != null)
;
}
addAll方法就是将一个集合的元素放入队列中,底层就是通过遍历集合,循环调用add方法进行元素入队。
public boolean addAll(Collection<? extends E> c) {
if (c == null)
throw new NullPointerException();
if (c == this)
throw new IllegalArgumentException();
boolean modified = false;
for (E e : c)
if (add(e))
modified = true;
return modified;
}
通过以上源码解析,我们已经对AbstractQueue里的几个方法的实现有了初步了解,add依赖offer,remove依赖poll,element依赖peek。那么offer、poll、peek的又是怎样实现的呢?这三个方法的实现放到了LinkedBlockingQueue中实现。下面我们就来了解LinkedBlockingQueue的底层实现原理。
三、BlockingQueue接口定义解析
BlockingQueue接口继承于Queue接口,在Queue的基础上增加了阻塞方法(Blocks
)和阻塞超时方法(Times out
)。对于插入(Insert)、移除(Remove)、检索(Examine)操作都提供了四种不同的形式,分别是抛出异常、返回特殊值、阻塞、阻塞超时。在BlockingQueue的源码中可以看到其提供了一份基于HTML的表格以总结每个方法属于哪种形式:
将源码注释提取出来以HTML打开后可以看到以下表格:
1.入列操作
对于插入队列的操作,表格上提供了四种方法,add我们已经了解过了,它的实现在AbstractQueue处理,重点put这个新方法,它的作用是将元素放入队列,如果放入元素为null值,那么将抛出空指针异常。如果出现无法入列的情况(满队列时
),那么该线程将一直阻塞直至能将元素放入队列。
/**
* Inserts the specified element into this queue, waiting if necessary
* for space to become available.(当队列容量可用时插入此元素。)
*
* @param e the element to add
* @throws InterruptedException if interrupted while waiting
* @throws ClassCastException if the class of the specified element
* prevents it from being added to this queue
* @throws NullPointerException if the specified element is null
* @throws IllegalArgumentException if some property of the specified
* element prevents it from being added to this queue
*/
void put(E e) throws InterruptedException;
除了put方法,还增加了一个带有阻塞超时的offer方法(offer(E e, long timeout, TimeUnit unit)
),该方法是尝试在指定的时间里内将元素插入队列,如果超过这个时间,则直接返回,并不会一直阻塞直至插入成功。
/**
* 在指定等待时间内将指定元素插入队列
* Inserts the specified element into this queue, waiting up to the
* specified wait time if necessary for space to become available.
*
* @param e the element to add
* @param timeout how long to wait before giving up, in units of
* {@code unit} 多久时间以后放弃插入
* @param unit a {@code TimeUnit} determining how to interpret the
* {@code timeout} parameter 时间单位
* @return {@code true} if successful, or {@code false} if
* the specified waiting time elapses before space is available
* @throws InterruptedException if interrupted while waiting
* @throws ClassCastException if the class of the specified element
* prevents it from being added to this queue
* @throws NullPointerException if the specified element is null 插入元素为null,则报异常
* @throws IllegalArgumentException if some property of the specified
* element prevents it from being added to this queue
*/
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
2.出列操作
对于出列操作(将元素移出队列),表格上也提供了四种方法,remove我们已经在第一小节介绍过了,它的实现在AbstractQueue中处理,重点take这个新方法。检索并移出队列头元素,如果队列为空,则阻塞等待元素插入队列后,将头元素移出。
/**
* Retrieves and removes the head of this queue, waiting if necessary
* until an element becomes available.
*
* @return the head of this queue
* @throws InterruptedException if interrupted while waiting
*/
E take() throws InterruptedException;
除了take这个方法以外,还有增加了一个带有阻塞超时的poll方法,该方法用于在指定时间内尝试移出头元素,如果超出这个时间则放弃本次操作,返回null值。
/**
* Retrieves and removes the head of this queue, waiting up to the
* specified wait time if necessary for an element to become available.
*
* @param timeout how long to wait before giving up, in units of
* {@code unit}
* @param unit a {@code TimeUnit} determining how to interpret the
* {@code timeout} parameter
* @return the head of this queue, or {@code null} if the
* specified waiting time elapses before an element is available
* @throws InterruptedException if interrupted while waiting
*/
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
3.其他操作
除了入列、出列操作以外,还有检索(element、peek
)操作,也就是查看返回队列中的头元素,但并不将其移出队列。
remainingCapacity:用于提供查询当前队列剩余可用量(队列总容量-入列元素总和
)
/**
* Returns the number of additional elements that this queue can ideally
* (in the absence of memory or resource constraints) accept without
* blocking, or {@code Integer.MAX_VALUE} if there is no intrinsic
* limit.
*
* <p>Note that you <em>cannot</em> always tell if an attempt to insert
* an element will succeed by inspecting {@code remainingCapacity}
* because it may be the case that another thread is about to
* insert or remove an element.
*
* @return the remaining capacity
*/
int remainingCapacity();
contains:查看队列中是否包含某个元素,对比两个元素是否相同,使用的是equal方法。如果包含则返回ture,否则返回false。
/**
* Returns {@code true} if this queue contains the specified element.
* More formally, returns {@code true} if and only if this queue contains
* at least one element {@code e} such that {@code o.equals(e)}.
*
* @param o object to be checked for containment in this queue
* @return {@code true} if this queue contains the specified element
* @throws ClassCastException if the class of the specified element
* is incompatible with this queue
* (<a href="../Collection.html#optional-restrictions">optional</a>)
* @throws NullPointerException if the specified element is null
* (<a href="../Collection.html#optional-restrictions">optional</a>)
*/
public boolean contains(Object o);
drainTo:移除队列中的所有元素,并且将元素全部添加到新给定的集合中。这个操作的效率要比重复使用poll方法要快的多。
/**
* Removes all available elements from this queue and adds them
* to the given collection. This operation may be more
* efficient than repeatedly polling this queue. A failure
* encountered while attempting to add elements to
* collection {@code c} may result in elements being in neither,
* either or both collections when the associated exception is
* thrown. Attempts to drain a queue to itself result in
* {@code IllegalArgumentException}. Further, the behavior of
* this operation is undefined if the specified collection is
* modified while the operation is in progress.
*
* @param c the collection to transfer elements into
* @return the number of elements transferred
* @throws UnsupportedOperationException if addition of elements
* is not supported by the specified collection
* @throws ClassCastException if the class of an element of this queue
* prevents it from being added to the specified collection
* @throws NullPointerException if the specified collection is null
* @throws IllegalArgumentException if the specified collection is this
* queue, or some property of an element of this queue prevents
* it from being added to the specified collection
*/
int drainTo(Collection<? super E> c);
在对BlockingQueue里面定义的方法有了初步的了解之后,我们就可以进入它的实现类LinkedBlockingQueue,深入了解这些方法的具体实现逻辑。
四、LinkedBlockingQueue源码解析
在第四节中,我们将挑选LinkedBlockingQueue中常用方法就行源码解析,了解其设计思想和是实现逻辑
1.LinkedBlockingQueue初步介绍
LinkedBlockingQueue在JAVA是比较常见的单向队列(只能在一端删除数据,另一端插入数据
),它是一个有边界队列(队列容量有一个固定大小的上限,一旦队列中的数据对象总量达到容量上限时,无法再进行插入操作
),在创建默认LinkedBlockingQueue时,其容量为Integer.MAX_VALUE,也就是2147483647,因为也可以把它理解为一个无边界队列,但严格来说还是有界的。队列的元素排序方式采用的是FIFO(first-in-first-out)先进先出,这意味着在head(队列头元素)在队列中存在的时间是最久的,而tail(队列尾元素)在队列中存在时间最短。当插入一个元素时,总会将其放到队列的尾部,而移出的元素,总是从队头移出。由于LinkedBlockingQueue带有阻塞的特性,它经常使用在生产-消费模式中。
/* An optionally-bounded {@linkplain BlockingQueue blocking queue} based on
* linked nodes.
* This queue orders elements FIFO (first-in-first-out).此队列按FIFO(先进先出)的顺序
* The <em>head</em> of the queue is that element that has been on the
* queue the longest time.头元素肯定是队列中存在时间最久的
* The <em>tail</em> of the queue is that element that has been on the
* queue the shortest time. New elements
* are inserted at the tail of the queue, and the queue retrieval
* operations obtain elements at the head of the queue.检索操作时获取队列头元素
* Linked queues typically have higher throughput than array-based queues but
* less predictable performance in most concurrent applications.
*
* <p>The optional capacity bound constructor argument serves as a
* way to prevent excessive queue expansion. The capacity, if unspecified,
* is equal to {@link Integer#MAX_VALUE}. Linked nodes are
* dynamically created upon each insertion unless this would bring the
* queue above capacity.
*
* <p>This class and its iterator implement all of the
* <em>optional</em> methods of the {@link Collection} and {@link
* Iterator} interfaces.
*/
2.链表节点Node介绍
Node在链表中统称为节点,节点与节点之间相互引用串联成了链表。在LinkedBlockingQueue中定义了链表节点Node:
/**
* Linked list node class
*/
static class Node<E> {
E item;
/**
* One of:
* - the real successor(后续) Node
* - this Node, meaning the successor is head.next
* - null, meaning there is no successor (this is the last node) 如果是null,表示没有后续节点,该节点为最后一个节点元素
*/
Node<E> next;
Node(E x) {
item = x;
}
}
item:一般称为数据域,存放该节点的真实数据。
next:一般称为指针域,维护着下一个节点的引用,以便于通过节点查找到一下一个节点。
如果您对链表结构比较陌生,您可以尝试先浏览链接数据结构之链表了解一下,此处我就不在阐述链表的特性和用法。可以看出Node的数据结构很简单,一个节点只维护当前节点的元素值和指向下一个节点的引用,如果指向下一个节点的引用为null,那么意味着当前节点为链表尾部节点。
3.LinkedBlockingQueue基本属性介绍
在LinkedBlockingQueue中定义了很多属性,我将根据源码顺序依次进行介绍:
(3.1).capacity队列总容量
capacity:队列总容量,该属性标识着一个队列最多能容纳多少个元素(
节点
),在初始化LinkedBlockingQueue的时候,会将改属性赋值为Integer.MAX_VALUE(2147483647
)。意味着队列最多可以容纳2147483647个节点。
/**
* The capacity bound, or Integer.MAX_VALUE if none 队列容量
*/
private final int capacity;
/**
* Creates a {@code LinkedBlockingQueue} with a capacity of
* {@link Integer#MAX_VALUE}. 初始化LinkedBlockingQueue对象
*/
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
/**
* Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
*
* @param capacity the capacity of this queue
* @throws IllegalArgumentException if {@code capacity} is not greater
* than zero
* 设置capacity为Integer.MAX_VALUE
*/
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
(3.2).count队列节点计数器
count:在LinkedBlockingQueue中作为统计队列节点数量的计数器,当有新的元素入列时,count会增加1,出列是会减少1。设计这个计数器的作用是可以方便得知队列有效长度,而不需要每次从头节点遍历一次来得出队列有效长度。
/**
* Current number of elements 当前队列元素数量
*/
private final AtomicInteger count = new AtomicInteger();
(3.3).head队列头结点
head:作为整个队列的头节点,要注意的是在LinkedBlockingQueue中,头节点的数据域(
item
)永远是null,不维护任何信息,当前队列不会空时,它的指针域必定不为空。在初始化LinkedBlockingQueue时,head既是头节点,也是尾部节点,即head==last。
/**
* Head of linked list.
* Invariant: head.item == null
*/
transient Node<E> head;
(3.4).last尾部节点
last:作为队列的尾部节点,该变量永远指向队列最后一个节点。因此它的指针域必定为null。在初始化LinkedBlockingQueue时,last既是尾部点,也是头节点,即last==head。
(3.5).入队锁putLock、notFull
在LinkedBlockingQueue中如果要进行入列操作,一般调用方法put或者是offer,而我们知道put和offer是有阻塞效果的。导致其阻塞的就是入队锁putLock。在LinkedBlockingQueue中,将putLock定义为ReentrantLock类型,notFull为putLock的Condition。
/**
* Lock held by put, offer, etc
*/
private final ReentrantLock putLock = new ReentrantLock();
/**
* Wait queue for waiting puts
*/
private final Condition notFull = putLock.newCondition();
putLock与putLock的组合,就能实现线程等待、唤醒等效果,以此来实现入列阻塞。在调用put或者offer方法时,线程会尝试去获取对象锁,如果锁不可用,那么为了线程调度目的,当前线程将被禁用,并处于休眠状态,直到获得锁。获取到锁时。
(3.6).出队锁takeLock、notEmpty
既然有入队锁,那么肯定就有出队锁,在LinkedBlockingQueue中如果要进行出列操作,一般调用方法take或者是poll,而我们知道take和poll是有阻塞效果的。导致其阻塞的就是出队锁takeLock。在LinkedBlockingQueue中,将takeLock定义为ReentrantLock类型,notEmpty为takeLock的Condition。
/**
* Lock held by take, poll, etc
*/
private final ReentrantLock takeLock = new ReentrantLock();
/**
* Wait queue for waiting takes
*/
private final Condition notEmpty = takeLock.newCondition();
takeLock与notEmpty的组合,就能实现线程等待、唤醒等效果,以此来实现出列阻塞。在调用take或者poll方法时,线程会尝试去获取对象锁,如果锁不可用,那么为了线程调度目的,当前线程将被禁用,并处于休眠状态,直到获得锁。获取到锁时。到此为止,LinkedBlockingQueue中的基本属性就结束完了。接下来将介绍LinkedBlockingQueue中的常用重点方法。
4.LinkedBlockingQueue核心方法源码解析
(4.1).signalNotEmpty唤醒在notEmpty条件上等待的线程
signalNotEmpty根据名字就可以猜测是唤醒某个等待线程,not empty意味着队列不为空,如果队列不为空时,那么就可以做出列操作。那么这里的signalNotEmpty方法就是唤醒某个等待进行出列操作的线程。也就是某个线程调用了take或者poll方法。可能由于队列为空,导致线程阻塞休眠,而当队列不为空时,则调用该方法唤醒线程,进行出列操作。
/**
* Signals a waiting take. Called only from put/offer (which do not
* otherwise ordinarily lock takeLock.take操作信息将被唤醒,但是这个唤醒操作由put/offer两个操作来触发)
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal();
} finally {
takeLock.unlock();
}
}
既然该方法是用于唤醒等待进行出列操作的线程,那么是由谁来调用的呢?我们可以猜测一下,当队列为空时,调用take或者poll将导致线程阻塞,当队列不为空时,将通过signalNotEmpty唤醒调用take或者poll阻塞的线程,什么时候队列不为空?那肯定是有元素人列的时候,队列就不会为空。那么signalNotEmpty就极有可能是在调用put或者offer的时候,元素入列完成后进行调用。我们可以通过IDEA查看哪些地方调用了signalNotEmpty这个方法,得出的结果如下图,很显然验证了我们的猜想。当调用put或者offer入列完成后,会调用signalNotEmpty唤醒出列阻塞线程。
(4.2).signalNotFull唤醒在notFull条件上等待的线程
如果你以及理解了signalNotEmpty方法的原理,那么signalNotFull就变得相当简单。signal是唤醒的意思,not full则是不处于饱和状态。该方法用于当队列不是满队列的情况时,唤醒等待入列的某个线程。put或者offer用于将元素插入队列。但是当队列满了的情况下,线程调用put或者offer将会被阻塞休眠,直到队列不处于满状态,将元素入列。由此可见,什么时候队列会从满队列变为不满状态,那肯定是有出列操作(take或者poll)时,才会将满队列变得空闲。那么显而易见,signalNotFull则是在进行出列操作时进行调用,以此唤醒入列线程。
/**
* Signals a waiting put. Called only from take/poll.唤醒put操作,这个操作由take/poll进行触发
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
同样的可以使用IDEA查看哪些地方使用了signalNotFull,以此证明我们的猜想是否正确。
(4.3).fullyLock锁住入列、出列操作
在源码中,fullyLock分别调用putLock、takeLock进行锁定:
/**
* Locks to prevent both puts and takes(同时锁住put和takes操作).
*/
void fullyLock() {
putLock.lock();
takeLock.lock();
}
我们可以想象一样,什么时候需要将入列和出列操作锁住呢?当对队列进行遍历时,进行指定移除某个元素操作,或者说是判断队列是否包含某个元素时,往往就需要对入列和出列进行上锁以确保程序准确性。同样的可以使用IDEA查看哪些地方用到该方法:
(4.4).fullyUnlock解锁入列、出列操作
当对队列进行遍历,进行指定移除某个元素操作,或者说是判断队列是否包含某个元素,完成以上操作后,需要将入列和出列操作解锁。以便于不影响后续出入列操作。
/**
* Unlocks to allow both puts and takes.(解锁允许put和take操作)
*/
void fullyUnlock() {
takeLock.unlock();
putLock.unlock();
}
(4.5).LinkedBlockingQueue构造函数
LinkedBlockingQueue源码提供了三个默认的构造函数,LinkedBlockingQueue()和LinkedBlockingQueue(int capacity)以及LinkedBlockingQueue(Collection<? extends E> c)三个构造函数,使用不带参数的构造函数时,默认的队列容量为Integer.MAX_VALUE(2147483647),当然在构建LinkedBlockingQueue时我们也可以自定义队列容量。当初始化队列容量的同时,也分别给head节点和last节点初始化值。
第三个构造参数可以指定集合加入队列LinkedBlockingQueue(Collection<? extends E> c),默认调用有参构造函数初始化队列容量,使用putLock.lock加锁,循环集合插入队列,并记录当前队列有效节点数量,操作完成后 putLock.unlock解锁以便于后续操作。
/**
* Creates a {@code LinkedBlockingQueue} with a capacity of
* {@link Integer#MAX_VALUE}, initially containing the elements of the
* given collection,
* added in traversal order of the collection's iterator.
*
* @param c the collection of elements to initially contain
* @throws NullPointerException if the specified collection or any
* of its elements are null
*/
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);//调用有参构造函数初始化队列
final ReentrantLock putLock = this.putLock;
putLock.lock(); // Never contended, but necessary for visibility(此处不会出现竞争关系,但是加锁也是必要的)
try {
int n = 0;
for (E e : c) { //循环集合
if (e == null)
throw new NullPointerException();
if (n == capacity)
throw new IllegalStateException("Queue full");
enqueue(new Node<E>(e)); //初始化一个node并插入队列
++n;
}
count.set(n);//记录当前队列节点数量
} finally {
putLock.unlock();
}
}
(4.6).enqueue入列函数
在LinkedBlockingQueue中,节点入列操作都是调用enqueue函数实现的,一般是由put或者offer发起操作。enqueue函数源代码也十分简洁,源码中仅用一行代码搞定,如果你对队列不熟悉,那么此处你将十分疑惑为何一行代码就能完成队列插入的操作。
我们知道last节点是队列中的尾节点,如果有新的元素需要插入队列时,那么该元素节点(Node包含数据域和指针域
)就该链接到当前队列的尾部节点之后,也就是将尾部节点的指针域指向新节点,即:this.last.next=新节点,所以源码中的
last = last.next = node; 节点入列操作
就可以分为两步操作。
第一步将当前尾部节点的指针域指向新插入的节点,也就是
last.next=node;
第二步则是更新尾部节点last的指向,因为last节点永远要指向队列中最后一个节点,所以要更新last节点指向新插入的节点
this.last = node;
具体操作流程如下图所示:
通过分析我们已经将last = last.next = node; 分成两步完成整个入列操作,但是有一个疑问在于,我们只看到对last的处理,对head的处理并未在enqueue函数中有所体现,而且出列时是从head节点进行操作的。让我们再次回到构造函数查看源码:
你会发现,在初始化LinkedBlockingQueue时,初始化化了一个数据域为null的节点,并且该节点同时指向last和head,也就是说在初始化完成LinkedBlockingQueue时,last==head是成立的。那么在第一次调用enqueue函数了,last = last.next = node;,就变成了:
head.next=node; 头节点指针域指向node
last=node;
这样一来,head节点就与last节点关联起来,而后续再次调用enqueue函数时,由于head和last并不指向同一个节点Node,因此head的指针域(next)不会改变,只会改变last的后续指针域并将last指向新增节点。
(4.7).dequeue出列函数
在LinkedBlockingQueue中,节点出列操作都是调用dequeue函数实现的,一般是由take或者poll发起操作。
在(3.3).head队列头结点
介绍中我们知道head并不存储数据,它的下一个节点才是我们正真使用的节点。出队操作时,先得到头节点(head)的下一个节点first节点,将当前头节点的next指针域指向自己,代码中说是help gc,大概意思就是帮助头节点更好的被回收。然后将first作为头节点head,并将head节点的数据域(元素数据)拿出,然后将head数据域置为null并将刚刚拿出的元素数据返回。
如果用动态图演示可以,如下所示:
(4.8).size函数统计当前队列节点个数
size方法只是将属性count的值进行返回,我们知道在进行入列(put、offer)和出列(take、poll)时,count会进行对应的加或者减。这里count的值就代表这整个队列中,节点个数总和。
(4.9).remainingCapacity函数计算当前队列剩余空间容量
remainingCapacity函数是将队列容量减去当前有效节点数,获得最终剩余空间容量
/**
* Returns the number of additional elements that this queue can ideally
* (in the absence of memory or resource constraints) accept without
* blocking. This is always equal to the initial capacity of this queue
* less the current {@code size} of this queue.
*
* <p>Note that you <em>cannot</em> always tell if an attempt to insert
* an element will succeed by inspecting {@code remainingCapacity}
* because it may be the case that another thread is about to
* insert or remove an element.
*/
/**
* 返回队列剩余容量
*
* @return
*/
public int remainingCapacity() {
return capacity - count.get();
}
(5.0).阻塞入列put函数
在LinkedBlockingQueue中,入列操作都具有一般的具有阻塞特性,put函数融入了入队锁(putLock)来实现线程入列安全性和阻塞效果:
/**
* Inserts the specified element at the tail of this queue, waiting if
* necessary for space to become available.等待队列由可用量时将元素插入队列
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();/**获取锁,如果有多个线程同时尝试使用lockInterruptibly获取锁,没有获取锁的线程,可用使用interrupt终止获取锁等待**/
try {
/*
* Note that count is used in wait guard even though it is
* not protected by lock. This works because count can
* only decrease at this point (all other puts are shut
* out by lock), and we (or some other waiting put) are
* signalled if it ever changes from capacity. Similarly
* for all other uses of count in other wait guards.
*/
while (count.get() == capacity) {
notFull.await();//当前队列已经到达最大容量,notFull睡眠,此时,不允许进行插入操作,等到take或者poll操作时,将其唤醒(signalNotFull方法)
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();//唤醒take
}
put函数主要做以下几件事:
1.构建一个Node节点对象,将元素放入节点数据域中。
Node<E> node = new Node<E>(e);
2.线程尝试获取入队锁,如果获取失败,线程将阻塞休眠在这一步。
putLock.lockInterruptibly();
3.如果获取到入队锁,那么判断队列是否是满队列,当前队列节点数量是否等于队列最大容量count.get() == capacity,如果是满队列,那么不满足插入条件,线程将进入休眠状态,等待有出列操作时(take,poll)调用notFull.signal唤醒线程。
/*
* Note that count is used in wait guard even though it is
* not protected by lock. This works because count can
* only decrease at this point (all other puts are shut
* out by lock), and we (or some other waiting put) are
* signalled if it ever changes from capacity. Similarly
* for all other uses of count in other wait guards.
*/
while (count.get() == capacity) {
notFull.await();//当前队列已经到达最大容量,notFull睡眠,此时,不允许进行插入操作,等到take或者poll操作时,将其唤醒(signalNotFull方法)
}
4.如果满足入队条件(非满队列情况
),则将新的节点入列,调用enqueue方法,将计数器count值赋值给变量c,然后计数器count自增1,判断如果非满队列,则调用notFull.signal唤醒拥有入队锁的睡眠线程。
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
5.判断变量c是否为0,c变量的值是计数器count未自增1时的值,如果c为0,那么表示之前队列属于空队列,那么可能存在操作出列的线程处理于休眠状态,此时调用signalNotEmpty函数唤醒拥有出队锁的休眠线程,告知线程当前队列不为空,可以进行元素出列操作。
if (c == 0)
signalNotEmpty();//唤醒take
(5.1).入列offer函数
offer函数在LinkedBlockingQueue中有两个具体的实现,一个是带有阻塞超时效果的offer(E e, long timeout, TimeUnit unit) 另一个是带有阻塞效果的offer(E e)。
(1).offer(E e) 的实现于put函数逻辑大体一致,只是在构建Node对象之前,优先判断队列是否已满,如果已满则直接返回false表示插入失败。不同于put函数,offer判断是否满队列的逻辑在构建Node节点之前,因此当满队列时,offer不会出现线程阻塞效果,而是直接返回false,而put函数则会一直等待直到队列空闲,将节点插入队列。
/**
* Inserts the specified element at the tail of this queue if it is
* possible to do so immediately without exceeding the queue's capacity,
* returning {@code true} upon success and {@code false} if this queue
* is full.
* When using a capacity-restricted queue, this method is generally
* preferable to method {@link BlockingQueue#add add}, which can fail to
* insert an element only by throwing an exception.
*
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;//满队列直接返回失败
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
}
(2).offer(E e, long timeout, TimeUnit unit) 可以指定在线程等待插入时间,如果超过指定时间,则返回false表示插入失败。
unit.toNanos(timeout) 会将指定超时时间转化为毫秒处理,count.get() == capacity如果成立,则表示当前队列已经处于满队列的状态,则线程将调用awaitNanos方法进入睡眠状态。awaitNanos方法在await方法的基础上,增加了超时跳出的机制,如果睡眠时间超过nanos 毫秒,则自动唤醒睡眠线程,此时返回的nanos 值为小于等于0。唤醒线程再次判断当前队列是否为满队,如果count.get() == capacity依然成立,则返回false。如果不成立则跳出while循环进行插入操作。另一种情况则是线程由take或者poll 函数调用 notFull.signal(); 唤醒,这种被动唤醒的方式,notFull.awaitNanos(nanos) 返回的值肯定大于等于0,由于调用了take或者poll 函数,进行了出列操作,则count.get() == capacity 并不成立,则线程将跳出循环,进行插入操作。
/**
* Inserts the specified element at the tail of this queue, waiting if
* necessary up to the specified wait time for space to become available.
*
* @return {@code true} if successful, or {@code false} if
* the specified waiting time elapses before space is available
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}
(5.3).阻塞出列take函数
如果你掌握刚刚讲诉的put函数和offer函数的实现逻辑,那么take和poll函数的底层实现就变得简单明了。take函数实现逻辑则是先获取入队锁,如果获取失败则阻塞,获取成功则判断队列是否为空队列,如果count.get() == 0成立,则表示当前队列为空队列,则线程调用notEmpty.await() 进入休眠状态,直到其他线程调用put或者poll函数,将新的节点插入队列后,调用notEmpty.signal方法唤醒该线程,告知该线程当前队列不为空队列,可以进行出列操作。计数器count将未自减的值赋值给变量c,当前c == capacity成立时,则表示在未出列时,队列处于满列状态,可能存在请求入列操作的休眠线程,当出列完成后,队列处于未满状态,则通过调用signalNotFull方法唤醒休眠的入列线程。
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity) //c的值是count未自减的值,如果未自减是时满队列,则自减后处于非满状态,则应该唤醒休眠的入列线程。
signalNotFull();
return x;
}
(5.4).出列poll函数
poll函数于take函数相比,拥有阻塞超时的效果,其原理和offer函数十分类似,这里则不在进行讲诉。您可以通过源码自行理解其实现逻辑。
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0) //判断是否为空队列,是则直接返回null
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
if (nanos <= 0)
return null;
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
(5.5).检索peek函数
peek函数通常用于查看当前队列中第一个元素,通过head.next找到第一个正真的节点对象,如果节点存在,则返回节点的数据域(item)。
public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}