当前位置: 首页 > article >正文

深度了解LinkedBlockingQueue底层实现原理

文章目录

  • 前言
  • 一、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为例,讲解LinkedBlockingQueueJAVA中的实现原理。


一、Queue接口的定义

JAVAQueue被定义为一个次顶层接口,它的父接口是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方法都是分别依赖于offerpollpeek方法来实现的。也就是说add方法底层就是调用的offer方法,只是在offer上进行了封装而已。remove底层调用pollelement底层调用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依赖offerremove依赖pollelement依赖peek。那么offerpollpeek的又是怎样实现的呢?这三个方法的实现放到了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,而我们知道putoffer是有阻塞效果的。导致其阻塞的就是入队锁putLock。在LinkedBlockingQueue中,将putLock定义为ReentrantLock类型,notFullputLockCondition

    /**
     * Lock held by put, offer, etc
     */
    private final ReentrantLock putLock = new ReentrantLock();

    /**
     * Wait queue for waiting puts
     */
    private final Condition notFull = putLock.newCondition();

putLockputLock的组合,就能实现线程等待、唤醒等效果,以此来实现入列阻塞。在调用put或者offer方法时,线程会尝试去获取对象锁,如果锁不可用,那么为了线程调度目的,当前线程将被禁用,并处于休眠状态,直到获得锁。获取到锁时。

(3.6).出队锁takeLock、notEmpty

既然有入队锁,那么肯定就有出队锁,在LinkedBlockingQueue中如果要进行出列操作,一般调用方法take或者是poll,而我们知道takepoll是有阻塞效果的。导致其阻塞的就是出队锁takeLock。在LinkedBlockingQueue中,将takeLock定义为ReentrantLock类型,notEmptytakeLockCondition

    /**
     * Lock held by take, poll, etc
     */
    private final ReentrantLock takeLock = new ReentrantLock();

    /**
     * Wait queue for waiting takes
     */
    private final Condition notEmpty = takeLock.newCondition();

takeLocknotEmpty的组合,就能实现线程等待、唤醒等效果,以此来实现出列阻塞。在调用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分别调用putLocktakeLock进行锁定:

    /**
     * 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的节点,并且该节点同时指向lasthead,也就是说在初始化完成LinkedBlockingQueue时,last==head是成立的。那么在第一次调用enqueue函数了,last = last.next = node;,就变成了:

head.next=node; 头节点指针域指向node
last=node;

这样一来,head节点就与last节点关联起来,而后续再次调用enqueue函数时,由于headlast并不指向同一个节点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函数的实现逻辑,那么takepoll函数的底层实现就变得简单明了。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();
        }
    }

http://www.kler.cn/a/137024.html

相关文章:

  • 密码学期末考试笔记
  • 《计算机组成及汇编语言原理》阅读笔记:p48-p81
  • 基于python+django的外卖点餐系统
  • STM32单片机使用CAN协议进行通信
  • 计算机毕业设计原创定制(免费送源码):NodeJS+MVVM+MySQL 樱花在线视频网站
  • 【序】前端监控:打造高效稳定的用户体验
  • SpringSecurity+JWT权限认证
  • 2023年中国聚氨酯树脂涂料需求量、市场规模及行业趋势分析[图]
  • Goby 漏洞发布|大华智慧园区综合管理平台 poi 文件上传漏洞
  • JMM并发三大特性
  • 7-tcp 三次握手和四次挥手、osi七层协议,哪七层,每层有哪些?tcp和udp的区别?udp用在哪里了?
  • 在Spring Boot中使用Thymeleaf开发Web页面
  • 数据库管理-第117期 拿下19c OCM(202301121)
  • Go语言中获取协程ID
  • Golang抓包:实现网络数据包捕获与分析
  • 后端接口测试,令牌校验住,获取tocken 接口的方式
  • centos更换yum源
  • uni-app(1)pages. json和tabBar
  • 如果文件已经存在与git本地库中,配置gitignore能否将其从git库中删除
  • 【python学习】基础篇-常用函数-sorted() 对可迭代对象进行排序
  • 黄鹤楼
  • 大数据研发工程师课前环境搭建
  • java获取shp文件坐标系
  • 计算机类编程课学生编写的代码应该如何管理
  • 一个完备的手游地形实现方案
  • “三面一体”的业务调度方案在运营商订单运营的实践