JDK源码那些事儿之ConcurrentLinkedQueue

Posted 2020-11-21 freeorange

阻塞队列的实现前面已经讲解完毕，今天我们继续了解源码中非阻塞队列的实现，接下来就看一看ConcurrentLinkedQueue非阻塞队列是怎么完成操作的

前言

JDK版本号：1.8.0_171

ConcurrentLinkedQueue是一个基于链表实现的无界的线程安全的FIFO非阻塞队列。最大的不同之处在于非阻塞特性，不会进行阻塞等待直接返回操作结果。其中head和tail的更新类似之前在LinkedTransferQueue中讲解的slack(松弛度)机制，只有在slack阈值大于等于2时才会进行更新，尽量减少CAS的操作次数，当然，这样的操作也提高了代码实现的复杂度

类定义

从关系图上我们也可以看到ConcurrentLinkedQueue没有去实现BlockingQueue接口

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable

实现流程

为了了解其内部实现的操作，可以看下面的过程图理解其内部结点入队出队的转换过程

常量/变量

除了CAS需要使用的常量，就只剩下head和tail两个引用结点，在其注释部分可以看到作者的说明，这里解释下：

head结点：

• 所有未删除结点都可以head结点通过执行succ()方法访问
• head结点非null
• head结点next不能指向自己
• head结点的item可能为null，也可能不为null
• tail结点可以落后于head结点，此时，从head结点不能访问到tail结点

tail结点（tail的next为null）：

• 队列最后一个结点可以通过tail结点执行succ()方法得访问
• tail结点非null
• tail结点的item可能为null，也可能不为null
• tail结点可以落后于head结点，此时，从head结点不能访问到tail结点
• tail结点next可能指向自己，也可能不指向自己

由于head结点和tail结点不是实时更新，达到松弛度阈值才进行更新，有可能导致head结点在tail结点之后的现象

    /**
     * A node from which the first live (non-deleted) node (if any)
     * can be reached in O(1) time.
     * Invariants:
     * - all live nodes are reachable from head via succ()
     * - head != null
     * - (tmp = head).next != tmp || tmp != head
     * Non-invariants:
     * - head.item may or may not be null.
     * - it is permitted for tail to lag behind head, that is, for tail
     *   to not be reachable from head!
     */
    private transient volatile Node<E> head;

    /**
     * A node from which the last node on list (that is, the unique
     * node with node.next == null) can be reached in O(1) time.
     * Invariants:
     * - the last node is always reachable from tail via succ()
     * - tail != null
     * Non-invariants:
     * - tail.item may or may not be null.
     * - it is permitted for tail to lag behind head, that is, for tail
     *   to not be reachable from head!
     * - tail.next may or may not be self-pointing to tail.
     */
    private transient volatile Node<E> tail;
    
    // CAS操作
    private static final sun.misc.Unsafe UNSAFE;
    private static final long headOffset;
    private static final long tailOffset;
    static {
        try {
            UNSAFE = sun.misc.Unsafe.getUnsafe();
            Class<?> k = ConcurrentLinkedQueue.class;
            headOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("head"));
            tailOffset = UNSAFE.objectFieldOffset
                (k.getDeclaredField("tail"));
        } catch (Exception e) {
            throw new Error(e);
        }
    }

内部类

Node实现比较简单，没复杂的部分，主要是通过CAS操作进行更新变量

    private static class Node<E> {
        volatile E item;
        volatile Node<E> next;

        /**
         * Constructs a new node.  Uses relaxed write because item can
         * only be seen after publication via casNext.
         */
        Node(E item) {
            UNSAFE.putObject(this, itemOffset, item);
        }

        boolean casItem(E cmp, E val) {
            return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
        }

        void lazySetNext(Node<E> val) {
            UNSAFE.putOrderedObject(this, nextOffset, val);
        }

        boolean casNext(Node<E> cmp, Node<E> val) {
            return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
        }

        // Unsafe mechanics

        private static final sun.misc.Unsafe UNSAFE;
        private static final long itemOffset;
        private static final long nextOffset;

        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = Node.class;
                itemOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("item"));
                nextOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("next"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }

构造方法

无参构造方法创建了空结点同时头尾结点指向这个空结点，集合参数构造时循环添加结点，比较简单，主要需要理解默认无参构造函数创建时发生的变化

    public ConcurrentLinkedQueue() {
        head = tail = new Node<E>(null);
    }

    public ConcurrentLinkedQueue(Collection<? extends E> c) {
        Node<E> h = null, t = null;
        for (E e : c) {
            checkNotNull(e);
            Node<E> newNode = new Node<E>(e);
            if (h == null)
                h = t = newNode;
            else {
                t.lazySetNext(newNode);
                t = newNode;
            }
        }
        if (h == null)
            h = t = new Node<E>(null);
        head = h;
        tail = t;
    }

重要方法

updateHead

h != p的前提条件下尝试更新head指向到p，成功则尝试更新原head结点指向到自己，表示结点离队

    /**
     * Tries to CAS head to p. If successful, repoint old head to itself
     * as sentinel for succ(), below.
     */
    final void updateHead(Node<E> h, Node<E> p) {
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }

succ

获取p结点的后继结点，当next指向自己时，结点本身可能已经处于离队状态，此时返回head结点

    /**
     * Returns the successor of p, or the head node if p.next has been
     * linked to self, which will only be true if traversing with a
     * stale pointer that is now off the list.
     */
    final Node<E> succ(Node<E> p) {
        Node<E> next = p.next;
        return (p == next) ? head : next;
    }

offer

入队操作核心方法，入队必成功，返回为true，表示入队会一直尝试操作直到成功，循环尝试中主要分为3种情况：

• 找到最后一个结点，尝试更新next指向新结点，失败则表示next被其他线程更新，此时重新循环判断，成功则判断tail结点指向是否已经滞后一个结点以上，如果是则尝试更新tail
• 之前找到的最后一个结点已经离队（p = p.next），如果tail已经被其他线程更新则更新到tail，否则从head结点开始找到最后一个结点（因为tail可以落后于head）
• 非最后一个结点同时这个结点也未离队，如果tail已经被其他线程更新则更新到tail，否则从当前结点的next开始继续循环

    public boolean offer(E e) {
        // 判空
        checkNotNull(e);
        final Node<E> newNode = new Node<E>(e);

        // 循环直到成功
        for (Node<E> t = tail, p = t;;) {
            Node<E> q = p.next;
            // p此时是最后一个结点
            if (q == null) {
                // 则开始尝试更新p的next指向新插入的结点
                // p的next未更新成功说明next被其他结点抢先更新了，重新循环判断尝试
                if (p.casNext(null, newNode)) {
                    // tail指向结点后已经添加了1个结点以上时才更新tail结点指向
                    // 即slack >= 2时才尝试更新
                    if (p != t) // hop two nodes at a time
                        // 失败可能被其他线程更新了
                        casTail(t, newNode);  // Failure is OK.
                    return true;
                }
                // Lost CAS race to another thread; re-read next
            }
            // p非最后一个结点，同时p = p.next则表示p本身已经离队，需要更新p
            else if (p == q)
                // tail结点已经被更新则更新tail否则从head结点开始寻找最后一个结点
                p = (t != (t = tail)) ? t : head;
            else
                // p非最后一个结点，同时p未离队删除，如果tail被其他线程更新了则p更新成新的tail,否则p更新成p.next继续循环
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

poll

出队操作核心方法，一直尝试直到成功，循环尝试中主要分为4种情况：

• 找到头结点，item非null则尝试更新成null以表示结点已出队，成功则判断是否需要更新head结点返回item
• item为null或item已经被其他线程获取，同时当前结点已经为最后一个结点，则尝试更新头head指向当前结点，返回null
• 当前结点非最后一个结点，如果已经离队则从head重新进行循环
• 当前结点未离队则更新到下一个结点进行循环判断

    public E poll() {
        restartFromHead:
        // 循环尝试直到成功
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                E item = p.item;
                
                // item非null则尝试更新成null(表示结点已出队)
                if (item != null && p.casItem(item, null)) {
                    // 出队结点非之前的头结点，旧头结点h距离实际的head结点已经大于1则更新head
                    if (p != h) // hop two nodes at a time
                        // 出队结点后无结点则尝试更新head结点为出队结点本身（item = null）,有结点则更新到出队结点后的那个结点
                        updateHead(h, ((q = p.next) != null) ? q : p);
                    return item;
                }
                // item为空或item已被其他线程获取
                // p结点本身为最后一个结点，则尝试更新head并更新原h结点指向自己，返回null
                else if ((q = p.next) == null) {
                    updateHead(h, p);
                    return null;
                }
                // p非最后一个结点，p == p.next 则表示p结点已经离队，则跳转restartFromHead从头重新循环判断
                else if (p == q)
                    continue restartFromHead;
                // p非最后一个结点，p也未离队，则更新p指向p的下一个结点，循环判断
                else
                    p = q;
            }
        }
    }

peek

和poll方法类似，主要在于不会对结点进行出队操作，仅仅是获取头结点的item值，当然中间也可能帮助更新下head指向

    public E peek() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                E item = p.item;
                if (item != null || (q = p.next) == null) {
                    updateHead(h, p);
                    return item;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }

first

和poll方法类似，poll返回的是item这里返回的是结点，如果是null结点（item == null）,这里判断下直接返回null

    Node<E> first() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, 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;
                else
                    p = q;
            }
        }
    }

remove

从队列中删除元素，通过item是否为null判断结点是否已经离队，是则继续后继判断，casItem(item, null)成功则表示移除结点成功，失败则表示被其他线程出队操作了，则继续后继判断

    public boolean remove(Object o) {
        if (o != null) {
            Node<E> next, pred = null;
            for (Node<E> p = first(); p != null; pred = p, p = next) {
                boolean removed = false;
                E item = p.item;
                // item判断（非离队结点），不满足则继续判断后继结点
                if (item != null) {
                    if (!o.equals(item)) {
                        next = succ(p);
                        continue;
                    }
                    // 找到匹配结点则尝试更新item为null，表示当前结点已经离队
                    removed = p.casItem(item, null);
                }
                
                // 后继结点，到这说明匹配的结点已经删除了（别的线程删除或者当前线程删除）
                next = succ(p);
                // 取消匹配结点的关联
                if (pred != null && next != null) // unlink
                    pred.casNext(p, next);
                // 假如是当前线程删除的结点则返回，否则继续判断后继直到匹配或没有匹配结点才返回
                if (removed)
                    return true;
            }
        }
        return false;
    }

addAll

将集合c中的元素添加到队列中，添加到原队列尾部类似于上面的offer方法

    public boolean addAll(Collection<? extends E> c) {
        if (c == this)
            // As historically specified in AbstractQueue#addAll
            throw new IllegalArgumentException();

        // 定义两个指针结点指向集合c的头尾
        // 先将c改造成Node链表
        Node<E> beginningOfTheEnd = null, last = null;
        for (E e : c) {
            checkNotNull(e);
            Node<E> newNode = new Node<E>(e);
            if (beginningOfTheEnd == null)
                beginningOfTheEnd = last = newNode;
            else {
                last.lazySetNext(newNode);
                last = newNode;
            }
        }
        if (beginningOfTheEnd == null)
            return false;

        for (Node<E> t = tail, p = t;;) {
            Node<E> q = p.next;
            // p为队列最后一个结点
            if (q == null) {
                // 将队列与上面新创建的链表连接起来，更新失败再循环继续
                if (p.casNext(null, beginningOfTheEnd)) {
                    // tail更新失败重新尝试
                    if (!casTail(t, last)) {
                        t = tail;
                        if (last.next == null)
                            casTail(t, last);
                    }
                    return true;
                }
            }
            // p非最后一个结点且已经离队
            else if (p == q)
                // tail结点已经被更新则更新为tail否则从head结点开始寻找最后一个结点
                p = (t != (t = tail)) ? t : head;
            else
                // p非最后一个结点，同时p未离队删除，如果tail被其他线程更新了则p更新成新的tail,否则p更新成p.next继续循环
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

迭代器

迭代器和之前队列讲解的迭代器相似，源码不是很复杂，同时remove方法这里是将item置为null，前后结点关联关系并不会操作，防止多线程遍历出现问题

构造方法中执行了advance()方法，提前设置好下次next执行时的结点nextNode，以及其item引用，hasNext判断nextNode即可，保证了迭代器的弱一致性，一旦hasNext返回true，那么调用next一定会得到相对应的item，即使该结点item已经被置为null

    public Iterator<E> iterator() {
        return new Itr();
    }

    private class Itr implements Iterator<E> {
        /**
         * next返回的Node
         */
        private Node<E> nextNode;

        /**
         * 保存next的item，防止hasNext为true后结点被删除再调用next获取不到值的情况
         */
        private E nextItem;

        /**
         * 最近一次调用next返回的结点，如果通过调用remove删除了此元素，则重置为null，避免删除了不该删除的元素
         */
        private Node<E> lastRet;

        /**
         * 构造的时候就先保存了第一次调用next返回的Node
         */
        Itr() {
            advance();
        }

        /**
         * Moves to next valid node and returns item to return for
         * next(), or null if no such.
         */
        private E advance() {
            lastRet = nextNode;
            E x = nextItem;

            Node<E> pred, p;
            if (nextNode == null) {
                p = first();
                pred = null;
            } else {
                pred = nextNode;
                p = succ(nextNode);
            }

            for (;;) {
                if (p == null) {
                    nextNode = null;
                    nextItem = null;
                    return x;
                }
                E item = p.item;
                if (item != null) {
                    nextNode = p;
                    nextItem = item;
                    return x;
                } else {
                    // 跳过null结点
                    Node<E> next = succ(p);
                    if (pred != null && next != null)
                        pred.casNext(p, next);
                    p = next;
                }
            }
        }

        public boolean hasNext() {
            return nextNode != null;
        }

        public E next() {
            if (nextNode == null) throw new NoSuchElementException();
            return advance();
        }

        public void remove() {
            Node<E> l = lastRet;
            if (l == null) throw new IllegalStateException();
            // rely on a future traversal to relink.
            l.item = null;
            lastRet = null;
        }
    }

总结

ConcurrentLinkedQueue是一个基于链表实现的无界的线程安全的FIFO非阻塞队列，整体源码上最主要的部分在于两点：

• 全程无锁操作，无阻塞操作，使用CAS更新变量
• head，tail结点非实时更新，在slack >= 2时才进行更新操作

结合图解很容易理清其实现以及操作流程，相比较于之前的LinkedTransferQueue源码算是简单了很多

以上内容如有问题欢迎指出，笔者验证后将及时修正，谢谢