Java并發(fā)編程之ConcurrentLinkedQueue隊列詳情

更新時間：2022年04月15日 15:07:41 作者：派大大大星?

這篇文章主要介紹了Java并發(fā)編程之ConcurrentLinkedQueue隊列詳情，ConcurrentLinkedQueue?內部的隊列使用單向鏈表方式實現，下文更多相關內容敘述需要的小伙伴可以參考一下

ConcurrentLinkedQueue

JDK中提供了一系列場景的并發(fā)安全隊列?？偟膩碚f，按照實現方式的不同可分為阻塞隊列和非阻塞隊列，前者使用鎖實現，而后則使用CAS非阻塞算法實現。

ConcurrentLinkedQueue 內部的隊列使用單向鏈表方式實現，其中有兩個volatile 類型的 Node 節(jié)點分別用來存放隊列的首、尾節(jié)點。從下面的無參構造函數可知，默認頭、尾節(jié)點都是指向 item 為null 的哨兵節(jié)點。新元素會被插入隊列末尾，出隊時從隊列頭部獲取一個元素。

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

在 Node 節(jié)點內部則維護一個使用volatile 修飾的變量 item，用來存放節(jié)點的值;next用來存放鏈表的下一個節(jié)點，從而鏈接為一個單向無界鏈表。其內部則使用 UNSafe 工具類提供的CAS 算法來保證出入隊時操作鏈表的原子性。

下面通過介紹ConcurrentLinkedQueue的幾個方法來介紹其實現原理。

offer操作： offer操作是在隊列末尾添加一個元素，如果傳遞的參數是null則拋出NPE異常，否則由于ConcurrentLinkedQueue是無界隊列，該方法一直會返回true。另外，由于使用CAS無阻塞算法，因此該方法不會阻塞掛起調用線程。下面具體看下實現原理。

public boolean offer(E e) {
//（1）e為null這拋出空指針異常
    checkNotNull(e);
    //(2)構造Node節(jié)點，在構造函數內部調用unsafe.putObject
    final Node<E> newNode = new Node<E>(e);
    //(3) 從尾節(jié)點插入
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
 //(4) 如果q==null說明p是尾節(jié)點，則執(zhí)行插入
        if (q == null) {
            // p is last node
            //（5）使用CAS設置p節(jié)點的next節(jié)點
            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".
                   //（6）CAS成功，則說明新增節(jié)點已經放入鏈表，然后設置當前尾巴節(jié)點

                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 = (p != t && t != (t = tail)) ? t : q;
    }
}

首先看當一個線程調用offer(item)時的情況。首先代碼(1)對傳參進行空檢查，由于使用如果為null 則拋出NPE 異常，否則執(zhí)行代碼(2)并使用item作為構造函數參數創(chuàng)建一個新的節(jié)點，然后代碼(3)從隊列尾部節(jié)點開始循環(huán)，打算從隊列尾部添加元素。這時候節(jié)點p、t、head、tail同時指向了item為null的哨兵節(jié)點，由于哨兵節(jié)點的next 節(jié)點為null，所以這里q也指向null。代碼(4)發(fā)現q->null則執(zhí)行代碼(5),通過CAS 原子操作判斷p節(jié)點的next節(jié)點是否為null，如果為null 則使用節(jié)點newNode替換p的next節(jié)點，然后執(zhí)行代碼(6)，這里由于p=t所以沒有設置尾部節(jié)點，然后退出 offer方法。
上面是一個線程調用offer方法的情況，如果多個線程同時調用，就會存在多個線程同時執(zhí)行到代碼(5)的情況。假設線程A調用offer(item1)，線程B調用 ofer(item2)，同時執(zhí)行到代碼(5)p.casNext(null, newNode)。由于CAS的比較設置操作是原子性的，所以這里假設線程A先執(zhí)行了比較設置操作，發(fā)現當前p的 next 節(jié)點確實是null，則會原子性地更新next節(jié)點為iteml，這時候線程B也會判斷p的next節(jié)點是否為null，結果發(fā)現不是null(因為線程A已經設置了p的next節(jié)點為iteml)，則會跳到代碼(3)，然后執(zhí)行到代碼(4)。

可見，offer 操作中的關鍵步驟是代碼(5)，通過原子CAS 操作來控制某時只有一個線程可以追加元素到隊列末尾。進行CAS 競爭失敗的線程會通過循環(huán)一次次嘗試進行 CAS操作，直到CAS 成功才會返回，也就是通過使用無限循環(huán)不斷進行 CAS 嘗試方式來替代阻塞算法掛起調用線程。相比阻塞算法，這是使用CPU資源換取阻塞所帶來的開銷。

add操作：

add操作是在鏈表尾部添加一個元素，其實在內部調用的還是offer操作。

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

poll操作：

poll操作是在隊列頭部獲取并移除一個元素，如果隊列為空則返回null。

public E poll() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;

            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
                    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;
        }
    }
}

poll方法在移除一個元素時，只是簡單地使用 CAS操作把當前節(jié)點的item值設置為null，然后通過重新設置頭節(jié)點將該元素從隊列里面移除，被移除的節(jié)點就成了孤立節(jié)點，這個節(jié)點會在垃圾回收時被回收掉。另外，如果在執(zhí)行分支中發(fā)現頭節(jié)點被修改了，要跳到外層循環(huán)重新獲取新的頭節(jié)點。

peak操作：

peak操作是獲取隊列頭部獲一個元素，如果隊列為空則返回null。

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;
        }
    }
}

Peek操作的代碼結構與poll操作類似，不同之處在于我們在代碼中標記注釋的地方中少了castItem操作。其實這很正常，因為peek只是獲取隊列頭元素值，并不清空其值。根據前面的介紹我們知道第一次執(zhí)行offer后head指向的是哨兵節(jié)點(也就是item為null的節(jié)點)，那么第一次執(zhí)行peek時在注釋處會發(fā)現item==null，然后執(zhí)行q=p.next，這時候q節(jié)點指向的才是隊列里面第一個真正的元素，或者如果隊列為 null 則 q 指向 null。

到此這篇關于Java并發(fā)編程之ConcurrentLinkedQueue隊列詳情的文章就介紹到這了,更多相關Java并發(fā)編程 ConcurrentLinkedQueue 內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: