ConcurrentLinkedQueue

比眉伴天荒 2022-09-07 14:38 175阅读 0赞

#### ConcurrentLinkedQueue ####

ConcurrentLinkedQueue 的实现原理和AQS内部的阻塞队列类似，AQS是基于双向链表，通过对head/tail进行CAS操作，实现入队和出队，队列里面放的是线程，而ConcurrentLinkedQueue是一个单向链表，里面放的是实际的元素

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> implements
    Queue<E>, java.io.Serializable {
        
        private static class Node<E> {
        
            volatile E item;
            //只有指向后一个的node，没有指向前一个的
            volatile Node<E> next;
    	//...
        }
        
        private transient volatile Node<E> head;
        private transient volatile Node<E> tail;
        //...
    }

在AQS的阻塞队列中，每次入队后，tail一定后移一个位置，指向新的最后一个元素;每次出队，head一定后移一个位置，以保证head指向队列头部，tail指向链表尾部。

但在ConcurrentLinkedQueue中，head/tail的更新可能落后于节点的入队和出队，因为它不是直接对head/tail指针进行CAS操作的，而是对Node中的item进行操作。

ConcurrentLinkedQueue代码实现

public ConcurrentLinkedQueue() {
        
    	//初始化的时候head和tail都指向了node的item为null节点
        head = tail = new Node<E>(null);
    }

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

//入队列
    public boolean offer(E e) {
        
        checkNotNull(e);
        //要添加的节点
        final Node<E> newNode = new Node<E>(e);
    	//定义两个指向tail node的节点
        for (Node<E> t = tail, p = t;;) {
        
        	//获得tail节点的下一个节点
            Node<E> q = p.next;
            //如果q是null说明p是最后一个
            if (q == null) {
        
                //用cas将p的下一节点null替换成newNode
                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".
                    //每入列两个节点，把tail设置成新的newNode
                    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)
                // 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
                // Check for tail updates after two hops.
                //后移p node
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

假设队列中有1个节点item1，tail指向该节点，假设有一个线程要入队item2节点

1.  p=tail,t=tail, p.next=NULL, q=p.next
2.  p.casNext对p.next进行cas操作，设置p.next = item2,执行后p依然等于tail,即p等于t所以casTail不会执行，return true

然后有一个线程要入队item3

1.  p=tail,q=p.next
2.  q!=NULL，因此不会入队新节点。p=q 即p=p.next,for循环回到第一排
3.  q=p.next即q=null，对p的next执行CAS操作，入队item3节点
4.  p!=t, 执行casTail操作, tail后移2个位置，到达队列尾部,每追加2个节点，才后移1次tail指针

//出队列
    public E poll() {
        
        restartFromHead:
        for (;;) {
        
            for (Node<E> h = head, p = h, q;;) {
        
                E item = p.item;
    			//p的item设置为null
                if (item != null && p.casItem(item, null)) {
        
                    // Successful CAS is the linearization point
                    // for item to be removed from this queue.
                    //每设置两个节点的item为null，把head后移2位
                    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;
            }
        }
    }

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

1.  p=head, p.item = null
2.  q=p.next，p!=q执行p=q(p指针后移)进入下次for循环
3.  p.item!=null,p.casItem设置item为null
4.  p!=head，此时队列中有了2个item为 NULL 的节点（head节点和p节点），移动head指针，对其执行updateHead 操作。

因为head/tail 并不是精确地指向队列头部和尾部，所以不能简单地通过比较 head/tail 指针来判断 队列是否为空，而是需要从head指针开始遍历，找第1个不为NULL的节点。如果找到，则队列不为空; 如果找不到，则队列为空。

public boolean isEmpty() {
        
            return first() == null;
        }
    
        Node<E> first() {
        
            restartFromHead:
            for (;;) {
        
            	//从head开始遍历找第一个item不是null的节点
                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;
                }
            }
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTAyOTI4Ng_size_16_color_FFFFFF_t_70]: /images/20220829/bf8d7b606e534e15a49833ae0fcf7f05.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTAyOTI4Ng_size_16_color_FFFFFF_t_70 1]: /images/20220829/8b4926d52859451aac6f78127417c443.png