前言

在并發(fā)編程中，有時(shí)候需要使用線程安全的隊(duì)列。如果要實(shí)現(xiàn)一個(gè)線程安全的隊(duì)列有兩種方式：一種是使用阻塞算法，另一種是使用非阻塞算法。

使用阻塞算法的隊(duì)列可以用一個(gè)鎖（入隊(duì)和出隊(duì)用同一把鎖）或兩個(gè)鎖（入隊(duì)和出隊(duì)用不同的鎖）等方式來實(shí)現(xiàn)。非阻塞的實(shí)現(xiàn)方式則可以使用循環(huán) CAS 的方式來實(shí)現(xiàn)。

ConcurrentLinkedQueue 是一個(gè)基于鏈接節(jié)點(diǎn)的無界線程安全隊(duì)列，它采用先進(jìn)先出的規(guī)則對(duì)節(jié)點(diǎn)進(jìn)行排序，當(dāng)我們添加一個(gè)元素的時(shí)候，它會(huì)添加到隊(duì)列的尾部；當(dāng)我們獲取一個(gè)元素時(shí)，它會(huì)返回隊(duì)列頭部的元素。它采用了“wait-free”算法（即 CAS 算法） 來實(shí)現(xiàn)，該算法在 Michael&Scott 算法上進(jìn)行了一些修改。

ConcurrentLinkedQueue 的結(jié)構(gòu)

ConcurrentLinkedQueue 由 head 節(jié)點(diǎn)和 tail 節(jié)點(diǎn)組成，每個(gè)節(jié)點(diǎn)（Node）由節(jié)點(diǎn)元素 （item）和指向下一個(gè)節(jié)點(diǎn)（next）的引用組成，節(jié)點(diǎn)與節(jié)點(diǎn)之間就是通過這個(gè) next 關(guān)聯(lián)起來，從而組成一張鏈表結(jié)構(gòu)的隊(duì)列。默認(rèn)情況下 head 節(jié)點(diǎn)存儲(chǔ)的元素為空，tail 節(jié) 點(diǎn)等于 head 節(jié)點(diǎn)。

入隊(duì)列

1) 入隊(duì)列的過程

每添加一個(gè)節(jié)點(diǎn)就做了一個(gè)隊(duì)列的快照?qǐng)D

• 添加元素 1。隊(duì)列更新 head 節(jié)點(diǎn)的 next 節(jié)點(diǎn)為元素 1 節(jié)點(diǎn)。又因?yàn)?tail 節(jié)點(diǎn)默認(rèn) 情況下等于 head 節(jié)點(diǎn)，所以它們的 next 節(jié)點(diǎn)都指向元素 1 節(jié)點(diǎn)。
• 添加元素 2。隊(duì)列首先設(shè)置元素 1 節(jié)點(diǎn)的 next 節(jié)點(diǎn)為元素 2 節(jié)點(diǎn)，然后更新 tail 節(jié)點(diǎn)指向元素 2 節(jié)點(diǎn)。
• 添加元素 3，設(shè)置 tail 節(jié)點(diǎn)的 next 節(jié)點(diǎn)為元素 3 節(jié)點(diǎn)。
• 添加元素 4，設(shè)置元素 3 的 next 節(jié)點(diǎn)為元素 4 節(jié)點(diǎn)，然后將 tail 節(jié)點(diǎn)指向元素 4 節(jié)點(diǎn)。

入隊(duì)主要做兩件事情：

第一是將入隊(duì)節(jié)點(diǎn)設(shè)置成當(dāng)前隊(duì)列尾節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)；

第二是更新 tail 節(jié)點(diǎn)，如果 tail 節(jié)點(diǎn)的 next 節(jié)點(diǎn)不為空，則將入隊(duì)節(jié)點(diǎn)設(shè)置成 tail 節(jié)點(diǎn)，如果 tail 節(jié)點(diǎn)的 next 節(jié)點(diǎn)為 空，則將入隊(duì)節(jié)點(diǎn)設(shè)置成 tail 的 next 節(jié)點(diǎn)，所以 tail 節(jié)點(diǎn)不總是尾節(jié)點(diǎn)

/*在此隊(duì)列的尾部插入指定元素。由于隊(duì)列是無界的，這個(gè)方法永遠(yuǎn)不會(huì)返回false 。
返回值：
true （由Queue.offer指定）
拋出：
NullPointerException – 如果指定元素為 null*/
public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    // 入隊(duì)前，創(chuàng)建一個(gè)入隊(duì)節(jié)點(diǎn)
    Node<E> n = new Node<E>(e);
    retry:
    // 死循環(huán)，入隊(duì)不成功反復(fù)入隊(duì)。
    for (; ; ) {
        // 創(chuàng)建一個(gè)指向 tail 節(jié)點(diǎn)的引用
        Node<E> t = tail;
        // p 用來表示隊(duì)列的尾節(jié)點(diǎn)，默認(rèn)情況下等于 tail 節(jié)點(diǎn)。
        Node<E> p = t;
        for (int hops = 0; ; hops++) { // 獲得 p 節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)。
            Node<E> next = succ(p);
            // next 節(jié)點(diǎn)不為空，說明 p 不是尾節(jié)點(diǎn)，需要更新 p 后在將它指向 next 節(jié)點(diǎn)
            if (next != null) {
                // 循環(huán)了兩次及其以上，并且當(dāng)前節(jié)點(diǎn)還是不等于尾節(jié)點(diǎn)
                if (hops > HOPS && t != tail) continue retry;
                p = next;
            }
            // 如果 p 是尾節(jié)點(diǎn)，則設(shè)置 p 節(jié)點(diǎn)的 next 節(jié)點(diǎn)為入隊(duì)節(jié)點(diǎn)。
            else if (p.casNext(null, n)) {
/*如果 tail 節(jié)點(diǎn)有大于等于 1 個(gè) next 節(jié)點(diǎn)，則將入隊(duì)節(jié)點(diǎn)設(shè)置成 tail 節(jié)點(diǎn)， 
更新失敗了也沒關(guān)系，因?yàn)槭×吮硎居衅渌€程成功更新了 tail 節(jié)點(diǎn)*/
                if (hops >= HOPS)
                    casTail(t, n); // 更新 tail 節(jié)點(diǎn)，允許失敗
                return true;
            }
            // p 有 next 節(jié)點(diǎn),表示 p 的 next 節(jié)點(diǎn)是尾節(jié)點(diǎn)，則重新設(shè)置 p 節(jié)點(diǎn)
            else {
                p = succ(p);
            }
        }
    }
}

整個(gè)入隊(duì)過程主要做兩件事情：第一是定位出尾節(jié)點(diǎn)；第二是 使用 CAS 算法將入隊(duì)節(jié)點(diǎn)設(shè)置成尾節(jié)點(diǎn)的 next 節(jié)點(diǎn)，如不成功則重試。

2) 定位尾節(jié)點(diǎn)

tail 節(jié)點(diǎn)并不總是尾節(jié)點(diǎn)，所以每次入隊(duì)都必須先通過 tail 節(jié)點(diǎn)來找到尾節(jié)點(diǎn)。尾節(jié)點(diǎn)可能是 tail 節(jié)點(diǎn)，也可能是 tail 節(jié)點(diǎn)的 next 節(jié)點(diǎn)。代碼中循環(huán)體中的第一個(gè) if 就是判 斷 tail 是否有 next 節(jié)點(diǎn)，有則表示 next 節(jié)點(diǎn)可能是尾節(jié)點(diǎn)。獲取 tail 節(jié)點(diǎn)的 next 節(jié)點(diǎn)需要注意的是 p 節(jié)點(diǎn)等于 p 的 next 節(jié)點(diǎn)的情況，只有一種可能就是 p 節(jié)點(diǎn)和 p 的 next 節(jié)點(diǎn) 都等于空，表示這個(gè)隊(duì)列剛初始化，正準(zhǔn)備添加節(jié)點(diǎn)，所以需要返回 head 節(jié)點(diǎn)。

final Node<E> succ(Node<E> p) {
     Node<E> next = p.getNext();
     return (p == next) head:next;
}

3) 設(shè)置入隊(duì)節(jié)點(diǎn)為尾節(jié)點(diǎn)

p.casNext（null，n）方法用于將入隊(duì)節(jié)點(diǎn)設(shè)置為當(dāng)前隊(duì)列尾節(jié)點(diǎn)的 next 節(jié)點(diǎn)，如果 p.next 是 null，表示 p 是當(dāng)前隊(duì)列的尾節(jié)點(diǎn)，如果不為 null，表示有其他線程更新了尾節(jié)點(diǎn)， 則需要重新獲取當(dāng)前隊(duì)列的尾節(jié)點(diǎn)。

4) HOPS 的設(shè)計(jì)意圖

/*讓 tail 節(jié)點(diǎn)永遠(yuǎn)作為隊(duì)列的尾節(jié)點(diǎn),每次都需要使用循環(huán) CAS 更新 tail 節(jié)點(diǎn)。如果能減少 CAS
更新 tail 節(jié)點(diǎn)的次數(shù)，就能提高入隊(duì)的效率*/
public boolean offer(E e) {
     if (e == null) throw new NullPointerException();
     Node<E> n = new Node<E>(e);
     for (; ; ) {
         Node<E> t = tail;
         if (t.casNext(null, n) && casTail(t, n)) {
             return true;
         }
     }
}

使用 hops 變量來控制并減少 tail 節(jié)點(diǎn)的更新頻率，并不是每次節(jié)點(diǎn)入隊(duì)后都將 tail 節(jié)點(diǎn)更新成尾節(jié)點(diǎn)，而是當(dāng) tail 節(jié) 點(diǎn)和尾節(jié)點(diǎn)的距離大于等于常量 HOPS 的值（默認(rèn)等于 1）時(shí)才更新 tail 節(jié)點(diǎn)，tail 和尾節(jié)點(diǎn)的距離越長(zhǎng)，使用 CAS 更新 tail 節(jié)點(diǎn)的次數(shù)就會(huì)越少，但是距離越長(zhǎng)帶來的負(fù)面效果就是每次入隊(duì)時(shí)定位尾節(jié)點(diǎn)的時(shí)間就越長(zhǎng)，因?yàn)檠h(huán)體需要多循環(huán)一次來定位出尾節(jié) 點(diǎn)，但是這樣仍然能提高入隊(duì)的效率，因?yàn)閺谋举|(zhì)上來看它通過增加對(duì) volatile 變量的讀 操作來減少對(duì) volatile 變量的寫操作，而對(duì) volatile 變量的寫操作開銷要遠(yuǎn)遠(yuǎn)大于讀操作，所以入隊(duì)效率會(huì)有所提升。 private static final int HOPS = 1;

出隊(duì)列

出隊(duì)列的就是從隊(duì)列里返回一個(gè)節(jié)點(diǎn)元素，并清空該節(jié)點(diǎn)對(duì)元素的引用。

并不是每次出隊(duì)時(shí)都更新 head 節(jié)點(diǎn)，當(dāng) head 節(jié)點(diǎn)里有元素時(shí)，直接彈出 head 節(jié)點(diǎn)里的元素，而不會(huì)更新 head 節(jié)點(diǎn)。只有當(dāng) head 節(jié)點(diǎn)里沒有元素時(shí)，出隊(duì)操作才會(huì)更新 head 節(jié)點(diǎn)。這種做法也是通過 hops 變量來減少使用 CAS 更新 head 節(jié)點(diǎn)的 消耗，從而提高出隊(duì)效率。

public E poll() {
    Node<E> h = head;
    // p 表示頭節(jié)點(diǎn)，需要出隊(duì)的節(jié)點(diǎn)
    Node<E> p = h;
    for (int hops = 0; ; hops++) {
        // 獲取 p 節(jié)點(diǎn)的元素
        E item = p.getItem();
        // 如果 p 節(jié)點(diǎn)的元素不為空，使用 CAS 設(shè)置 p 節(jié)點(diǎn)引用的元素為 null, 
        // 如果成功則返回 p 節(jié)點(diǎn)的元素。
        if (item != null && p.casItem(item, null)) {
            if (hops >= HOPS) {
                // 將 p 節(jié)點(diǎn)下一個(gè)節(jié)點(diǎn)設(shè)置成 head 節(jié)點(diǎn)
                Node<E> q = p.getNext();
                updateHead(h, (q != null)q :p);
            }
            return item;
        }
        // 如果頭節(jié)點(diǎn)的元素為空或頭節(jié)點(diǎn)發(fā)生了變化，這說明頭節(jié)點(diǎn)已經(jīng)被另外
        // 一個(gè)線程修改了。那么獲取 p 節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)
        Node<E> next = succ(p);
        // 如果 p 的下一個(gè)節(jié)點(diǎn)也為空，說明這個(gè)隊(duì)列已經(jīng)空了
        if (next == null) { // 更新頭節(jié)點(diǎn)。
            updateHead(h, p);
            break;
        }
        // 如果下一個(gè)元素不為空，則將頭節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)設(shè)置成頭節(jié)點(diǎn)
        p = next;
    }
    return null;
}

首先獲取頭節(jié)點(diǎn)的元素，然后判斷頭節(jié)點(diǎn)元素是否為空，如果為空，表示另外一個(gè)線程已經(jīng)進(jìn)行了一次出隊(duì)操作將該節(jié)點(diǎn)的元素取走，如果不為空，則使用 CAS 的方式將頭節(jié)點(diǎn)的引用設(shè)置成 null，如果 CAS 成功，則直接返回頭節(jié)點(diǎn)的元素，如果不成功，表示另外一個(gè)線程已經(jīng)進(jìn)行了一次出隊(duì)操作更新了 head 節(jié)點(diǎn)，導(dǎo)致元素發(fā)生了變化，需要 重新獲取頭節(jié)點(diǎn)。

到此這篇關(guān)于Java的非阻塞隊(duì)列ConcurrentLinkedQueue解讀的文章就介紹到這了,更多相關(guān)非阻塞隊(duì)列ConcurrentLinkedQueue內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論