非阻塞队列详解 ConcurrentLinkedQueue详解源码分析 -- 基于JDK 1.8

╰+哭是因爲堅強的太久メ 2021-11-09 05:28 228阅读 0赞

上一篇，我们分析了阻塞队列的内容：

[阻塞队列详解 LinkedBlockingQueue 源码分析 -- 基于JDK 1.8][LinkedBlockingQueue _ -- _JDK 1.8]

现在来看一下非阻塞队列：  
要实现一个线程安全的队列有两种方式：阻塞和非阻塞。前面我们分析了阻塞队列LinkedBlockingQueue和ArrayBlockingQueue， 阻塞队列本质就是锁Reentrantlock和条件Condition等待唤醒操作的结合应用，而我们这次要介绍的非阻塞队列CoucurrentLinkedQueue，它不是阻塞队列，也不能用于线程池中，但是它采用了CAS算法来实现，是线程安全的，可用于多线程环境中；

**目录**

1、继承结构

2、添加元素方法

（1）、构造方法

（2）、添加元素add（offer）方法

（3）、offer方法大致流程图

3、出队元素方法poll

--------------------

# 1、继承结构 #

在继承结构上，与之前的阻塞队列，并无太多变化；唯一区别就是没有实现阻塞队列接口BlockingQueue；从名字上看，就知道是链表的结构，有头节点、尾节点，并且是一个单向链表结构的队列，也作为FIFO，队尾插入，队头取，下面直接从源码开始分析吧：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70][]

# 2、添加元素方法 #

因为它不是阻塞队列，所以只有两个入队的方法，add(e)和offer(e)。同样做了非空校验，并且队列满时，add(e)方法插入失败时，不会抛出异常；

head和tail都可能会滞后，这其实是一种避免频繁CAS的优化。当然过度的滞后也是会影响操作效率的，所以在具体实现的时候，会尽可能能有机会更新head和tail就去更新它们。

## （1）、构造方法 ##

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

## （2）、添加元素add（offer）方法 ##

public boolean add(E e) {
        return offer(e);
    }
    /**
     * 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
    //该方法永远不会返回false，要么空指针异常，要么返回true
     */
    
    public boolean offer(E e) {
    	//同样不能为空元素
        checkNotNull(e);
        final Node<E> newNode = new Node<E>(e);
        //创建新的节点，加入到链表队列的尾部, 循环到队列尾部
        //（a） 
        for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        // q为空，说明p是尾部了，尝试加入到队尾，
        // CAS原子更新p的next为新节点，成功更新返回true
        //（b）
        if (q == null) {
                // （c）
                if (p.casNext(null, newNode)) {
                    //如果p！=t，说明有其它线程先更新到了队尾
                    //需要重新继续更新newNode为尾节点
                    //（d）
                    if (p != t) // hop two nodes at a time
                    //如果casTail执行失败也ok
                    //就先不更新尾节点了；offer方法返回true；
                        casTail(t, newNode);  // Failure is OK.
                    return true;
                }
                    // Lost CAS race to another thread; re-read next
            }
         //并发情况下，有出队线程取元素；
         //此时重新赋值p引用的指针
         //（e）
         else if (p == q)
              p = (t != (t = tail)) ? t : head;
         else
             // Check for tail updates after two hops.
         //（f）
              p = (p != t && t != (t = tail)) ? t : q;
        }
    }
    
    boolean casNext(Node<E> cmp, Node<E> val) {
    //这里调用了UNSAFE类里面的方法
    //简单讲一下这个类。Java无法直接访问底层操作系统，而是通过本地
    //（native）方法来访问。不过尽管如此，JVM还是开了一个后门，JDK
    //中有一个类Unsafe，它提供了硬件级别的原子操作。
    //Unsafe就像这个类名一样，这个类本身就不是很安全，Unsafe类使Java
    //拥有了像C语言的指针一样操作内存空间的能力，同时也带来了指针的问
    //题。过度的使用Unsafe类会使得出错的几率变大，因此Java官方并不
    //建议使用的；
    //也就是说CAS是一条CPU的原子指令，不会造成所谓的数据不一致问题
    //第一个参数o（当前为this）为给定对象，offset（nextOffset）为对象
    //内存的偏移量，通过这个偏移量，
    //迅速定位字段并设置或获取该字段的值，
    //expected（cmp）表示期望值（原始值），x（val）表示要设置的值
    
        return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
    }

## （3）、offer方法大致流程图 ##

a、需要结合图和代码一起看，构造方法创建了一个对象Node；head和tail同时指向第一个Node，初始化的Node他的item为空，next为空；

添加队列元素开始，for循环，指针p和t都指向tail，由于是第一次插入，tail也是head；如上述代码行（a）处：

![20190805090724752.png][]

b、然后Node对象q指向了p的next：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1][]

c、此时q为空，说明p是尾节点，如代码行（b），将p的下一节点指向为新节点：执行代码行（c），p的next用casNext方法指向到新插入节点；

并且如果casNext执行成功了，节点成功插入到链表队列的尾部；

如果casNext不成功，那么for循环又重新执行，直到插入成功；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 2][]

d、代码行（d），在上述方法casNext成功插入新节点以后，如果p！=t，说明有其它线程先更新到了队尾，需要重新继续更新newNode为尾节点;

后面的casTail不管是否执行成功，整个offer方法都可以返回true了，并不是每次成功插入一个新节点后，都总需要更新tail节点；

e、回到代码行（c）处，如果casNext执行成功返回true，那么还需要之后的else if代码块干什么呢？

此时，我们再来插入第二个node节点：for循环重新开始，指针p和t都指向tail；此时执行到代码（b）处判断，q不为空，执行代码行（e）处之后的代码：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 3][]

f、如果p和q相等了，代码行（e）处，（只有在并发时有队列头元素出对列，出队方法重新了更新头节点时，p指针指向到了q节点这种情况）:

p = (t != (t = **tail**)) ? t : **head**;

由于操作运算符顺序执行优先级关系，=最后执行，

该代码等价于如下：p = ( (t != (t = **tail**)) ? t : **head**);

如果t==tail，则让p指向t，否则p指向head；

g、执行走到代码（f）处：p = (p != t && t != (t = **tail**)) ? t : q；

代码等价于如下：p = ((p != t && t != (t = **tail**)) ? t : q)；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 4][]

如果p != t ，上述不满足，则让p指向q，然后从for循环处代码行（a），

q又指向p的next，如此循环，发现又重新与a~d的情况相似了，新的节点可以插入了；

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 5][]

# 3、出队元素方法poll #

public E poll() {
        restartFromHead:
        for (;;) {
    
            //开始时，h和p都指向头节点head
            for (Node<E> h = head, p = h, q;;) {
                E item = p.item;
                //节点的item元素属性值不为空，则将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;
            }
        }
    }

参考贴：[https://www.jianshu.com/p/08e8b0c424c0][https_www.jianshu.com_p_08e8b0c424c0]

[LinkedBlockingQueue _ -- _JDK 1.8]: https://blog.csdn.net/lejustdoit/article/details/98107435
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70]: /images/20211109/db249e91ad0b4b19baf548a99912d437.png
[20190805090724752.png]: /images/20211109/a058f9c707d94515802251e88a047040.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1]: /images/20211109/4511467aa5c4424f8b25eddd06c60fbb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 2]: /images/20211109/94147df54ea4480891f0107eac9a9ce7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 3]: /images/20211109/38e0fc83ed51468c985a84421522ee73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 4]: /images/20211109/4dc1011a11d948f1825343e2d9eaabd4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 5]: /images/20211109/c89bdd45a3a24c7a8c1515d2fa16a162.png
[https_www.jianshu.com_p_08e8b0c424c0]: https://www.jianshu.com/p/08e8b0c424c0