移步java多線程系列文章

• ConcurrentLinkedQueue 在并發(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)。

讓我們一起來研究一下Doug Lea是如何使用非阻塞的方式來實(shí)現(xiàn)線程安全隊(duì)列ConcurrentLinkedQueue的

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)行了一些修改。

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

ConcurrentLinkedQueue的類圖來分析一下它的結(jié)構(gòu)

ConcurrentLinkedQueue的類圖.jpg

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)。

private transient volatile Node<E> tail = head;

2 入隊(duì)列

2.1 入隊(duì)列的過程

• 入隊(duì)列就是將入隊(duì)節(jié)點(diǎn)添加到隊(duì)列的尾部。
• 假設(shè)我們想在一個(gè)隊(duì)列中依次插入4個(gè)節(jié)點(diǎn)
• 為了幫助大家理解，每添加一個(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)閠ail節(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)。

qq_pic_merged_1538983662965.jpg

發(fā)現(xià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ì)的角度理解了入隊(duì)過程，但是多個(gè)線程同時(shí)進(jìn)行入隊(duì)的情況就變得更加復(fù)雜了，因?yàn)榭赡軙?huì)出現(xiàn)其他線程插隊(duì)的情況。

如果有一個(gè)線程正在入隊(duì)，那么它必須先獲取尾節(jié)點(diǎn)，然后設(shè)置尾節(jié)點(diǎn)的下一個(gè)節(jié)點(diǎn)為入隊(duì)節(jié)點(diǎn)，但這時(shí)可能有另外一個(gè)線程插隊(duì)了，那么隊(duì)列的尾節(jié)點(diǎn)就會(huì)發(fā)生變化，這時(shí)當(dāng)前線程要暫停入隊(duì)操作，然后重新獲取尾節(jié)點(diǎn)。

讓我們?cè)偻ㄟ^源碼來詳細(xì)分析一下它是如何使用CAS算法來入隊(duì)的。

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.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)。獲取p節(jié)點(diǎn)的next節(jié)點(diǎn)代碼如下。

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

2.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是null，表示p是當(dāng)前隊(duì)列的尾節(jié)點(diǎn)，如果不為null，表示有其他線程更新了尾節(jié)點(diǎn)，則需要重新獲取當(dāng)前隊(duì)列的尾節(jié)點(diǎn)。

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

上面分析過對(duì)于先進(jìn)先出的隊(duì)列入隊(duì)所要做的事情是將入隊(duì)節(jié)點(diǎn)設(shè)置成尾節(jié)點(diǎn)，doug lea寫的代碼和邏輯還是稍微有點(diǎn)復(fù)雜。那么，我用以下方式來實(shí)現(xiàn)是否可行？

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

• 讓tail節(jié)點(diǎn)永遠(yuǎn)作為隊(duì)列的尾節(jié)點(diǎn)，這樣實(shí)現(xiàn)代碼量非常少，而且邏輯清晰和易懂。但是，這么做有個(gè)缺點(diǎn)，每次都需要使用循環(huán)CAS更新tail節(jié)點(diǎn)。
• 如果能減少CAS更新tail節(jié)點(diǎn)的次數(shù)，就能提高入隊(duì)的效率，所以doug lea使用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ì)方法永遠(yuǎn)返回true，所以不要通過返回值判斷入隊(duì)是否成功。

3 出隊(duì)列

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

• 讓我們通過每個(gè)節(jié)點(diǎn)出隊(duì)的快照來觀察一下head節(jié)點(diǎn)的變化

  
  
  qq_pic_merged_1538992600176.jpg

• 從圖中可知，并不是每次出隊(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ì)效率。

• 首先獲取頭節(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)。

參考

《java并發(fā)編程的藝術(shù)》