简介

队列是一种先进先出的数据结构，在排队、削峰、缓存等多种场景下都会用到。今天学习下JUC中提供的并发队列-ConcurrentLinkedQueue

可以看他的继承和实现接口非常简单，继承了AbstractQueue类，实现了Queue接口。

ConcurrentLinkedQueue

add

在队列尾部新添加一个元素

/**
 * Inserts the specified element at the tail of this queue.
 * As the queue is unbounded, this method will never throw
 * {@link IllegalStateException} or return {@code false}.
 *
 * @return {@code true} (as specified by {@link Collection#add})
 * @throws NullPointerException if the specified element is null
 */
 //在队列尾部加入新的元素，元素不能为null。这里直接调用看offer方法。
public boolean add(E e) {
    return offer(e);
}

offer

在队列尾部新添加一个元素

/**
 * Inserts the specified element at the tail of this queue.
 * As the queue is unbounded, this method will never return {@code false}.
 *
 * @return {@code true} (as specified by {@link Queue#offer})
 * @throws NullPointerException if the specified element is null
 */
public boolean offer(E e) {
    //创建一个新的节点，新的节点元素不能为null。
    final Node<E> newNode = new Node<E>(Objects.requireNonNull(e));

    //从tail开始遍历队列，直到队尾
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        //如果p是最后一个节点
        if (q == null) {
            // p is last node
            //通过CAS的方式把newNode加到下一个节点
            if (NEXT.compareAndSet(p, null, newNode)) {
                // Successful CAS is the linearization point
                // for e to become an element of this queue,
                // and for newNode to become "live".
                //插入成功后，再次判断，tail是否已经改变，如果p != tail，则尝试将newNode设置为Tail，设置失败也没关系，因为有其它线程设置了
                if (p != t) // hop two nodes at a time; failure is OK
                    TAIL.weakCompareAndSet(this, t, newNode);
                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
        else if (p == q)
            // We have fallen off list.  If tail is unchanged, it
            // will also be off-list, in which case we need to
            // jump to head, from which all live nodes are always
            // reachable.  Else the new tail is a better bet.
            //当p == q时如并发时节点被删除等。需要重新设置p
            p = (t != (t = tail)) ? t : head;
        else
            // Check for tail updates after two hops.
            //如果p在上次赋值后，if处理前，节点有新增，则走到这一步
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

poll

取出队列头部的一个元素删除

public E poll() {
    restartFromHead: for (;;) {
        for (Node<E> h = head, p = h, q;; p = q) {
            final E item;
            //从队列首部head取出元素，并使用cas将头部设为null。
            if ((item = p.item) != null && p.casItem(item, null)) {
                // Successful CAS is the linearization point
                // for item to be removed from this queue.
                if (p != h) // hop two nodes at a time
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //如果队列空了，返回null
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //如果p已经被删除 返回头部重新开始获取
            else if (p == q)
                continue restartFromHead;
        }
    }
}

peek

取出队列头的元素，但不删除

public E peek() {
    restartFromHead: for (;;) {
        for (Node<E> h = head, p = h, q;; p = q) {
            final E item;
            //返回head的元素，如果队列为空，返回null
            if ((item = p.item) != null
                || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            }
            //如果head被删除掉，则返回头部继续获取
            else if (p == q)
                continue restartFromHead;
        }
    }
}

remove

//取出队列头的元素并在队列中删除，和poll的区别是，如果队列为空，remove会抛出异常。而poll会返回null
/**
 * 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.
 *
 * <p>This implementation returns the result of {@code poll}
 * unless the queue is empty.
 *
 * @return the head of this queue
 * @throws NoSuchElementException if this queue is empty
 */
public E remove() {
    E x = poll();
    //调用poll方法，获取队列头部的元素并删除，如果队列为空，抛出异常
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

//移除队列中的某个元素
/**
 * Removes a single instance of the specified element from this queue,
 * if it is present.  More formally, removes an element {@code e} such
 * that {@code o.equals(e)}, if this queue contains one or more such
 * elements.
 * Returns {@code true} if this queue contained the specified element
 * (or equivalently, if this queue changed as a result of the call).
 *
 * @param o element to be removed from this queue, if present
 * @return {@code true} if this queue changed as a result of the call
 */
public boolean remove(Object o) {
    if (o == null) return false;
    restartFromHead: for (;;) {
        //从队列头部开始查找
        for (Node<E> p = head, pred = null; p != null; ) {
            Node<E> q = p.next;
            final E item;
            //如果找到等于o的元素，则移除并返回true
            if ((item = p.item) != null) {
                if (o.equals(item) && p.casItem(item, null)) {
                    skipDeadNodes(pred, p, p, q);
                    return true;
                }
                pred = p; p = q; continue;
            }
            //如果是队列尾部，判断是否新插入了数据。如果插入了，则返回头部重新查找
            for (Node<E> c = p;; q = p.next) {
                if (q == null || q.item != null) {
                    pred = skipDeadNodes(pred, c, p, q); p = q; break;
                }
                if (p == (p = q)) continue restartFromHead;
            }
        }
        //如果没找到 返回false
        return false;
    }
}

element

获取队列头的元素，但不会从队列中删除，和peek的区别是，如果队列为空，则element会抛出异常，peek是返回null

/**
 * 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.
 *
 * <p>This implementation returns the result of {@code peek}
 * unless the queue is empty.
 *
 * @return the head of this queue
 * @throws NoSuchElementException if this queue is empty
 */
public E element() {
    E x = peek();
    //调用peek方法返回队列头部数据但不删除，如果队列为空返回null，则抛出异常
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

isEmpty

判断队列是否为空

/**
 * Returns {@code true} if this queue contains no elements.
 *
 * @return {@code true} if this queue contains no elements
 */
public boolean isEmpty() {
    //如果第一个节点时null 则是空的
    return first() == null;
}

/**
 * Returns the first live (non-deleted) node on list, or null if none.
 * This is yet another variant of poll/peek; here returning the
 * first node, not element.  We could make peek() a wrapper around
 * first(), but that would cost an extra volatile read of item,
 * and the need to add a retry loop to deal with the possibility
 * of losing a race to a concurrent poll().
 */
 //会返回第一个Node，和peek的区别是，peek是返回的第一个Node中的元素
Node<E> first() {
    restartFromHead: for (;;) {
        for (Node<E> h = head, p = h, q;; p = q) {
            boolean hasItem = (p.item != null);
            if (hasItem || (q = p.next) == null) {
                updateHead(h, p);
                return hasItem ? p : null;
            }
            else if (p == q)
                continue restartFromHead;
        }
    }
}

size

获取队列的大小，会遍历队列计算。但是因为此方法无锁，获取的数据可能会不准确。

/**
 * Returns the number of elements in this queue.  If this queue
 * contains more than {@code Integer.MAX_VALUE} elements, returns
 * {@code Integer.MAX_VALUE}.
 *
 * <p>Beware that, unlike in most collections, this method is
 * <em>NOT</em> a constant-time operation. Because of the
 * asynchronous nature of these queues, determining the current
 * number of elements requires an O(n) traversal.
 * Additionally, if elements are added or removed during execution
 * of this method, the returned result may be inaccurate.  Thus,
 * this method is typically not very useful in concurrent
 * applications.
 *
 * @return the number of elements in this queue
 */
public int size() {
    restartFromHead: for (;;) {
        int count = 0;
        
        //会遍历队列中的节点。count值每次加一。因此，size方法是比较耗时的，且在队列中的节点删除或添加时，这个值可能不准。因此建议还是用isEmpty来判断队列是否为空。尽量别使用此方法。
        for (Node<E> p = first(); p != null;) {
            if (p.item != null)
                if (++count == Integer.MAX_VALUE)
                    break;  // @see Collection.size()
            if (p == (p = p.next))
                continue restartFromHead;
        }
        return count;
    }
}

总结

通过对源码的学习，我们了解到以下几个重点：

• ConcurrentLinkedQueue是基于CAS来进行并发控制的，因此同步的代价较小，并发性能比较好。是一个非阻塞队列
• ConcurrentLinkedQueue是***的，队列中的元素无个数限制
• size方法需要遍历队列的节点时间复杂度为O(n)，性能会随着队列长度降低，且因为无锁，队列可能被增删，不一定能获取到正确的结果。

更多文章

见我的博客：https://nc2era.com

written by AloofJr，转载请注明出处