快捷導(dǎo)航

Java并發(fā)容器之ConcurrentLinkedQueue詳解

更新時間：2023年12月26日 09:51:06 作者：fastjson_

這篇文章主要介紹了Java并發(fā)容器之ConcurrentLinkedQueue詳解,加鎖隊(duì)列的實(shí)現(xiàn)較為簡單,這里就略過,我們來重點(diǎn)來解讀一下非阻塞隊(duì)列,
從點(diǎn)到面, 下面我們來看下非阻塞隊(duì)列經(jīng)典實(shí)現(xiàn)類ConcurrentLinkedQueue,需要的朋友可以參考下

簡介

并編程中，一般需要用到安全的隊(duì)列，如果要自己實(shí)現(xiàn)安全隊(duì)列，可以使用2種方式：

加鎖，這種實(shí)現(xiàn)方式就是我們常說的阻塞隊(duì)列。
使用循環(huán)CAS算法實(shí)現(xiàn)，這種方式實(shí)現(xiàn)隊(duì)列稱之為非阻塞隊(duì)列。

加鎖隊(duì)列的實(shí)現(xiàn)較為簡單，這里就略過，我們來重點(diǎn)來解讀一下非阻塞隊(duì)列。從點(diǎn)到面，下面我們來看下非阻塞隊(duì)列經(jīng)典實(shí)現(xiàn)類：ConcurrentLinkedQueue （JDK1.8版）

看下ConcurrentLinkedQueue的結(jié)構(gòu)圖

從內(nèi)圖可以了解ConcurrentLinkedQueue一個大概，ConcurrentLinkedQueue內(nèi)部持有2個節(jié)點(diǎn)：head頭結(jié)點(diǎn)，負(fù)責(zé)出列， tail尾節(jié)點(diǎn)，負(fù)責(zé)入列。

而元素節(jié)點(diǎn)Node，使用item存儲入列元素，next指向下一個元素節(jié)點(diǎn)。

    private static class Node<E> {
        volatile E item;
        volatile Node<E> next;
        //....
    }

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable {
        private transient volatile Node<E> head;
        private transient volatile Node<E> tail;  
        //....
}

ConcurrentLinkedQueue使用特點(diǎn)

不允許null入列
在入隊(duì)的最后一個元素的next為null
隊(duì)列中所有未刪除的節(jié)點(diǎn)的item都不能為null且都能從head節(jié)點(diǎn)遍歷到
刪除節(jié)點(diǎn)是將item設(shè)置為null, 隊(duì)列迭代時跳過item為null節(jié)點(diǎn)
head節(jié)點(diǎn)跟tail不一定指向頭節(jié)點(diǎn)或尾節(jié)點(diǎn)，可能存在滯后性

之所以有這奇葩約定，全因ConcurrentLinkedQueue是并發(fā)非阻塞隊(duì)列決定的。我們從源碼上看一下ConcurrentLinkedQueue實(shí)現(xiàn)過程

ConcurrentLinkedQueue源碼詳解

入列

我們印象中鏈表特點(diǎn)：tail節(jié)點(diǎn)表示最后一個節(jié)點(diǎn)， head表示第一個節(jié)點(diǎn)。ConcurrentLinkedQueue 跟傳統(tǒng)的鏈表有點(diǎn)區(qū)別，在單線程環(huán)境下符合傳統(tǒng)鏈表特點(diǎn)，但涉及到多線程環(huán)境，ConcurrentLinkedQueue 中的tail節(jié)點(diǎn)不一定是最后一個節(jié)點(diǎn)，他可能是倒數(shù)第二個。所以ConcurrentLinkedQueue判斷隊(duì)尾條件是節(jié)點(diǎn)的next為null。

   public boolean offer(E e) {
        checkNotNull(e);   //為空判斷，e為null是拋異常
        final Node<E> newNode = new Node<E>(e); //將e包裝成newNode
        for (Node<E> t = tail, p = t;;) {  //循環(huán)cas，直至加入成功
            //t = p = tail 
            Node<E> q = p.next;
            if (q == null) {   //判斷p是否為尾節(jié)點(diǎn)
                //如果是，p.next = newNode
                if (p.casNext(null, newNode)) {
                    //首次添加時，p 等于t，不進(jìn)行尾節(jié)點(diǎn)更新，所以所尾節(jié)點(diǎn)存在滯后性  
                    //并發(fā)環(huán)境，可能存添加/刪除，tail就更難保證正確指向最后節(jié)點(diǎn)。
                    if (p != t) 
                        //更新尾節(jié)點(diǎn)為最新元素
                        casTail(t, newNode);  
                    return true;
                }
            }
            else if (p == q)
                //當(dāng)tail不執(zhí)行最后節(jié)點(diǎn)時，如果執(zhí)行出列操作，很有可能將tail也給移除了    
                //此時需要對tail節(jié)點(diǎn)進(jìn)行復(fù)位，復(fù)位到head節(jié)點(diǎn)
                p = (t != (t = tail)) ? t : head;
            else
                //推動tail尾節(jié)點(diǎn)往隊(duì)尾移動
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

分析

1、初始化

2、添加A元素

3、添加B元素

4、添加C

從圖上看tail不一定執(zhí)行最后一個節(jié)點(diǎn)，但可以確定最后節(jié)點(diǎn)的next節(jié)點(diǎn)為null。

到這可能朋友問他，并發(fā)環(huán)境什么情況都有可能，ConcurrentLinkedQueue是怎么保證線程安全的？我們觀察offer方法的設(shè)計(jì)，

1：是一個死循環(huán)，就是不停使用cas判斷直到添加元素入隊(duì)成功。

for (Node<E> t = tail, p = t;;)

2：2個cas判斷方法 p.casNext(null, newNode) 確保隊(duì)列在入列時是原子操作

 private boolean casTail(Node<E> cmp, Node<E> val) {
     return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
 }

casTail(t, newNode); 確保tail隊(duì)尾在移動改變時是原子操作

boolean casNext(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}

而在并發(fā)環(huán)境，ConcurrentLinkedQueue入列線程安全考慮具體可分2類：

1>線程1線程2同時入列這個好理解，線程1，線程2不管在offer哪個位置開始并發(fā)，他們最終的目的都是入列，也即都需要執(zhí)行casNext方法，我們只需要確保所有線程都有機(jī)會執(zhí)行casNext方法，并且保證casNext方法是原子操作即可。casNext失敗的線程，可以進(jìn)入下一輪循環(huán)，人品好的話就可以入列，衰的話繼續(xù)循環(huán)

2>線程1遍歷，線程2入列 ConcurrentLinkedQueue 遍歷是線程不安全的，線程1遍歷，線程2很有可能進(jìn)行入列出列操作，所以ConcurrentLinkedQueue 的size是變化。換句話說，要想安全遍歷ConcurrentLinkedQueue 隊(duì)列，必須額外加鎖。

但換一個角度想， ConcurrentLinkedQueue 的設(shè)計(jì)初衷非阻塞隊(duì)列，我們更多關(guān)注入列與出列線程安全，這2點(diǎn)能保證就可以啦。

出列

    public E poll() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                //入列折騰的tail，那出列折騰的就是head
                E item = p.item;
                //出列判斷依據(jù)是節(jié)點(diǎn)的item=null
                //item ！= null， 并且能將操作節(jié)點(diǎn)的item設(shè)置null， 表示出列成功
                if (item != null && p.casItem(item, null)) {
                    if (p != h) 
                        //一旦出列成功需要對head進(jìn)行移動
                        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)
                    //第一輪操作失敗，下一輪繼續(xù)，調(diào)回到循環(huán)前
                    continue restartFromHead;
                else
                    //推動head節(jié)點(diǎn)移動
                    p = q;
            }
        }
    }