Java并发编程--并发队列原理之ConcurrentLinkedQueue

小咪咪 2023-07-01 15:54 57阅读 0赞

### 文章目录 ###

*  ConcurrentLinkedQueue原理探究
 *   *   *  (1). 结构
         *  (2). ConcurrentLinkedQueue原理介绍
         *   *  1). offer操作
             *  2). poll操作
             *  3). peek操作
             *  4). size操作
             *  5). remove操作
             *  6). contains操作
         *  (3). 小结

# ConcurrentLinkedQueue原理探究 #

ConcurrentLinkedQueue是线程安全的无界非阻塞队列,其底层数据结构使用单项链表实现,对于入队和出队操作使用CAS来实现线程安全.

### (1). 结构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]  
 **内部使用了两个Volatile类型的Node节点分别用来存放队列的首尾节点.Node节点内部则维护一个使用volatile修饰的变量item来存放节点的值**

### (2). ConcurrentLinkedQueue原理介绍 ###

#### 1). offer操作 ####

在队列末尾添加一个元素,不能传入null(抛出NPE异常).内部使用CAS无阻塞算法,不会阻塞挂起线程.

public boolean offer(E e) {
        
        // 检查如果传入空数据,抛出异常
        checkNotNull(e);
        final Node<E> newNode = new Node<E>(e);
    
        // 自旋式的从尾节点插入
        // 1、根据tail节点定位出尾节点（last node）；2、将新节点置为尾节点的下一个节点；3、casTail更新尾节点
        for (Node<E> t = tail, p = t;;) {
        
            // p用来表示队列的尾节点，初始情况下等于tail节点
            // q是p的next节点
            Node<E> q = p.next;
            if (q == null) {
        
                // p是尾节点
                // 设置p节点的下一个节点为新节点，设置成功则casNext返回true
                // 否则返回false，说明有其他线程更新过尾节点
                if (p.casNext(null, newNode)) {
        
                    // 如果p != t，则将入队节点设置成tail节点，更新失败了也没关系
                    // 因为失败了表示有其他线程成功更新了tail节点
                    // 这里使队列每添加两次,尾节点更新一次
                    if (p != t)
                        casTail(t, newNode);
                    return true;
                }
            }
        	// 执行了poll后可能会出现头节点自引用的情况
        	// 所以这里需要重新找新的head，因为新的head后面的节点才是激活的节点
            else if (p == q)
                // 先取得t的值，在执行t = tail，并取得新的t的值，然后比较这两个值是否相等。
                // 这种情况表示在比较的过程中，tail被其他线程修改了，这时，我们就用新的tail为链表的尾
                // 但如果tail没有被修改，则返回head，要求从头部开始，重新查找链表末尾。
                p = (t != (t = tail)) ? t : head;
            else
                // 判断尾节点是否被改变,如果没有将p向后移动
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

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

总而言之,当添加一个节点时会出现两种状态,p节点是尾节点,这种情况下可以插入.p节点不是尾节点(被其他线程修改),这种情况下需要走最后一个else分支将p指针向后移动.

另外,poll操作可能将头节点自引用,那么需要将p指向新的head然后重新寻找尾节点.

#### 2). poll操作 ####

在队列头部获取并移除一个元素,如果队列为空返回null.

public E poll() {
        
        // 这个是goto标记
        restartFromHead:// (1)
        for (;;) {
        // (2)
            for (Node<E> h = head, p = h, q;;) {
        
                // p节点表示首节点，即需要出队的节点
                E item = p.item;// (3)
                // 不是空队列,且CAS操作成功,将头结点后一个节点的元素置空
                if (item != null && p.casItem(item, null)) {
        // (4)
        			// 之前q被移动过,将p设置为头节点
                    if (p != h)// (5)
        				// 这一步将头结点自引用了,目的是为了下一步走向(7)
                        updateHead(h, ((q = p.next) != null) ? q : p);
                    return item;
                }
                // 如果头节点的元素为空或头节点发生了变化，这说明头节点已经被另外一个线程修改了。
                // 那么获取p节点的下一个节点，如果p节点的下一节点为null，则表明队列已经空了
                else if ((q = p.next) == null) {
        // (6)
        			// 这种情况下多是其他线程将队列中的元素取光了,那么重新设置头结点,返回null
                    updateHead(h, p);
                    return null;
                }
                // 运行到这里说明有其他线程添加了尾节点,使该队列不为空队列
                else if (p == q)// (7)
        			// 重新执行该方法
                    continue restartFromHead;
                // 将p向后移动,
                else// (8)
                    p = q;
            }
        }
    }
    
    // 设置头结点,并将原来的头结点自引用,提醒其他线程更新头结点
    final void updateHead(Node<E> h, Node<E> p) {
        
        if (h != p && casHead(h, p))
            // 将旧的头结点h的next域指向为h
            h.lazySetNext(h);
    }

总结一下,当没有其他线程打扰,方法将一步走到(5),然后重新设置头节点,并退出方法.如果其他线程这时将队列中的元素取光了,那么运行到(6).如果碰巧有其他线程添加了尾节点,那么运行到(7)或者(8),一般先运行(8),将p向后移动一个节点,下一次循环中走到(5)之后会将重新设置头结点,并将原h节点(尾节点)自引用,这样的情况下其他线程的代码会走向(7),重新执行该方法.之前提到的offer方法中也有这种情况的相对策略.

并不是每次出队时都更新head节点，当head节点里有元素时，直接弹出head节点里的元素，而不会更新head节点。只有当head节点里没有元素时，出队操作才会更新head节点。

#### 3). peek操作 ####

// 获取链表的首部元素（只读取而不移除）
    public E peek() {
        
        restartFromHead:
        for (;;) {
        
            for (Node<E> h = head, p = h, q;;) {
        
                E item = p.item;
                if (item != null || (q = p.next) == null) {
        
     				// 执行peek()方法后head会指向第一个具有非空元素的节点。
                    updateHead(h, p);
                    return item;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }

#### 4). size操作 ####

计算当前队列元素个数,统计元素是不准确的

public int size() {
        
        int count = 0;
        // first()获取第一个具有非空元素的节点，若不存在，返回null
        // succ(p)方法获取p的后继节点，若p == p的后继节点，则返回head
        for (Node<E> p = first(); p != null; p = succ(p))
            if (p.item != null)
                // Collection.size() spec says to max out
                // 最大返回Integer.MAX_VALUE
                if (++count == Integer.MAX_VALUE)
                    break;
        return count;
    }
    
    // 获取队列中的第一个有效节点
    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;
            }
        }
    }
    
    // 获取传入节点的后继节点,如果该节点自引用,返回真正的头结点
    final Node<E> succ(Node<E> p) {
        
        Node<E> next = p.next;
        return (p == next) ? head : next;
    }

#### 5). remove操作 ####

如果队列中存在该元素则删除该元素,存在多个则删除第一个.

public boolean remove(Object o) {
        
        if (o != null) {
        
            // 删除为空直接返回false
            Node<E> next, pred = null;
            for (Node<E> p = first(); p != null; pred = p, p = next) {
        
                boolean removed = false;
                E item = p.item;
                // 节点元素不为null
                if (item != null) {
        
                    // 匹配不上让p和pred向后移动
                    if (!o.equals(item)) {
        
                        next = succ(p);
                        continue;
                    }
                    // 匹配上将该元素置空
                    removed = p.casItem(item, null);
                }
    
                // 获取删除节点的后继节点
                next = succ(p);
                // 将被删除的节点移除队列
                if (pred != null && next != null) // unlink
                    pred.casNext(p, next);
                if (removed)
                    return true;
            }
        }
        return false;
    }

#### 6). contains操作 ####

判断队列中是否有制定对象,结果并不精确,但不牵扯方法内的多线程影响.

public boolean contains(Object o) {
        
        if (o == null) return false;
        // 遍历队列
        for (Node<E> p = first(); p != null; p = succ(p)) {
        
            E item = p.item;
            // 若找到匹配节点，则返回true
            if (item != null && o.equals(item))
                return true;
        }
        return false;
    }

### (3). 小结 ###

ConcurrentLinkedQueue底层使用单向链表数据结构来保存队列元素,使用非阻塞CAS算法,没有加锁.因为head和tail两个节点都是由volatile修饰的,本身可以保证可见性,所以只要保证对这两个变量操作的原子性即可.

offer操作是在tail后添加元素,实际上是调用CASNext方法,只有一个线程能成功,其他线程需要重新寻找尾节点.(队列新增两次,尾节点更新一次)

poll操作一样

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200123153146721.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200123153212466.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNTk2NTY4,size_16,color_FFFFFF,t_70