併發隊列之ConcurrentLinkedQueue

　　原本想着直接說線程池的，不過在說線程池以前，咱們必需要知道併發安全隊列；由於通常狀況下線程池中的線程數量是必定的，確定不會超過某個閾值，那麼當任務太多了的時候，咱們必須把多餘的任務保存到併發安全隊列中，當線程池中的線程空閒下來了，就會到併發安全隊列中拿任務；

　　那麼什麼是併發安全隊列呢？其實能夠簡單看做是一個鏈表，而後咱們先辦法去存取節點；總的來講，併發安全隊列分爲兩種，一種是阻塞的，一種是非阻塞的，前者是用鎖來實現的，後者用CAS實現的；

 

一.簡單介紹ConcurrentLinkedQueue

　　這個隊列用法沒什麼好說的，就相似LinkedList的用法，怎麼對一個鏈表繼續增刪改查，很少說，咱們就說一下其中幾個關鍵的方法；

　　首先，這個隊列是一個線程安全的無界非阻塞隊列，其實就是一個單向鏈表，無界的意思就是沒有限制最大長度，非阻塞表示用CAS實現入隊和出隊操做，咱們打開這個類就能夠知道，有一個內部類Node，其中重要的屬性以下所示：

//用於存放節點的值
volatile E item;
//指向下一個節點
volatile Node<E> next;
//這裏也是用的是UNSAFE類，前面說過了，這個類直接提供CAS操做
private static final sun.misc.Unsafe UNSAFE;
//item字段的偏移量
private static final long itemOffset;
//next的偏移量
private static final long nextOffset;

 

 

　　而後ConcurrentLinkedQueue中幾個重要的屬性，好像也沒什麼重要的，就保存了頭節點和尾節點，注意，默認狀況下頭節點和尾節點都是哨兵節點，也就是一個存null的Node節點

//存放鏈表的頭節點
private transient volatile Node<E> head;
//存放鏈表的尾節點
private transient volatile Node<E> tail;
//UNSAFE對象
private static final sun.misc.Unsafe UNSAFE;
//head字段的偏移量
private static final long headOffset;
//tail字段偏移量
private static final long tailOffset;

 

 

 

 

　　下面咱們直接看一些重要方法吧！慢慢分析其中的算法纔是關鍵的

 

二.offer方法

　　這個方法的做用就是在隊列末端添加一個節點，若是傳遞的參數是null，就拋出空指針異常，不然因爲該隊列是無界隊列，該方法會一直返回true，並且該方法使用CAS算法實現的，因此不會阻塞線程；

//隊列末端添加一個節點
public boolean offer(E e) {
    //若是e爲空，那麼拋出空指針異常
    checkNotNull(e);
    //將傳進來的元素封裝成一個節點，Node的構造器中調用UNSAFE.putObject(this, itemOffset, item)把e賦值給節點中的item
    final Node<E> newNode = new Node<E>(e);

    //[1] //這裏的for循環從最後的節點開始
    for (Node<E> t = tail, p = t;;) {
      Node<E> q = p.next;
      //[2]若是q爲null，說明p就是最後的節點了
        if (q == null) {
            //[3]CAS更新：若是p節點的下一個節點是null，就把寫個節點更新爲newNode
            if (p.casNext(null, newNode)) {
                //[4]CAS成功，可是這時p==t,因此不會進入到這裏的if裏面，直接返回true
                //那麼何時會走到這裏面來呢？實際上是要有另一個線程也在調用offer方法的時候，會進入到這裏面來
                if (p != t) 
                    casTail(t, newNode);  
                return true;
            }
        }
        else if (p == q) //[5]
            
            p = (t != (t = tail)) ? t : head;
        else //[6]
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

 

　　

　　上面執行到[3]的時候，因爲頭節點和尾節點默認都是指向哨兵節點的，因爲這個時候p的下一個節點爲null，因此當前線程A執行CAS會成功，下圖所示；

 

 

　　若是此時還有一個線程B也來嘗試[3]中CAS，因爲此時p節點的下一個節點不是null了，因而線程B會跳到[1]出進行第二次循環，而後會到[6]中，因爲p和t此時是相等的，因此這裏是false，即p=q，下圖所示：

 

 

　　而後線程B又會跳到[1]處進行第三次循環，因爲執行了Node<E> q = p.next，因此此時q指向最後的null，就到了[3]處進行CAS，此次是能夠成功的，成功以後以下圖所示：

 

 

 　　

　　這個時候由於p！=t，因此能夠進入到[4]，這裏又會進行一個CAS：若是tail和t指向的節點同樣，那麼就將tail指向新添加的節點，如圖所示，這個時候線程B也就執行完了；

 

 　　

　　其實還有[5]沒有走到，這個是在poll操做以後才執行的，咱們先跳過，等說完poll方法以後再回頭看看；另外說一下，add方法其實就是調用的是offer方法，就很少說了；

 

 

三.poll方法

　　這個方法是獲取頭部的這個節點，若是隊列爲空則返回null；

public E poll() {
    //這裏其實就是一個goto的標記，用於跳出for循環
    restartFromHead:
    //[1]
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;
            //[2]若是當前節點中存的值不爲空，則CAS設置爲null
            if (item != null && p.casItem(item, null)) {
                //[3]CAS成功就更新頭節點的位置
                if (p != h) 
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //[4]當前隊列爲空，就返回null
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //[5]當前節點和下一個節點同樣，說明節點自引用，則從新找頭節點
            else if (p == q)
                continue restartFromHead;
            //[6]
            else
                p = q;
        }
    }
}

final void updateHead(Node<E> h, Node<E> p) {
    if (h != p && casHead(h, p))
        h.lazySetNext(h);
}

　　

　　分爲幾種狀況，第一種狀況是線程A調用poll方法的時候，發現隊列是空的，即頭節點和尾節點都指向哨兵節點，就會直接到[4]，返回null

　　第二種狀況，線程A執行到了[4]，此時有一個線程卻調用offer方法添加了一個節點，下圖所示，那麼此時線程A就不會走[4]了，[5]也不知足，因而會到[6]這裏來，而後線程A又會跳到[1]處進行循環，此時p指向的節點中item不爲null，因此會到[2]中；

 

 　　

　　到了[2]中將p指向的節點中item用CAS設置爲null，而後就到了[3],下面第一個圖，因爲p！=h，q=null，因此最後調用的是updateHead(h,p)，這方法：若是頭節點和h指向的是同樣的，就將頭節點指向p，咱們還能看到updateHead方法中h.lazySetNext(h)表示h的下一個節點指向本身，下面圖二

 

 　　到了這裏還沒完，還記不記得offer方法中有一個地方的代碼沒有執行的啊！就是這種狀況，尾節點本身引用本身，咱們再調用offer會怎麼樣呢？

　　回到offer方法，先會到[1]，而後q指向本身這個哨兵節點（注意，此時雖然p指向的節點中存的是null，可是p！=null}，因而再到[5]，此時的圖以下左圖所示；此時因爲t==tail，因此p=head；

 

 　　再在offer方法循環一次，此時q指向null，下面左圖所示，而後就能夠進入[2]中進行CAS，CAS成功，由於此時p！=t，因此還要進行CAS將tail指向新節點，下面右圖所示，可讓GC回收那個垃圾！

媽耶，這裏比較繞！哈哈哈哈哈哈哈哈哈哈哈

 

 

 

四.peek方法

　　這個方法的做用就是獲取隊列頭部的元素，只獲取不移除，注意這個方法和上面的poll方法的區別啊！

public E peek() {
    //[1]goto標誌
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            //[2]
            E item = p.item;
            //[3]
            if (item != null || (q = p.next) == null) {
                updateHead(h, p);
                return item;
            }
            //[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);
  }

 

　　

　　若是隊列中爲空的時候，走到[3]的時候，就會以下圖所示，因爲h==p,因此updateHead方法啥也不作，而後返回就返回item爲null

 

 

　　若是隊列不爲空，那麼以下左圖所示，此時進入循環內不知足條件，會到[5]這裏，將p指向q，而後再進行一次循環到[3]，將q指向p的後一個節點，下面右圖所示；

 

 

　　而後調用updateHead方法，用CAS將頭節點指向p這裏，而後將h本身指向本身，下圖所示，最後返回item

 

 

五.總結

　　其實還有幾個方法沒說，可是感受比較容易就不浪費篇幅了，有興趣的能夠看看：size方法用於計算隊列中節點的數量，但是因爲沒有加鎖，在併發的條件下不許確；remove方法刪除某個節點，其實就是遍歷而後用equals方法比較item是否是同樣，只不過若是存在多個符合條件的節點只刪除第一個，而後返回true，不然返回false；contains方法判斷隊列中是否包含指定item的節點，也就是遍歷，很容易；

　　最麻煩的就是offer方法和poll方法，offer方法是在隊列的最後面添加節點，而poll是獲取頭節點，而且刪除第一個真正的隊列節點(注意，節點分爲兩種，一種是哨兵節點，一種是真正的存了數據的節點啊)，還簡單的說了一下poll方法和peek方法的區別，後者只是獲取，而不刪除啊！用下面這個圖幫助記憶一下；