ConcurrentLinkedQueue线程安全的队列源码解析

刺骨的言语ヽ痛彻心扉 2022-02-26 02:44 300阅读 0赞

## ConcurrentLinkedQueue简介 ##

在并发编程中我们有时候需要使用线程安全的队列。如果我们要实现一个线程安全的队列有两种实现方式一种是使用阻塞算法，另一种是使用非阻塞算法。使用阻塞算法的队列可以用一个锁（入队和出队用同一把锁）或两个锁（入队和出队用不同的锁）等方式来实现，而非阻塞的实现方式则可以使用循环CAS的方式来实现。

`ConcurrentLinkedQueue`是一个基于链接节点的无界线程安全队列，它采用先进先出的规则对节点进行排序，当我们添加一个元素的时候，它会添加到队列的尾部，当我们获取一个元素时，它会返回队列头部的元素。本文就看看JDK中无锁方式实现的线程安全队列。

可以通过参考网上`ConcurrentLinkedQueue`的类图，很好的了解其结构：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70]  
`ConcurrentLinkedQueue`由`head`节点和`tair`节点组成，每个节点（`Node`）由`节点元素（item）`和`指向下一个节点的引用(next)`组成，节点与节点之间就是通过这个next关联起来，从而组成一张链表结构的队列。默认情况下head节点存储的元素为空，`tair`节点在初始化状态下等于`head`节点。

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

## 源码解析： ##

#### 一、入队 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 1]

*  第一步添加元素1。队列更新head节点的next节点为元素1节点。又因为tail节点默认情况下等于head节点，所以它们的next节点都指向元素1节点。
 *  第二步添加元素2。队列首先设置元素1节点的next节点为元素2节点，然后更新tail节点指向元素2节点。
 *  第三步添加元素3，设置tail节点的next节点为元素3节点。
 *  第四步添加元素4，设置元素3的next节点为元素4节点，然后将tail节点指向元素4节点。

这边主要看一下入队的源码：

public boolean offer(E e) { 
    
            if (e == null) throw new NullPointerException();
    
            //入队前，创建一个入队节点
            Node</e><e> n = new Node</e><e>(e);
    
            retry:
    
            //死循环，入队不成功反复入队。
            for (;;) { 
    
                //创建一个指向tail节点的引用
                Node</e><e> t = tail;
    
                //p用来表示队列的尾节点，默认情况下等于tail节点。
                Node</e><e> p = t;
    
                for (int hops = 0; ; hops++) { 
    
                //获得p节点的下一个节点。
    
                     Node</e><e> next = succ(p);
    
                    //next节点不为空，说明p不是尾节点，需要更新p后在将它指向next节点
    
                    if (next != null) { 
    
                       //循环了两次及其以上，并且当前节点还是不等于尾节点
    
                        if (hops > HOPS && t != tail)
    
                            continue retry;
    
                        p = next;
    
                    }
    
                    //如果p是尾节点，则设置p节点的next节点为入队节点。
    
                    else if (p.casNext(null, n)) { 
    
                      //如果tail节点有大于等于1个next节点，则将入队节点设置成tair节点，更新失败了也没关系，因为失败了表示有其他线程成功更新了tair节点。
    
                    if (hops >= HOPS)
    
                            casTail(t, n); // 更新tail节点，允许失败
    
                        return true;
    
                    }
    
                   // p有next节点,表示p的next节点是尾节点，则重新设置p节点
    
                    else { 
    
                        p = succ(p);
    
                    }
    
                }
    
            }
    
        }

从源代码角度来看整个入队过程主要做两件事情。第一是定位出尾节点，第二是使用CAS算法能将入队节点设置成尾节点的next节点，如不成功则重试。

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

#### 一、出队 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 2]  
从上图可知，并不是每次出队时都更新head节点，当head节点里有元素时，直接弹出head节点里的元素，而不会更新head节点。只有当head节点里没有元素时，出队操作才会更新head节点。这种做法也是通过hops变量来减少使用CAS更新head节点的消耗，从而提高出队效率。

出队源码：

public E poll() { 
     
               Node</e><e> h = head;
     
                // p表示头节点，需要出队的节点
     
               Node</e><e> p = h;
     
               for (int hops = 0;; hops++) { 
     
                    // 获取p节点的元素
     
                    E item = p.getItem();
     
                    // 如果p节点的元素不为空，使用CAS设置p节点引用的元素为null,如果成功则返回p节点的元素。
     
                    if (item != null && p.casItem(item, null)) { 
    
                         if (hops >= HOPS) { 
     
                              //将p节点下一个节点设置成head节点
     
                              Node</e><e> q = p.getNext();
     
                              updateHead(h, (q != null) ? q : p);
     
                         }
     
                         return item;
     
                    }
     
                    // 如果头节点的元素为空或头节点发生了变化，这说明头节点已经被另外一个线程修改了。那么获取p节点的下一个节点
     
                    Node</e><e> next = succ(p);
     
                    // 如果p的下一个节点也为空，说明这个队列已经空了
     
                    if (next == null) { 
     
                    // 更新头节点。
    
                         updateHead(h, p);
     
                         break;
     
                    }
     
                    // 如果下一个元素不为空，则将头节点的下一个节点设置成头节点
     
                    p = next;
     
               }
     
               return null;
     
         }

首先获取头节点的元素，然后判断头节点元素是否为空，如果为空，表示另外一个线程已经进行了一次出队操作将该节点的元素取走，如果不为空，则使用CAS的方式将头节点的引用设置成null，如果CAS成功，则直接返回头节点的元素，如果不成功，表示另外一个线程已经进行了一次出队操作更新了head节点，导致元素发生了变化，需要重新获取头节点。

想更详细了解`ConcurrentLinkedQueue`可以到《并发编程网》中学习。  
参考链接：[http://ifeve.com/concurrentlinkedqueue/][http_ifeve.com_concurrentlinkedqueue]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70]: /images/20220226/e33de763a1da49a6aa480aea457e2998.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 1]: /images/20220226/1a7aaa0399db466fb66eb67445729d0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNDA0Mzk1_size_16_color_FFFFFF_t_70 2]: /images/20220226/3991a46d92974492a2072a12e976d458.png
[http_ifeve.com_concurrentlinkedqueue]: http://ifeve.com/concurrentlinkedqueue/