ConcurrentLinkedQueue的入队和出队原理详解

缺乏、安全感 2021-08-29 14:37 357阅读 0赞

### 入队（offer） ###

/** * Inserts the specified element at the tail of this queue. * As the queue is unbounded, this method will never return {@code false}. * * @return {@code true} (as specified by {@link Queue#offer}) * @throws NullPointerException if the specified element is null * 入队 * * 队列刚初始化如下： * head * |__null__| * head * * 入队A元素： * t=p=tail，q=p.next，q==null执行①尝试入队 * head * |__null__| null * tail q * t * p * * 入队成功：执行②,p==t，不更新tail，直接退出 * head * |__null__| ----> |__A__| * tail next q * t * p * * * 再入队B： * t=p=tail,q=p.next=A，q!=null，执行④p!=q，执行⑥： * p==t && t==tail 所以三元结果为q,p=q,第一次循环结束 * * head * |__null__| ----> |__A__| * tail next q * t * p * * 第一次循环结束，开始第二次循环：q=p.next,q==null，执行①尝试入队 * * head * |__null__| ----> |__A__| * tail next p q * t * * 入队成功则执行②:p!=t,需要更新tail，执行③ * head newNode * |__null__| ----> |__A__| ----> |__B__| * tail next p next q * t * * 执行③之后更新tail无论成功与否都返回true * head newNode * |__null__| ----> |__A__| ----> |__B__| * next p next q * t tail * * * 继续入队C：t=p=tail,q=p.next,q==null，执行①尝试入队 * head * |__null__| ----> |__A__| ----> |__B__| null * next next tail q * t * p * 入队成功则执行②：p==t，不需要更新tail，直接返回 * head newNode * |__null__| ----> |__A__| ----> |__B__| ----> |__C__| * next next tail next q * t * p * * * 现在考虑一种情况：在并发情况下，该线程正在offer()时，tail因其他线程的poll()被删了 * 此时就需要重新定位tail到head位置（tail被删，说明所有元素都出队了，队列为空） * * 下图为连续出队ABC之后的结果： * * <-------------| <-----------| head * |__null__| ------| |__null__| -----> |__null__| -----| |__null__| null * next tail * * 下图为出队AB，然后该线程想要入队D时候的结果：也就是第一次循环时 * <-------------| head * |__null__| ------| |__null__| -----> |__null__| -----> |__C__| * next tail next * t q * p * t=p=tail,q=p.next,q!=null,准备执行④，但在执行④之前，其他线程把C出队了，导致q=p=null， * 所以就要执行⑤，复位tail到head的位置。 * * 如果不复位，入队结点就会插入到tail.next.next位置，新节点更新为tail后，在tail前面会出现一个 * null值！ * 错误的结果如下： * <-------------| head ( C ) tail * |__null__| ------| |__null__| -----> |__null__| -----> |__null__| ----> |__D__| * next next next * t p q * */
    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;
            //定位尾结点，tail不一定是尾结点，可能是tail，也可能是tail.next
            //提升了效率，两次入队中，只有一次需要cas操作
            if (q == null) { 
                // p is last node
                //q为null，则p为尾结点
                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".
                    //首次添加时，p==t，不进行尾结点更新，所以尾结点存在滞后性
                    //并发环境，可能存在添加/删除，tail就更难保证正确指向最后结点
                    if (p != t) // hop two nodes at a time②
                        //更新尾结点为最新元素
                        casTail(t, newNode);  // Failure is OK.③
                        //如果失败也可以接受，说明有其它线程更新tail成功了
                    return true;
                }
                // Lost CAS race to another thread; re-read next
            }
            //p==q==null，则空队列，说明其他线程此时进行了出队，将q给删了
            //如果tail不是尾结点，此时q又是唯一的队列元素，则tail被删掉了
            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.
                //此时需要对tail结点进行复位，复位到head结点
                p = (t != (t = tail)) ? t : head;//⑤
            else
                // Check for tail updates after two hops.
                //推动tail尾结点往队尾移动
                p = (p != t && t != (t = tail)) ? t : q;//⑥
        }
    }

#### 出队(poll) ####

/** * 如果head结点元素不为null，则出队head元素，不更新head结点； * 如果head结点元素为null，则更新head结点 * * 假设已入队3个元素A、B、C * 出队A： * p.item==null，然后执行④，q=A,q!=null,然后执行⑦， * head * |__null__| ------> |__A__| -----> |__B__| -----> |__C__| * h next next tail next * p q * * * 执行⑦之后，p=q,进行第二次循环：p.item!=null，出队；然后执行②p!=h，执行③更新head： * q=p.next==B,B!=null，三元结果为q，移动head=q，同时将原先的头结点设置为自引用 * * head * |__null__| ------> |__A__| -----> |__B__| -----> |__C__| * h next next tail next * q * p * * 执行③之后：A出队成功 * <-------------| head * |__null__| ------| |__null__| -----> |__B__| -----> |__C__| * h next tail next * p q * * * 出队B： * h=p=head,执行①p.item!=null，直接尝试出队； * <-------------| head * |__null__| ------| |__null__| -----> |__B__| -----> |__C__| * next tail next * h * p * 出队成功，执行②，p==h，直接返回，结束。 * <-------------| head * |__null__| ------| |__null__| -----> |__null__| -----> |__C__| * next tail next * h * p * * 出队C： * h=p=head，执行①，p.item==null；进而执行④，q=p.next,q!=null；然后执行⑦ * <-------------| head * |__null__| ------| |__null__| -----> |__null__| -----> |__C__| * next tail next * h q * p * * 执行⑦，p=q；进入第二轮循环：执行①，p.item=!=null，尝试出队，出队成功执行②，p!=h， * 执行③：q=p.next,q==null,三元结果为p，更新head到p，将原head设置为自引用 * <-------------| head * |__null__| ------| |__null__| -----> |__null__| -----> |__C__| * next tail next * h q * p * 执行③之后结果： * <-------------| <-----------| head * |__null__| ------| |__null__| -----> |__null__| -----| |__null__| null * next tail p q * h * * 此时发现一个问题，因为连续出队，tail被GC了...如果再入队，需要重新定位tail=head * 才可以继续入队，详解见入队的图示。 */
    public E poll() { 
        //标记
        restartFromHead:
        for (;;) { 
            for (Node<E> h = head, p = h, q;;) { 
                //获取head元素
                E item = p.item;
                //如果head元素不为空，则直接cas尝试出队
                if (item != null && p.casItem(item, null)) { //①
                    // Successful CAS is the linearization point
                    // for item to be removed from this queue.
                    if (p != h) // hop two nodes at a time ②
                        //更新head，如果p.next为null，则head为p，如果p.next!=null，则head为q（p.next）
                        updateHead(h, ((q = p.next) != null) ? q : p);//③
                    //返回元素
                    return item;
                }
                //p.next == null则空队列，更新head，返回null
                else if ((q = p.next) == null) { //④
                    updateHead(h, p);//⑤
                    return null;
                }
                //第一轮更新失败（没有竞争过其他线程），跳回标志位重新开始
                else if (p == q)//⑥
                    continue restartFromHead;
                //移动head结点为head.next
                else
                    p = q;//⑦
            }
        }
    }