小网站推荐一个,延安做网站的公司电话,高度国际装饰公司官网,seo搜索引擎优化主要做什么文章目录 前言一、Queue接口的定义二、AbstractQueue实现Queue的基本操作1.AbstractQueue源码注释解析2.方法add、remove、element、clear、addAll的实现原理 三、BlockingQueue接口定义解析1.入列操作2.出列操作3.其他操作 四、LinkedBlockingQueue源码解析1.LinkedBlockingQu… 文章目录 前言一、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被定义为一个次顶层接口它的父接口是CollectionCollection这个接口是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底层调用pollelement底层调用peek。它们之间的差距只是在于一个是抛出异常一个是返回false或者null值。
2.方法add、remove、element、clear、addAll的实现原理
通过查看add方法可以看到其底层就是调用了offer方法而offer方法将元素插入对队列成功则返回true失败则返回falseadd判断如果offer如果返回false则直接抛出IllegalStateException异常。 /*** Inserts the specified element into this queue if it is possible to do so* immediately without violating capacity restrictions, returning* tttrue/tt upon success and throwing an ttIllegalStateException/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方法在于当队列为空时抛出异常** pThis implementation returns the result of ttpoll/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方法时如果没有检索到队列元素(队列为空)那么就返回nullelement只是在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方法在于当队列为空的情况下抛出异常** pThis implementation returns the result of ttpeek/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.** pThis implementation repeatedly invokes {link #poll poll} until it* returns ttnull/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依赖offerremove依赖pollelement依赖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.** pNote that you emcannot/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-restrictionsoptional/a)* throws NullPointerException if the specified element is null* (a href../Collection.html#optional-restrictionsoptional/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 emhead/em of the queue is that element that has been on the* queue the longest time.头元素肯定是队列中存在时间最久的* The emtail/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.** pThe 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.** pThis class and its iterator implement all of the* emoptional/em methods of the {link Collection} and {link* Iterator} interfaces.*/2.链表节点Node介绍
Node在链表中统称为节点节点与节点之间相互引用串联成了链表。在LinkedBlockingQueue中定义了链表节点Node: /*** Linked list node class*/static class NodeE {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表示没有后续节点该节点为最后一个节点元素*/NodeE 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 NodeE(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既是头节点也是尾部节点即headlast。 /*** Head of linked list.* Invariant: head.item null*/transient NodeE head;(3.4).last尾部节点 last:作为队列的尾部节点该变量永远指向队列最后一个节点。因此它的指针域必定为null。在初始化LinkedBlockingQueue时last既是尾部点也是头节点即lasthead。 (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 collections 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 NodeE(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.nextnode; 第二步则是更新尾部节点last的指向因为last节点永远要指向队列中最后一个节点所以要更新last节点指向新插入的节点 this.last node; 具体操作流程如下图所示:
通过分析我们已经将last last.next node; 分成两步完成整个入列操作但是有一个疑问在于我们只看到对last的处理对head的处理并未在enqueue函数中有所体现而且出列时是从head节点进行操作的。让我们再次回到构造函数查看源码: 你会发现在初始化LinkedBlockingQueue时初始化化了一个数据域为null的节点并且该节点同时指向last和head,也就是说在初始化完成LinkedBlockingQueue时lasthead是成立的。那么在第一次调用enqueue函数了last last.next node;就变成了: head.nextnode; 头节点指针域指向node lastnode; 这样一来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.** pNote that you emcannot/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;NodeE node new NodeE(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节点对象将元素放入节点数据域中。
NodeE node new NodeE(e);2.线程尝试获取入队锁如果获取失败线程将阻塞休眠在这一步。
putLock.lockInterruptibly();3.如果获取到入队锁那么判断队列是否是满队列当前队列节点数量是否等于队列最大容量count.get() capacity如果是满队列那么不满足插入条件线程将进入休眠状态等待有出列操作时(takepoll)调用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是否为0c变量的值是计数器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 queues 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;NodeE node new NodeE(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 NodeE(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 {NodeE first head.next;if (first null)return null;elsereturn first.item;} finally {takeLock.unlock();}}
