源码解读之（八）ConcurrentLinkedQueue

小咪咪 2022-12-05 02:11 256阅读 0赞

### 源码解读之（八）ConcurrentLinkedQueue ###

*   *  一、前言
     *  二、ConcurrentLinkedQueue介绍与结构
     *  三、offer操作
     *  四、poll操作
     *  五、peek操作
     *  六、size操作
     *  七、remove操作
     *  八、ConcurrentLinkedQueue遇到的问题
     *  九、总结

## 一、前言 ##

在并发编程中我们有时候需要使用线程安全的队列。如果我们要实现一个线程安全的队列有两种实现方式：一种是使用`阻塞算法`，另一种是使用`非阻塞算法`。使用阻塞算法的队列可以用一个`锁（入队和出队用同一把锁）或两个锁（入队和出队用不同的锁）`等方式来实现，而非阻塞的实现方式则可以使用循环`CAS`的方式来实现，本文让我们一起来研究下`Doug Lea`是如何使用非阻塞的方式来实现线程安全队列`ConcurrentLinkedQueue`的，相信从大师身上我们能学到不少并发编程的技巧。

## 二、ConcurrentLinkedQueue介绍与结构 ##

**1、ConcurrentLinkedQueue整体属性与内部类**  
![在这里插入图片描述][20200908205609985.png]  
其中有一个很重要的内部类`Node`

![在这里插入图片描述][20200908205926959.png]

// 用于存放节点的值
    volatile E item;
    // 指向下一个节点
    volatile Node<E> next;
    // 这里也是用的是UNSAFE类，这个类直接提供CAS操作
    private static final sun.misc.Unsafe UNSAFE;
    // item字段的偏移量
    private static final long itemOffset;
    // next的偏移量
    private static final long nextOffset;

**2、ConcurrentLinkedQueue类图结构**

![在这里插入图片描述][2020090821004854.png]

`ConcurrentLinkedQueue`中有两个`volatile`类型的`Node`节点分别用来存放列表的`首尾节点`，其中head节点存放链表第一个`item`为`null`的节点，`tail`则并不是总指向最后一个节点。`Node`节点内部则维护一个变量`item`用来存放节点的值，`next`用来存放下一个节点，从而链接为一个`单向无界列表`。

注意，默认情况下`头节点`和`尾节点`都是`哨兵节点`，也就是一个存`null`的`Node`节点。

// 存放链表的头节点
    private transient volatile Node<E> head;
    // 存放链表的尾节点
    private transient volatile Node<E> tail;
    // UNSAFE对象
    private static final sun.misc.Unsafe UNSAFE;
    // head字段的偏移量
    private static final long headOffset;
    // tail字段偏移量
    private static final long tailOffset;

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

如上构造函数初始化时候会构建一个`item`为`NULL`的空节点作为链表的首尾节点。

## 三、offer操作 ##

这个方法的作用就是在队列末端添加一个节点，如果传递的参数是`null`，就抛出空指针异常，否则由于该队列是无界队列，该方法会一直返回`true`，而且该方法使用`CAS`算法实现的，所以不会阻塞线程；

// 队列末端添加一个节点
    public boolean offer(E e) { 
    	// 如果e为空，那么抛出空指针异常
        checkNotNull(e);
        // 将传进来的元素封装成一个节点，Node的构造器中调用UNSAFE.putObject(this, itemOffset, item)把e赋值给节点中的item
        final Node<E> newNode = new Node<E>(e);
    	// [1] 这里的for循环从最后的节点开始
        for (Node<E> t = tail, p = t;;) { 
            Node<E> q = p.next;
            // [2] 如果q为null，说明p就是最后的节点了
            if (q == null) { 
                // [3] CAS更新：如果p节点的下一个节点是null，就把写个节点更新为newNode
                if (p.casNext(null, newNode)) { 
                    // [4] CAS成功，但是这时p==t,所以不会进入到这里的if里面，直接返回true
                    // 那么什么时候会走到这里面来呢？其实是要有另外一个线程也在调用offer方法的时候，会进入到这里面来
                    if (p != t) 
                        casTail(t, newNode); 
                    return true;
                }
            }
            else if (p == q) // [5]
                p = (t != (t = tail)) ? t : head;
            else // [6]
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

1、上面执行到`[3]`的时候，由于头节点和尾节点默认都是指向哨兵节点的，由于这个时候`p`的下一个节点为`null`，所以当前线程`A`执行`CAS`会成功，下图所示；

![在这里插入图片描述][20200908212219779.png]

2、如果此时还有一个线程`B`也来尝试`[3]`中`CAS`，由于此时`p`节点的下一个节点不是`null`了，于是线程`B`会跳到`[1]`出进行第二次循环，然后会到`[6]`中，由于`p`和`t`此时是相等的，所以这里是`false`，即`p=q`，下图所示：

![在这里插入图片描述][20200908212401708.png]

3、然后线程`B`又会跳到`[1]`处进行第三次循环，由于执行了`Node<E> q = p.next`，所以此时`q`指向最后的`null`，就到了`[3]`处进行`CAS`，这次是可以成功的，成功之后如下图所示：

![在这里插入图片描述][20200908212512499.png]

4、这个时候因为`p!=t`，所以可以进入到`[4]`，这里又会进行一个`CAS`：如果`tail`和`t`指向的节点一样，那么就将tail指向新添加的节点，如图所示，这个时候线程B也就执行完了；

![在这里插入图片描述][20200908212650428.png]

5、其实还有`[5]`没有走到，这个是在`poll`操作之后才执行的，我们先跳过，等说完`poll`方法之后再回头看看；另外说一下，`add`方法其实就是调用的是`offer`方法，就不多说了；

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

## 四、poll操作 ##

这个方法是获取头部的这个节点，如果队列为空则返回`null`。

public E poll() { 
    	// 这里其实就是一个goto的标记，用于跳出for循环
        restartFromHead:
        // [1]
        for (;;) { 
            for (Node<E> h = head, p = h, q;;) { 
                E item = p.item;
    			// [2] 如果当前节点中存的值不为空，则CAS设置为null
                if (item != null && p.casItem(item, null)) { 
                    // [3] CAS成功就更新头节点的位置
                    if (p != h) 
                        updateHead(h, ((q = p.next) != null) ? q : p);
                    return item;
                }
                // [4] 当前队列为空，就返回null
                else if ((q = p.next) == null) { 
                    updateHead(h, p);
                    return null;
                }
                // [5] 当前节点和下一个节点一样，说明节点自引用，则重新找头节点
                else if (p == q)
                    continue restartFromHead;
                // [6] 
                else
                    p = q;
            }
        }
    }

final void updateHead(Node<E> h, Node<E> p) { 
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }

**分为几种情况：**

第一种情况是线程`A`调用`poll`方法的时候，发现队列是空的，即头节点和尾节点都指向哨兵节点，就会直接到`[4]`，返回`null`。

第二种情况，线程`A`执行到了`[4]`，此时有一个线程却调用`offer`方法添加了一个节点，下图所示，那么此时线程`A`就不会走`[4]`了，`[5]`也不满足，于是会到`[6]`这里来，然后线程`A`又会跳到`[1]`处进行循环，此时`p`指向的节点中`item`不为`null`，所以会到`[2]`中；

![在这里插入图片描述][2020090821414466.png]

到了这里还没完，还记不记得`offer`方法中有一个地方的代码没有执行的啊！就是这种情况，尾节点自己引用自己，我们再调用`offer`会怎么样呢？

回到offer方法，先会到`[1]`，然后`q`指向自己这个哨兵节点（注意，此时虽然`p`指向的节点中存的是`null`，但是`p!=null`\}，于是再到`[5]`，此时的图如下左图所示；此时由于`t==tail`，所以`p=head`；

![在这里插入图片描述][20200908214443293.png]

再在`offer`方法循环一次，此时`q`指向`null`，下面左图所示，然后就可以进入`[2]`中进行`CAS`，`CAS`成功，因为此时`p!=t`，所以还要进行`CAS`将`tail`指向新节点，下面右图所示，可以让`GC`回收那个垃圾！

![在这里插入图片描述][20200908214556500.png]

## 五、peek操作 ##

这个方法的作用就是获取队列头部的元素，只获取不移除，注意这个方法和上面的`poll`方法的区别啊！

public E peek() { 
    	// [1] goto标志
        restartFromHead:
        for (;;) { 
            for (Node<E> h = head, p = h, q;;) { 
            	// [2]
                E item = p.item;
                // [3]
                if (item != null || (q = p.next) == null) { 
                    updateHead(h, p);
                    return item;
                }
                // [4]
                else if (p == q)
                    continue restartFromHead;
                // [5]
                else
                    p = q;
            }
        }
    }
    
    final void updateHead(Node<E> h, Node<E> p) { 
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }

1、如果队列中为空的时候，走到`[3]`的时候，就会如下图所示，由于`h==p`,所以`updateHead`方法啥也不做，然后返回就返回`item`为`null`。

![在这里插入图片描述][20200908215315952.png]

2、如果队列不为空，那么如下左图所示，此时进入循环内不满足条件，会到`[5]`这里，将`p`指向`q`，然后再进行一次循环到`[3]`，将`q`指向`p`的后一个节点，下面右图所示；

![在这里插入图片描述][20200908215406840.png]

3、然后调用`updateHead`方法，用`CAS`将头节点指向`p`这里，然后将`h`自己指向自己，下图所示，最后返回`item`

![在这里插入图片描述][20200908215453935.png]

## 六、size操作 ##

获取当前队列元素个数，在并发环境下不是很有用，因为使用`CAS`没有加锁，所以从调用`size`函数到返回结果期间有可能增删元素，导致统计的元素个数不精确。

public int size() { 
        int count = 0;
        for (Node<E> p = first(); p != null; p = succ(p))
            if (p.item != null)
                // 最大返回Integer.MAX_VALUE
                if (++count == Integer.MAX_VALUE)
                    break;
        return count;
    }
    
    // 获取第一个队列元素（哨兵元素不算），没有则为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;
            }
        }
    }
    
    // 获取当前节点的next元素，如果是自引入节点则返回真正头节点
    final Node<E> succ(Node<E> p) { 
        Node<E> next = p.next;
        return (p == next) ? head : next;
    }

## 七、remove操作 ##

如果队列里面存在该元素则删除给元素，如果存在多个则删除第一个，并返回`true`，否者返回`false`。

public boolean remove(Object o) { 
    	// 查找元素为空，直接返回false
        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;
                // 相等则使用cas值null,同时一个线程成功，失败的线程循环查找队列中其他元素是否有匹配的。
                if (item != null) { 
                    if (!o.equals(item)) { 
                        // 获取next元素
                        next = succ(p);
                        continue;
                    }
                    removed = p.casItem(item, null);
                }
    
                next = succ(p);
                // 如果有前驱节点，并且next不为空则链接前驱节点到next,
                if (pred != null && next != null) 
                    pred.casNext(p, next);
                if (removed)
                    return true;
            }
        }
        return false;
    }

## 八、ConcurrentLinkedQueue遇到的问题 ##

这边`debugger`调试`ConcurrentLinkedQueue`遇到的一个坑！！！

测试主类：

public class ConcurrentLinkedQueueLocalTest { 
        public static void main(String[] args) { 
            ConcurrentLinkedQueueLocal queue = new ConcurrentLinkedQueueLocal();
            queue.offer(1);
            queue.offer(2);
        }
    }

复制源码到本地并重命名为`ConcurrentLinkedQueueLocal`：

其中第二个分支和第三个分支我加了日志打印输出

@Override
    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) { 
                // 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) { 
                System.out.println("第二次走了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 { 
                System.out.println("第二次走了最后的else分支");
                // Check for tail updates after two hops.
                p = (p != t && t != (t = tail)) ? t : q;
            }
        }
    }

下面问题来了！！！

**1、直接run方式启动**

第二次走了最后的else分支

**2、debugger方式启动**

第二次走了p==q分支

不信你可以自己调试下看看，很迷啊这操作！！！把我整了半天还是解决了。。。

**3、解决方式**

这是idea的坑，我们把这两个框不要选。  
![在这里插入图片描述][20200908222024462.png]  
参考链接：[http://www.nopapp.com/Ask/Question/1123839496278904832][http_www.nopapp.com_Ask_Question_1123839496278904832]

代码测试地址：[https://github.com/riemannChow/LeetCode/tree/master/src/main/java/com/test/ConcurrentLinkedQueueTest][https_github.com_riemannChow_LeetCode_tree_master_src_main_java_com_test_ConcurrentLinkedQueueTest]

## 九、总结 ##

`ConcurrentLinkedQueue`使用`CAS`非阻塞算法实现，使用`CAS`解决了当前节点与`next`节点之间的安全链接和对当前节点值的赋值。由于使用`CAS`没有使用锁，所以获取`size`的时候有可能进行`offer`，`poll`或者`remove`操作，导致获取的元素个数不精确，所以在并发情况下`size`函数不是很有用。另外第一次`peek`或者`first`时候会把`head`指向第一个真正的队列元素。

下面总结下如何实现线程安全的，可知入队出队函数都是操作`volatile`变量：`head`，`tail`。所以要保证队列线程安全只需要保证对这两个`Node`操作的可见性和原子性，由于`volatile`本身保证可见性，所以只需要看下多线程下如果保证对着两个变量操作的原子性。

对于`offer`操作是在`tail`后面添加元素，也就是调用`tail.casNext`方法，而这个方法是使用的`CAS`操作，只有一个线程会成功，然后失败的线程会循环一下，重新获取`tail`，然后执行`casNext`方法。对于`poll`也是这样的。

最后找的一张图方便帮助大家记忆：

![在这里插入图片描述][20200908220412722.png]

[20200908205609985.png]: /images/20221123/cdf4aecaafb74b88a07a32dd2431daf4.png
[20200908205926959.png]: /images/20221123/505cc08bddc44be1861e24ace040a36e.png
[2020090821004854.png]: /images/20221123/5868d77c74f44509bdb9729b8fa5bbf1.png
[20200908212219779.png]: /images/20221123/50619d47324143fcad3df9fbdea9f3b3.png
[20200908212401708.png]: /images/20221123/8d8dd28c8fe146fcaced2ddd7ca13ea2.png
[20200908212512499.png]: /images/20221123/1bc2c22ef8d9425a8b22085aa7649711.png
[20200908212650428.png]: /images/20221123/39aa5deac10044d390bac245de6251ce.png
[2020090821414466.png]: /images/20221123/48b4bd5afa93433db66924c74633d08c.png
[20200908214443293.png]: /images/20221123/d07202a4f7af4d4492daddfdee810d0b.png
[20200908214556500.png]: /images/20221123/25b445c169f1408d9218c46e199f1cac.png
[20200908215315952.png]: /images/20221123/648c44624c304d4e981609978b5d7836.png
[20200908215406840.png]: /images/20221123/e81f15e609db474d8e7fe0b751842782.png
[20200908215453935.png]: /images/20221123/54ec4e546814400c8be59fdf8f4779ee.png
[20200908222024462.png]: /images/20221123/93cdcda98ad84b38897f8cce58878880.png
[http_www.nopapp.com_Ask_Question_1123839496278904832]: http://www.nopapp.com/Ask/Question/1123839496278904832
[https_github.com_riemannChow_LeetCode_tree_master_src_main_java_com_test_ConcurrentLinkedQueueTest]: https://github.com/riemannChow/LeetCode/tree/master/src/main/java/com/test/ConcurrentLinkedQueueTest
[20200908220412722.png]: /images/20221123/78d5892f0b4b450ba6c4545d29281288.png