快捷導航

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

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

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

簡介

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

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

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

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

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

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

    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使用特點

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

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

ConcurrentLinkedQueue源碼詳解

入列

我們印象中鏈表特點：tail節(jié)點表示最后一個節(jié)點， head表示第一個節(jié)點。ConcurrentLinkedQueue 跟傳統(tǒng)的鏈表有點區(qū)別，在單線程環(huán)境下符合傳統(tǒng)鏈表特點，但涉及到多線程環(huán)境，ConcurrentLinkedQueue 中的tail節(jié)點不一定是最后一個節(jié)點，他可能是倒數(shù)第二個。所以ConcurrentLinkedQueue判斷隊尾條件是節(jié)點的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é)點
                //如果是，p.next = newNode
                if (p.casNext(null, newNode)) {
                    //首次添加時，p 等于t，不進行尾節(jié)點更新，所以所尾節(jié)點存在滯后性  
                    //并發(fā)環(huán)境，可能存添加/刪除，tail就更難保證正確指向最后節(jié)點。
                    if (p != t) 
                        //更新尾節(jié)點為最新元素
                        casTail(t, newNode);  
                    return true;
                }
            }
            else if (p == q)
                //當tail不執(zhí)行最后節(jié)點時，如果執(zhí)行出列操作，很有可能將tail也給移除了    
                //此時需要對tail節(jié)點進行復(fù)位，復(fù)位到head節(jié)點
                p = (t != (t = tail)) ? t : head;
            else
                //推動tail尾節(jié)點往隊尾移動
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

分析

1、初始化

2、添加A元素

3、添加B元素

4、添加C

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

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

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

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

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

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

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

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方法，我們只需要確保所有線程都有機會執(zhí)行casNext方法，并且保證casNext方法是原子操作即可。casNext失敗的線程，可以進入下一輪循環(huán)，人品好的話就可以入列，衰的話繼續(xù)循環(huán)

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

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

出列

    public E poll() {
        restartFromHead:
        for (;;) {
            for (Node<E> h = head, p = h, q;;) {
                //入列折騰的tail，那出列折騰的就是head
                E item = p.item;
                //出列判斷依據(jù)是節(jié)點的item=null
                //item ！= null， 并且能將操作節(jié)點的item設(shè)置null， 表示出列成功
                if (item != null && p.casItem(item, null)) {
                    if (p != h) 
                        //一旦出列成功需要對head進行移動
                        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é)點移動
                    p = q;
            }
        }
    }