Jdk1.6 JUC源码解析(24)-ConcurrentLinkedQueue

Jdk1.6 JUC源码解析(24)-ConcurrentLinkedQueue

作者：大飞

 

功能简介：

• ConcurrentLinkedQueue是一种基于单向链表实现的无界的线程安全队列。队列中的元素遵循先入先出(FIFO)的规则。新元素插入到队列的尾部，从队列头部取出元素。
• ConcurrentLinkedQueue内部采用一种wait-free(无等待)算法来实现。

 

源码分析：

• 首先看下内部结构：

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable {
    private static final long serialVersionUID = 196745693267521676L;

    private static class Node<E> {
        private volatile E item;
        private volatile Node<E> next;
        Node(E item) {
            // Piggyback on imminent casNext()
            lazySetItem(item);
         } 
        E getItem() {
            return item;
        }
        boolean casItem(E cmp, E val) {
            return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
        }
        void setItem(E val) {
             item = val;
        }
        void lazySetItem(E val) {
            UNSAFE.putOrderedObject(this, itemOffset, val);
        }
        void lazySetNext(Node<E> val) {
            UNSAFE.putOrderedObject(this, nextOffset, val);
        } 
        Node<E> getNext() {
            return next;
        }
        boolean casNext(Node<E> cmp, Node<E> val) {
            return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
        }
        // Unsafe mechanics
        private static final sun.misc.Unsafe UNSAFE =
        sun.misc.Unsafe.getUnsafe();
        private static final long nextOffset =
        objectFieldOffset(UNSAFE, "next", Node.class);
        private static final long itemOffset =
        objectFieldOffset(UNSAFE, "item", Node.class);
    }
    /**
    * 表示第一个存活的节点(头节点)，访问此节点时间复杂度为O(1)。
    * Invariants:
    * - 所有存活的节点都能通过头结点到达。
    * Non-invariants:
    * - 头节点的元素可以为null。
    * - 可以允许尾节点比头节点滞后，也就是说，可能从头节点无法到达尾节点。
    */
    private transient volatile Node<E> head = new Node<E>(null);
    /**
    * 队列的尾节点(唯一的next为null的节点)，访问此节点时间复杂度为O(1)。
    * Invariants:
    * - 最后的节点可以通过对tail调用succ()方法访问到。
    * Non-invariants:
    * - 尾节点的元素可以为null。
    * - 可以允许尾节点比头节点滞后，也就是说，可能从头节点无法到达尾节点。
    * - 尾节点的next可能指向自身。
    */ 
    private transient volatile Node<E> tail = head;

    public ConcurrentLinkedQueue() {}

       可见，内部并没有锁，只有一个头节点和一个尾节点。初始时头节点和尾节点都指向同一节点。

 

• 接下来看一下添加元素到队列的方法：

    public boolean add(E e) {
        return offer(e);
    }
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        Node<E> n = new Node<E>(e);
        retry:
        for (;;) {
            Node<E> t = tail;
            Node<E> p = t;
            for (int hops = 0; ; hops++) {
                Node<E> next = succ(p); // 1.获取p的后继节点。(如果p的next指向自身，返回head节点)
                if (next != null) { // 2.如果next不为null
                    if (hops > HOPS && t != tail) 
                        continue retry; // 3.如果自旋次数大于HOPS，且t不是尾节点，跳出2层循环重试。
                    p = next; // 4.如果自旋字数小于HOPS或者t是尾节点，将p指向next。
                } else if (p.casNext(null, n)) { // 5.如果next为null，尝试将p的next节点设置为n，然后自旋。
                    if (hops >= HOPS)
                        casTail(t, n); // 6.如果设置成功且自旋次数大于HOPS，尝试将n设置为尾节点，失败也没关系。 
                    return true; // 7.添加成功。
                } else {
                    p = succ(p); // 8。如果第5步尝试将p的next节点设置为n失败，那么将p指向p的后继节点，然后自旋。
                }
            }
        }
    }
     final Node<E> succ(Node<E> p) {
         Node<E> next = p.getNext();
         //如果p节点的next节点指向自身，那么返回head节点；否则返回p的next节点。
         return (p == next) ? head : next;
     }
    /**
     * 允许头尾节点的指针滞后，所以当头尾节点离"实际位置"的距离 
     * (按节点)小于HOPS时，不会去更新头尾指针。这里是假设volatile写代价比volatile读高。
     */
     private static final int HOPS = 1;

 

       简单分析一下offer方法里的过程：

              1.假设没有竞争的情况下，初始状态时head和tail都指向一个节点，这时来了一个新节点n1，代码走向上面注释第1行，得到的next为null，然后走向第5行，然后将p(也就是tail节点)的next节点设置为n1，成功返回。注意这里并没有调整tail指针，head和tail还是指向之前的节点。然后再来一个新节点n2，还是走向第1行，得到的next是n1，然后走向第8行，将p指向n1，自旋一次，又来到第1行，得到的next为null，然后走向第5行，将p(也就是n1)的next节点设置为n2。由于已经自旋一次，所以这时还会走向第6行，将tail指向n2。

             2.有竞争的情况也很好分析，假设在注释第5行竞争失败，那么会走向第8行，将p向队尾推进一步，然后重试；如果重试了多次后(超过一次)，在注释第一行获取的next仍然不为空，且同时尾节点也被其他线程推进了，那说明当前节点离尾节点太远了，跳出循环，重新定位尾节点然后再试，这样可以减少自旋次数。

 

• 再看一下从队列中获取元素的方法：

    public E poll() {
        Node<E> h = head;
        Node<E> p = h;
        for (int hops = 0; ; hops++) {
            E item = p.getItem(); // 1.获取p节点上的元素item。
            if (item != null && p.casItem(item, null)) { // 2.如果item不为null，尝试将p的item设置为null。
                if (hops >= HOPS) { 
                    Node<E> q = p.getNext();
                    updateHead(h, (q != null) ? q : p); // 3.如果自旋次数大于HOPS，尝试更新头节点。
                }
                return item; // 4.获取元素成功。
            }
            Node<E> next = succ(p); // 5.获取p的后继节点。(如果p的next指向自身，那么返回head节点)
            if (next == null) {
                updateHead(h, p); // 6.如果p的后继节点为null，尝试将p设置为头节点，然后跳出循环。
                break;
            }
            p = next;
        }
        return null; // 7.从第6步过来，没有成功获取元素，返回null。
    }

     final void updateHead(Node<E> h, Node<E> p) {
         if (h != p && casHead(h, p))
             h.lazySetNext(h); //将h的next指向自身，帮助GC。
     }

       和offer方法一样的思路。

       再看下peek方法，只获取元素，不移除节点： 

    public E peek() {
        Node<E> h = head;
        Node<E> p = h;
        E item; 
        for (;;) {
            item = p.getItem();
            if (item != null)
                break;
            Node<E> next = succ(p);
            if (next == null) {
                break;
            }
            p = next;
        }
        updateHead(h, p);
        return item;
    }

 

• 继续看一下其他方法：

 

       看其他方法之前，先看一个内部支持方法first：

     Node<E> first() {
         Node<E> h = head;
         Node<E> p = h;
         Node<E> result;
         for (;;) { 
             E item = p.getItem();
             if (item != null) {
                 result = p;
                 break;
             }
             Node<E> next = succ(p);
             if (next == null) {
                 result = null;
                 break;
             }
             p = next;
         }
         updateHead(h, p); // 这里会帮助推进一下头节点。
         return result;
     }

       可见，和peek里面的逻辑基本一致。但之所以没有利用first来实现peek，而是单独写peek，是因为可以减少一次volatile读。 

 

    public boolean isEmpty() {
        return first() == null;
    }
    /**
     * 注意这个方法内部使用遍历实现，时间复杂度为O(n)。
     */
    public int size() {
        int count = 0;
        for (Node<E> p = first(); p != null; p = succ(p)) {
            if (p.getItem() != null) {
                // Collections.size() spec says to max out
                if (++count == Integer.MAX_VALUE)
                    break;
            }
        }
        return count;
    }

    public boolean contains(Object o) {
        if (o == null) return false;
        for (Node<E> p = first(); p != null; p = succ(p)) {
            E item = p.getItem();
            if (item != null &&
                o.equals(item))
                return true;
        }
        return false;
    }

    public boolean remove(Object o) {
        if (o == null) return false;
        Node<E> pred = null;
        for (Node<E> p = first(); p != null; p = succ(p)) {
            E item = p.getItem();
            if (item != null &&
                o.equals(item) &&
                p.casItem(item, null)){
                Node<E> next = succ(p);
                if(pred != null && next != null)
                   pred.casNext(p, next);
                return true;
            }
            pred = p;
        }
        return false;
    }

    public Object[] toArray() {
        // Use ArrayList to deal with resizing.
        ArrayList<E> al = new ArrayList<E>();
        for (Node<E> p = first(); p != null; p = succ(p)) {
            E item = p.getItem();
            if (item != null)
                al.add(item);
        }
        return al.toArray();
    }

    @SuppressWarnings("unchecked")
    public <T> T[] toArray(T[] a) {
        // try to use sent-in array
        int k = 0;
        Node<E> p;
        for (p = first(); p != null && k < a.length; p = succ(p)) {
            E item = p.getItem();
            if (item != null)
                a[k++] = (T)item;
        }
        if (p == null) {
            if (k < a.length)
                a[k] = null;
            return a;
        }
        // If won't fit, use ArrayList version
        ArrayList<E> al = new ArrayList<E>();
        for (Node<E> q = first(); q != null; q = succ(q)) {
            E item = q.getItem();
            if (item != null)
                al.add(item);
        }
        return al.toArray(a);
    }

        这些方法都是基于first方法实现，代码很容易看懂，这里就不啰嗦了。

 

• 最后，ConcurrentLinkedQueue的迭代器也是弱一致的。

 

       小总结一下：

              ConcurrentLinkedQueue内部在插入或者移除元素时，不会即时设置头尾节点，而是有一个缓冲期(一个节点长的距离)，这样能减少一些CAS操作；其次在设置头尾节点的时候，也不会CAS-Loop直到成功，只尝试一次，失败也没关系，下一次操作或者其他线程在操作时会帮忙推进头尾节点(将头尾指针指向正确位置)。

 

       ConcurrentLinkedQueue的代码解析完毕！