juc 下的集合之三 (ConcurrentLinkedQueue)

一、基本思想

首先是它是线程安全的，其次基于链的节点的，然后是没有容量限制，最后一个是它是一个队列。

 

二、源码解析

2.1 内部类

里面相对来说比较简单，没有什么多余的东西。

private static class Node<E> //节点的内部类
private class Itr implements Iterator<E>  //迭代器的内部类

简单说下Node这个内部类，因为只是一个普通的队列也就是完成先进先出就行，所以我们只需要每个节点知道下一个节点是谁就行，而不需要知道自己的上一个是谁，不存在从后面到前面的遍历。

 private static class Node<E> {
        volatile E item;
        volatile Node<E> next;//只有next
        ......

 

2.2 变量

private transient volatile Node<E> head;
//队列的头
 private transient volatile Node<E> tail;
//队列的尾

 

2.3 添加一个元素

有两个添加的方法add, offer，在这里这两个方法没有区别。

  public boolean add(E e) {
        return offer(e);
  }

 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;
            if (q == null) { //直接遍历最后一个，也就是说他的next为空
                // p is last node
                if (p.casNext(null, newNode)) {//见后面解释
                    // Successful CAS is the linearization point
                    // for e to become an element of this queue,
                    // and for newNode to become "live".
                    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
            }
            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 = (t != (t = tail)) ? t : head;
            else
                // Check for tail updates after two hops.
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

 这个方法的核心是casNext这个函数，函数调用的Node内的无锁原语的函数

p.casNext(null, newNode)
boolean casNext(Node<E> cmp, Node<E> val) {
     return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}如果继续跟踪的话发现是一个Native的方法，到这里脑袋是一塌糊涂。

要解释清楚需要理解CAS原语，CAS是现代CPU支持的一个原语，大意是这样的，我希望内存地址上*addr的值是我希望的值expected,是的话就把想修改的值value赋到地址上。说的通俗点的就是，地址0x22222上的值原来等于1，现在你想修改成1*2=2的情况，因为你让CPU帮你做事的时候，可能地址上的值已经成3了，如果你再*2就是6了，结果超出你的想象了，所以你就说CPU麻烦你帮我check下，0x22222上的值是不是1，如果是的话帮我改成2，如果不是请通知我。就这样。

 bool CAS(T* addr, T expected, T value) {  
   if (*addr == expected) {  
      *addr = value;  
      return true;  
   }  
   return false;  
}   

 套到我们的类中，就有了因为q=p.next=null，当我们真正去把p.next=new value的时候其实p=next有可能被其它线程赋过值了，所以我们需要CPU来帮忙看看q是不是null如果是的话就存入我的新值，如果不是那就返回false。这个我们就可以在多线程环境下完成对数据的安全访问了。

 

2.4 获取头元素

有了上面对CAS的理解，下面的方法就简单了。无非就是看下现在的第一元素是不是第一个，如果是就返回，然后把下一个元素作为新的头。：）PS:昨天去怀旧了下巫妖王，永远要有一个巫妖王否则天灾军团要横扫这个世界的，所以就算阿尔萨斯倒下了，大领主就成了下一个巫妖王了。队列也一样，总要有一个头的。

public E poll() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                E item = p.item;

                if (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;
                }
                else if ((q = p.next) == null) {
                    updateHead(h, p);
                    return null;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }

 

三、适用范围

线程安全，不限制容量的东西听起来很高端大气上档次，但是要考虑好，你所有的资源都是有限的，内存CPU，如果你想固定你的size，或者随时返回可控的大小，不好意思做不多，任何数据结构都不是万能的，根据自己的需要去选择，另，选老婆也是如此。

 

四、测试

没时间呀，没时间。