java并发队列之非阻塞队列-ConcurrentLinkedQueue

太过爱你忘了你带给我的痛 2022-04-01 12:41 743阅读 0赞

## 前言 ##

前面我们讲了阻塞队列，阻塞队列使用put/take方法可以实现在队列已满或空的时候达到线程阻塞状态，阻塞这种方式在线程并发时固然安全，但是也会造成效率上的问题，所以说今天我们来讲一个非阻塞队列——ConcurrentLinkedQueue，他能保证并发安全，而且还可以提高效率。

## 正文 ##

通常 ConcurrentLinkedQueue 的性能好于 BlockingQueue。它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，也就是插入的时候往尾部插，取出元素从头部取。该队列不允许null元素。

ConcurrentLinkedQueue内部是遵循CAS（比较并交换）的方式来实现。想了解CAS原理，请看我之前文章——[深入理解CAS][CAS]

### **ConcurrentLinkedQueue重要方法:** ###

**add() 和offer() 都是加入元素的方法(在ConcurrentLinkedQueue中这俩个方法没有任何区别，不要看网上很多博客说不推荐使用add方法，说队列满了插入会抛异常，不要相信，下面贴出源码，add方法调用的就是offer，所以这两个没区别)**

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

并且官方文档如下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70][]  
**poll() 和peek() 都是取头元素节点，区别在于前者会删除元素，后者不会。**

下面我要对队列的入队和出队的方法进行讲解

### 一、入队 ###

![1018541-20170313214020416-738374759.jpg][]

元素插入需要做两件事

第一是将入队节点设置成当前队列的最后一个节点。

第二是更新tail节点，如果**原来的tail**节点的next节点不为空，则将tail更新为刚入队的节点（即队尾结点），如果原来的tail节点为空，则tail节点不动，把元素插入到tail的next节点处。也就是说每次tail移动都要隔着一个节点。

public boolean offer(E e) {
            //首先入队的对象不允许为null
            checkNotNull(e);
            //入队前，创建一个入队节点，构造一个内部函数
            final Node<E> newNode = new Node<E>(e);
            //死循环，入队不成功反复入队。
            for (Node<E> t = tail, p = t;;) {
                 //创建一个指向tail节点的引用
                Node<E> q = p.next;
                //如果q=null说明p是尾节点则插入
                if (q == null) {
                    // cas插入
                    if (p.casNext(null, newNode)) {
                        //cas成功说明新增节点已经被放入链表，然后设置当前尾节点
                        if (p != t) // 一次跳两个节点
                            casTail(t, newNode);  // 失败是可以的.
                        return true;
                    }
                    // 丢失的CAS与另一线程竞争；重新读取下一个
                }
                else if (p == q)
                //多线程操作时候，由于poll时候会把老的head变为自引用，然后head的next变为新head，所以这里需要
                //重新找新的head，因为新的head后面的节点才是激活的节点
                    p = (t != (t = tail)) ? t : head;
                else
                    // Check for tail updates after two hops.
                    p = (p != t && t != (t = tail)) ? t : q;
            }
        }
    
    
    private static void checkNotNull(Object v) {
            if (v == null)
                throw new NullPointerException();
        }

### 二、出队 ###

![1018541-20170313211713838-48054664.jpg][]

不是每次出队时都更新head节点，当**head节点里有元素时，直接弹出head节点里的元素**，而不会更新head节点。只有当head节点里没有元素时，则弹出head的next结点并更新head结点为原来head的next结点的next结点。

poll操作是在链表头部获取并且移除一个元素，下面看看实现原理。

public E poll() {
            restartFromHead:
            //死循环
            for (;;) {
                //死循环
                for (Node<E> h = head, p = h, q;;) {
                    //保存当前节点值
                    E item = p.item;
                    //当前节点有值则cas变为null
                    if (item != null && p.casItem(item, null)) {
                        //cas成功标志当前节点以及从链表中移除
                        if (p != h) // 类似tail间隔2设置一次头节点
                            updateHead(h, ((q = p.next) != null) ? q : p);
                        return item;
                    }
                    //当前队列为空则返回null
                    else if ((q = p.next) == null) {
                        updateHead(h, p);
                        return null;
                    }
                    //自引用了，则重新找新的队列头节点
                    else if (p == q)
                        continue restartFromHead;
                    else
                        p = q;
                }
            }
        }

小结：

观察入队和出队的源码可以发现，无论入队还是出队，都是在死循环中进行的，也就是说，当一个线程调用了入队、出队操作时，会尝试获取链表的tail、head结点进行插入和删除操作，而插入和删除是通过CAS操作实现的，而CAS具有原子性。

故此，如果有其他任何一个线程成功执行了插入、删除都会改变tail/head结点，那么当前线程的插入和删除操作就会失败，则通过循环再次定位tail、head结点位置进行插入、删除，直到成功为止。

也就是说，**ConcurrentLinkedQueue的线程安全是通过其插入、删除时采取CAS操作来保证的。不会出现同一个tail结点的next指针被多个同时插入的结点所抢夺的情况出现。**

## **Size方法** ##

public int size() {
            int count = 0;
            for (Node<E> p = first(); p != null; p = succ(p))
                if (p.item != null)
                    // Collection.size() spec says to max out
                    if (++count == Integer.MAX_VALUE)
                        break;
            return count;
        }

**注意1：  
ConcurrentLinkedQueue的.size() 是要遍历一遍集合的，很慢的，所以尽量要避免用size，  
如果判断队列是否为空最好用isEmpty()而不是用size来判断.**

**注意2:**

**此外，如果在执行期间添加或删除元素。对于此方法，返回的结果可能不准确。因此，此方法在并发时通常不太有用。**

这篇博客就不贴demo了，因为写的话和阻塞队列差不多。基本上常用的方法以及注意事项都以讲解，各位直接使用就好了。

[CAS]: https://blog.csdn.net/weixin_38003389/article/details/83343429
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODAwMzM4OQ_size_16_color_FFFFFF_t_70]: /images/20220401/15a1672a165a4a498a8ad38e4e2cbcbc.png
[1018541-20170313214020416-738374759.jpg]: /images/20220401/5128e1e0d8944ec88e4e21b0bd51bd09.png
[1018541-20170313211713838-48054664.jpg]: /images/20220401/0ef2a57375074e2f8a4696d4c2f31a1e.png