Java多線程和并發(fā)面試題（附答案）第4題

4、ConcurrentLinkedQueue非阻塞無界鏈表隊(duì)列

ConcurrentLinkedQueue是一個(gè)線程安全的隊(duì)列，基于鏈表結(jié)構(gòu)實(shí)現(xiàn)，是一個(gè)無界隊(duì)列，理論上來說隊(duì)列的長度可以無限擴(kuò)大。與其他隊(duì)列相同，ConcurrentLinkedQueue也采用的是先進(jìn)先出（FIFO）入隊(duì)規(guī)則，對(duì)元素進(jìn)行排序。當(dāng)我們向隊(duì)列中添加元素時(shí)，新插入的元素會(huì)插入到隊(duì)列的尾部；而當(dāng)我們獲取一個(gè)元素時(shí)，它會(huì)從隊(duì)列的頭部中取出。因?yàn)镃oncurrentLinkedQueue是鏈表結(jié)構(gòu)，所以當(dāng)入隊(duì)時(shí)，插入的元素依次向后延伸，形成鏈表；而出隊(duì)時(shí)，則從鏈表的第一個(gè)元素開始獲取，依次遞增；

值得注意的是，在使用ConcurrentLinkedQueue時(shí)，如果涉及到隊(duì)列是否為空的判斷，切記不可使用size()==0的做法，因?yàn)樵趕ize()方法中，是通過遍歷整個(gè)鏈表來實(shí)現(xiàn)的，在隊(duì)列元素很多的時(shí)候，size()方法十分消耗性能和時(shí)間，只是單純的判斷隊(duì)列為空使用isEmpty()即可。

public class ConcurrentLinkedQueueTest {
    public static int threadCount = 10;
    public static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
    static class Offer implements Runnable {
        public void run() {
            //不建議使用 queue.size()==0，影響效率?？梢允褂?queue.isEmpty()
            if (queue.size() == 0) {
                String ele = new Random().nextInt(Integer.MAX_VALUE) + "";
                queue.offer(ele);
                System.out.println("入隊(duì)元素為" + ele);
            }
        }
    }
    static class Poll implements Runnable {
        public void run() {
            if (!queue.isEmpty()) {
                String ele = queue.poll();
                System.out.println("出隊(duì)元素為" + ele);
            }
        }
    }
    public static void main(String[] agrs) {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        for (int x = 0; x < threadCount; x++) {
            executorService.submit(new Offer());
            executorService.submit(new Poll());
        }
        executorService.shutdown();
    }
}

一種輸出：

入隊(duì)元素為313732926

出隊(duì)元素為313732926

入隊(duì)元素為812655435

出隊(duì)元素為812655435

入隊(duì)元素為1893079357

出隊(duì)元素為1893079357

入隊(duì)元素為1137820958

出隊(duì)元素為1137820958

入隊(duì)元素為1965962048

出隊(duì)元素為1965962048

出隊(duì)元素為685567162

入隊(duì)元素為685567162

出隊(duì)元素為1441081163

入隊(duì)元素為1441081163

出隊(duì)元素為1627184732

入隊(duì)元素為1627184732

ConcurrentLinkedQuere類圖

如圖ConcurrentLinkedQueue中有兩個(gè)volatile類型的Node節(jié)點(diǎn)分別用來存在列表的首尾節(jié)點(diǎn)，其中head節(jié)點(diǎn)存放鏈表第一個(gè)item為null的節(jié)點(diǎn)，tail則并不是總指向最后一個(gè)節(jié)點(diǎn)。Node節(jié)點(diǎn)內(nèi)部則維護(hù)一個(gè)變量item用來存放節(jié)點(diǎn)的值，next用來存放下一個(gè)節(jié)點(diǎn)，從而鏈接為一個(gè)單向無界列表。

public ConcurrentLinkedQueue(){
    head=tail=new Node<E>(null);
}

如上代碼初始化時(shí)候會(huì)構(gòu)建一個(gè) item 為 NULL 的空節(jié)點(diǎn)作為鏈表的首尾節(jié)點(diǎn)。

Offer 操作offer 操作是在鏈表末尾添加一個(gè)元素，下面看看實(shí)現(xiàn)原理。

public boolean offer(E e) {
    //e 為 null 則拋出空指針異常
    checkNotNull(e);
    //構(gòu)造 Node 節(jié)點(diǎn)構(gòu)造函數(shù)內(nèi)部調(diào)用 unsafe.putObject，后面統(tǒng)一講
    final Node<E> newNode = new Node<E>(e);
    //從尾節(jié)點(diǎn)插入
    for (Node<E> t = tail, p = t; ; ) {
        Node<E> q = p.next;
        //如果 q=null 說明 p 是尾節(jié)點(diǎn)則插入
        if (q == null) {
            //cas 插入（1）
            if (p.casNext(null, newNode)) {
                //cas 成功說明新增節(jié)點(diǎn)已經(jīng)被放入鏈表，然后設(shè)置當(dāng)前尾節(jié)點(diǎn)（包含 head，1，3，5.。。個(gè)節(jié)點(diǎn)為尾節(jié)點(diǎn)）
                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)//(2)
            //多線程操作時(shí)候，由于 poll 時(shí)候會(huì)把老的 head 變?yōu)樽砸?，然?head 的 next 變?yōu)樾?head，所以這里需要
            //重新找新的 head，因?yàn)樾碌?head 后面的節(jié)點(diǎn)才是激活的節(jié)點(diǎn)
            p = (t != (t = tail)) ? t : head;
        else
            // 尋找尾節(jié)點(diǎn)(3)
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

從構(gòu)造函數(shù)知道一開始有個(gè)item為null的哨兵節(jié)點(diǎn)，并且head和tail都是指向這個(gè)節(jié)點(diǎn)。

如圖首先查找尾節(jié)點(diǎn)，q==null，p就是尾節(jié)點(diǎn)，所以執(zhí)行p.casNext通過cas設(shè)置p的next為新增節(jié)點(diǎn)，這時(shí)候p==t所以不重新設(shè)置尾節(jié)點(diǎn)為當(dāng)前新節(jié)點(diǎn)。由于多線程可以調(diào)用offer方法，所以可能兩個(gè)線程同時(shí)執(zhí)行到了（1）進(jìn)行cas，那么只有一個(gè)會(huì)成功（假如線程1成功了），成功后的鏈表為：

失敗的線程會(huì)循環(huán)一次這時(shí)候指針為：

這時(shí)候會(huì)執(zhí)行（3）所以 p=q，然后在循環(huán)后指針位置為：

 

所以沒有其他線程干擾的情況下會(huì)執(zhí)行（1）執(zhí)行 cas 把新增節(jié)點(diǎn)插入到尾部，沒有干擾的情況下線程 2 cas 會(huì)成功，然后去更新尾節(jié)點(diǎn) tail，由于 p!=t 所以更新。這時(shí)候鏈表和指針為：

假如線程 2cas 時(shí)候線程 3 也在執(zhí)行，那么線程 3 會(huì)失敗，循環(huán)一次后，線程 3 的節(jié)點(diǎn)狀態(tài)為：

這時(shí)候 p!=t ；并且 t 的原始值為 told，t 的新值為 tnew ，所以 told!=tnew，所以 p=tnew=tail

然后在循環(huán)一下后節(jié)點(diǎn)狀態(tài)：

q==null 所以執(zhí)行（1）。

現(xiàn)在就差 p==q 這個(gè)分支還沒有走，這個(gè)要在執(zhí)行 poll 操作后才會(huì)出現(xiàn)這個(gè)情況。poll 后會(huì)存在下面的狀態(tài)

這個(gè)時(shí)候添加元素時(shí)候指針分布為：

所以會(huì)執(zhí)行（2）分支 結(jié)果 p=head，然后循環(huán)，循環(huán)后指針分布：

所以執(zhí)行(1)，然后 p!=t 所以設(shè)置 tail 節(jié)點(diǎn)。現(xiàn)在分布圖：

自引用的節(jié)點(diǎn)會(huì)被垃圾回收掉。

● add 操作

add操作是在鏈表末尾添加一個(gè)元素，下面看看實(shí)現(xiàn)原理。

其實(shí)內(nèi)部調(diào)用的還是 offer

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

● poll 操作

poll 操作是在鏈表頭部獲取并且移除一個(gè)元素，下面看看實(shí)現(xiàn)原理。

public E poll() {
    restartFromHead:
    // 死 循 環(huán)
    for (; ; ) {
        //死循環(huán)
        for (Node<E> h = head, p = h, q;
                ; ) {
            //保存當(dāng)前節(jié)點(diǎn)值
            E item = p.item;
            //當(dāng)前節(jié)點(diǎn)有值則 cas 變?yōu)?null（1）
            if (item != null && p.casItem(item， null)){
                //cas 成功標(biāo)志當(dāng)前節(jié)點(diǎn)以及從鏈表中移除
                if (p != h) // 類似 tail 間隔 2 設(shè)置一次頭節(jié)點(diǎn)（2）
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //當(dāng)前隊(duì)列為空則返回 null（3）
        else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //自引用了，則重新找新的隊(duì)列頭節(jié)點(diǎn)（4）
            else if (p == q)
                continue restartFromHead;
            else//(5)
                p = q;
        }
    }
}
final void updateHead(Node<E> h,Node<E> p){
    if(h!=p&&casHead(h,p))
        h.lazySetNext(h);
}

● 當(dāng)隊(duì)列為空時(shí)候：

可知執(zhí)行（3）這時(shí)候有兩種情況，第一沒有其他線程添加元素時(shí)候(3)結(jié)果為 true 然后因?yàn)?h!=p 為 false 所以直接返回 null。第二在執(zhí)行 q=p.next 前，其他線程已經(jīng)添加了一個(gè)元素到隊(duì)列，這時(shí)候（3）返回 false，然后執(zhí)行（5）p=q，然后循環(huán)后節(jié)點(diǎn)分布：

這時(shí)候執(zhí)行（1）分支，進(jìn)行 cas 把當(dāng)前節(jié)點(diǎn)值值為 null，同時(shí)只有一個(gè)線程會(huì)成功，cas 成功 標(biāo)示該節(jié)點(diǎn)從隊(duì)列中移除了，然后 p!=h，調(diào)用 updateHead 方法，參數(shù)為 h，p;h!=p 所以把 p 變?yōu)楫?dāng)前鏈表 head 節(jié)點(diǎn)，然后 h 節(jié)點(diǎn)的 next 指向自己?，F(xiàn)在狀態(tài)為：

cas 失敗 后 會(huì)再次循環(huán)，這時(shí)候分布圖為：

這時(shí)候執(zhí)行（3）返回 null.

現(xiàn)在還有個(gè)分支（4）沒有執(zhí)行過，那么什么時(shí)候會(huì)執(zhí)行那？

這時(shí)候執(zhí)行（1）分支，進(jìn)行 cas 把當(dāng)前節(jié)點(diǎn)值值為 null，同時(shí)只有一個(gè)線程 A 會(huì)成功，cas 成功  標(biāo)示該節(jié)點(diǎn)從隊(duì)列中移除了，然后 p!=h，調(diào)用 updateHead 方法，假如執(zhí)行 updateHead 前另外一個(gè)線程 B 開始 poll 這時(shí)候它 p 指向?yàn)樵瓉淼?head 節(jié)點(diǎn)，然后當(dāng)前線程 A 執(zhí)行 updateHead 這時(shí)候 B 線程鏈表狀態(tài)為：

所以會(huì)執(zhí)行（4）重新跳到外層循環(huán)，獲取當(dāng)前 head，現(xiàn)在狀態(tài)為：

● peek 操作

peek 操作是獲取鏈表頭部一個(gè)元素（只讀取不移除），下面看看實(shí)現(xiàn)原理。

代碼與 poll 類似，只是少了 castItem.并且 peek 操作會(huì)改變 head 指向，offer 后 head 指向哨兵節(jié)點(diǎn)，第一次 peek 后 head 會(huì)指向第一個(gè)真的節(jié)點(diǎn)元素。

public E peek() {
    restartFromHead:
    for (; ; ) {
        for (Node<E> h = head, p = h, q; ; ) {
            E item = p.item;
            if (item != null || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            } else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}

● size 操作

獲取當(dāng)前隊(duì)列元素個(gè)數(shù)，在并發(fā)環(huán)境下不是很有用，因?yàn)槭褂?CAS 沒有加鎖所以從調(diào)用 size 函數(shù)到返回結(jié)果期間有可能增刪元素，導(dǎo)致統(tǒng)計(jì)的元素個(gè)數(shù)不精確。

public int size() {
    int count = 0;
    for (Node<E> p = first(); p != null; p = succ(p))
        if (p.item != null)
            // 最大返回 Integer.MAX_VALUE
            if (++count == Integer.MAX_VALUE) break;
    return count;
}

//獲取第一個(gè)隊(duì)列元素（哨兵元素不算），沒有則為 null
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;
        }
    }
}
//獲取當(dāng)前節(jié)點(diǎn)的 next 元素，如果是自引入節(jié)點(diǎn)則返回真正頭節(jié)點(diǎn)
final Node<E> succ(Node<E> p) {
    Node<E> next = p.next;
    return (p == next) ? head : next;
}

 ● remove 操作

如果隊(duì)列里面存在該元素則刪除該元素，如果存在多個(gè)則刪除第一個(gè)，并返回 true，否則返回 false

ublic boolean remove(Object o){
    //查找元素為空，直接返回 false
    if(o==null)return false;
    Node<E> pred=null;
    for(Node<E> p=first();p!=null;p=succ(p)){
        E item=p.item;
        //相等則使用 cas 值 null，同時(shí)一個(gè)線程成功，失敗的線程循環(huán)查找隊(duì)列中其他元素是否有匹配的。
        if(item!=null&&o.equals(item)&&p.casItem(item,null)){
            //獲取 next 元素
            Node<E> next=succ(p);
            //如果有前驅(qū)節(jié)點(diǎn)，并且 next 不為空則鏈接前驅(qū)節(jié)點(diǎn)到 next，
            if(pred!=null&&next!=null)
                pred.casNext(p,next);
            return true;
        }
        pred=p;
    }
    return false;
}

● contains 操作

判斷隊(duì)列里面是否含有指定對(duì)象，由于是遍歷整個(gè)隊(duì)列，所以類似 size 不是那么精確，有可能調(diào)用該方法時(shí)候元素還在隊(duì)列里面，但是遍歷過程中才把該元素刪除了，那么就會(huì)返回 false.

public boolean contains(Object o) {
    if (o == null) return false;
    for (Node<E> p = first(); p != null; p = succ(p)) {
        E item = p.item;
        if (item != null && o.equals(item)) return true;
    }
    return false;
}

關(guān)于ConcurrentLinkedQuere的offer方法有意思的問題：

offer 中有個(gè) 判斷 t != (t = tail）假如 t=node1;tail=node2;并且 node1!=node2 那么這個(gè)判斷是 true 還是 false 那，答案是 true，這個(gè)判斷是看當(dāng)前 t 是不是和 tail 相等，相等則返回 true 否者為 false，但是無論結(jié)果是啥執(zhí)行后 t 的值都是 tail。

下面從字節(jié)碼來分析下為啥。

舉一個(gè)例子：

public static void main(String[] args){
        int t=2;
        int tail=3;
        System.out.println(t!=(t=tail));
}

結(jié)果為：true

字節(jié)碼文件：

C：\Users\Simple\Desktop\TeacherCode\Crm_Test\build\classes\com\bjpowernode\crm\util>javap-c Test001

警告：二進(jìn)制文件Test001包含com.bjpowernode.crm.util.Test001 Compiled from"Test001.java"

public class com.bjpowernode.crm.util.Test001{
    public class com.bjpowernode.crm.util.Test001();
        Code:
        0:aload_0
        1:invokespecial #8    // Method java/lang/Object."<init>"：()V
        4:return

    public static void main(java.lang.String[]){
        Code:
        0:iconst_2
        1:istore_1
        2:iconst_3
        3:istore_2
        4:getstatic    #16    // Field java/lang/System.out：Ljava/io/PrintStream;
        7:iload_1
        8:iload_2
        9:dup
        10:istore_1
        11:if_icmpeq 18
        14:iconst_1
        15:goto 19
        18:iconst_0
        19:invokevirtual #22    // Method java/io/PrintStream.println：(Z)V
        22:return
}

我們從上面main方法的字節(jié)碼文件中分析

一開始棧為空：

棧

第0行指令作用是把值2入棧棧頂元素為2

棧

第1行指令作用是將棧頂int類型值保存到局部變量t中

棧

第2行指令作用是把值3入棧棧頂元素為3

棧

第3行指令作用是將棧頂int類型值保存到局部變量tail中。

棧

第4調(diào)用打印命令

第7行指令作用是把變量t中的值入棧

棧

第8行指令作用是把變量tail中的值入棧

棧

現(xiàn)在棧里面的元素為3、2，并且3位于棧頂

第9行指令作用是當(dāng)前棧頂元素入棧，所以現(xiàn)在棧內(nèi)容3，3，2

棧

第10行指令作用是把棧頂元素存放到t，現(xiàn)在棧內(nèi)容3，2

棧

第11行指令作用是判斷棧頂兩個(gè)元素值，相等則跳轉(zhuǎn)18。由于現(xiàn)在棧頂嚴(yán)肅為3，2不相等所以返回true.

第14行指令作用是把1入棧

然后回頭分析下!=是雙目運(yùn)算符，應(yīng)該是首先把左邊的操作數(shù)入棧，然后在去計(jì)算了右側(cè)操作數(shù)。

● ConcurrentLinkedQuere總結(jié)

ConcurrentLinkedQueue使用CAS非阻塞算法實(shí)現(xiàn)使用CAS解決了當(dāng)前節(jié)點(diǎn)與next節(jié)點(diǎn)之間的安全鏈接和對(duì)當(dāng)前節(jié)點(diǎn)值的賦值。由于使用CAS沒有使用鎖，所以獲取size的時(shí)候有可能進(jìn)行offer，poll或者remove操作，導(dǎo)致獲取的元素個(gè)數(shù)不精確，所以在并發(fā)情況下size函數(shù)不是很有用。另外第一次peek或者first時(shí)候會(huì)把head指向第一個(gè)真正的隊(duì)列元素。

下面總結(jié)下如何實(shí)現(xiàn)線程安全的，可知入隊(duì)出隊(duì)函數(shù)都是操作volatile變量：head，tail。所以要保證隊(duì)列線程安全只需要保證對(duì)這兩個(gè)Node操作的可見性和原子性，由于volatile本身保證可見性，所以只需要看下多線程下如果保證對(duì)著兩個(gè)變量操作的原子性。

對(duì)于offer操作是在tail后面添加元素，也就是調(diào)用tail.casNext方法，而這個(gè)方法是使用的CAS操作，只有一個(gè)線程會(huì)成功，然后失敗的線程會(huì)循環(huán)一下，重新獲取tail，然后執(zhí)行casNext方法。對(duì)于poll也是這樣的。