文章目录
- 前言
- 一、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;elsethrow 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;elsethrow 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;elsethrow 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) //判断是否为空队列,是则直接返回nullreturn 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;elsereturn first.item;} finally {takeLock.unlock();}}