【Interview】深刻理解ConcurrentLinkedQueue源碼

時間 2019-11-07

標籤 Interview 深刻理解 concurrentlinkedqueue 源碼欄目 Java 简体版

原文原文鏈接

概述

ConcurrentLinkedQueue是一個基於連接節點的無邊界的線程安全隊列，它採用先進先出原則對元素進行排序，插入元素放入隊列尾部，出隊時從隊列頭部返回元素，利用CAS方式實現的
ConcurrentLinkedQueue的結構由頭節點和尾節點組成的，都是使用volatile修飾的。每一個節點由節點元素item和指向下一個節點的next引用組成，組成一張鏈表結構。
ConcurrentLinkedQueue繼承自AbstractQueue類，實現Queue接口

經常使用方法

boolean add(E e) 將指定元素插入此隊列的尾部，當隊列滿時，拋出異常
boolean contains(Object o) 判斷隊列是否包含次元素
boolean isEmpty() 判斷隊列是否爲空
boolean offer(E e) 將元素插入隊列尾部，當隊列滿時返回false
E peek() 獲取隊列頭部元素但不刪除
E poll() 獲取隊列頭部元素，並刪除
boolean remove(Object o) 從隊列中移指定元素
int size() 返回此隊列中的元素數量,須要遍歷一遍集合。判斷隊列是否爲空時，不推薦此方法

源碼分析

// 頭節點
 private transient volatile Node<E> head;
 //尾節點
 private transient volatile Node<E> tail;
 
 public ConcurrentLinkedQueue() {
 		//初始化構造時 頭結點等於尾結點
        head = tail = new Node<E>(null);
 }
 //建立一個最初包含給定 collection 元素的 ConcurrentLinkedQueue，按照此 collection 迭代器的遍歷順序來添加元素。
 public ConcurrentLinkedQueue(Collection<? extends E> c) {
        Node<E> h = null, t = null;
        for (E e : c) {
            checkNotNull(e);
            Node<E> newNode = new Node<E>(e);
            if (h == null)
                h = t = newNode;
            else {
                t.lazySetNext(newNode);
                t = newNode;
            }
        }
        if (h == null)
            h = t = new Node<E>(null);
        head = h;
        tail = t;
    }
    
  private static class Node<E> {
    volatile E item;
    volatile Node<E> next;
    //構造一個新節點
   Node(E item) {
            UNSAFE.putObject(this, itemOffset, item);
    }
    boolean casItem(E cmp, E val) {
           return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
     }
     void lazySetNext(Node<E> val) {
     UNSAFE.putOrderedObject(this, nextOffset, val);
}
    boolean casNext(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
  }
 private static final sun.misc.Unsafe UNSAFE;
 //當前結點的偏移量
 private static final long itemOffset;
 //下一個結點的偏移量
 private static final long nextOffset;
     static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = Node.class;
                itemOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("item"));
                nextOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("next"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }

複製代碼

offer入隊操做

初始化

初始化操做就是建立一個新結點，而且head和tail相等,結點的數據域爲空。
當第一次入隊操做時，檢查插入的值是否爲空，爲空則拋空指針，而後用當前的值建立一個新Node結點。而後執行死循環開始入隊操做
首先定義了兩個指針p和t,都指向tail
而後定義q結點存儲p的next指向的結點,此時p的next是爲空沒有結點的
此時q==null 條件成立。執行p.casNext(null, newNode).以cas方式把p的下一個節點指向新建立出來的結點，而後往下執行,p=t 直接返回true。此時初始化構造的結點的next指向第一次入隊成功的結點
第二次入隊操做，首先也是非空檢查，而後建立一個新結點。此時死循環入隊操做。定義了兩個指針p和t,都指向tail。定義q結點存儲p的next指向的結點,此時p的next是不爲空的，指向了上面建立的結點。因此q==null不成立。執行else操做

此時p也不等於q。p！=t不成立，p和t都是指向tail。由於不成立因此讓p=q,此時p和q都是指向第二個結點。再次循循環操做。
而後再次p和t,都指向tail。定義q結點存儲p的next指向的結點。此時p的next指向仍是空，因此q=null成立。執行p.casNext(null, newNode).以cas方式把p的next指向新建立出來的結點。
此時if (p != t)是成立的執行casTail(t, newNode); 指望值是t,更新值新建立的結點。因而更新了tail結點移動到最後添加的結點

大概的入隊流程就是這樣重複上述操做。直到入隊成功。 tail結點並非每次都是尾結點。因此每次入隊都要經過 tail定位尾結點。

public boolean offer(E e) {
    	//檢查結點是爲null，若是插入null則拋出空指針
        checkNotNull(e);
        //構造一個新結點
        final Node<E> newNode = new Node<E>(e);
        //死循換，一直到入隊成功
        for (Node<E> t = tail, p = t;;) {
        	//p表示隊列尾結點，默認狀況尾巴=結點就是taill結點
			//獲取p結點的下一個節點
            Node<E> q = p.next;
            //q爲空,說明p就是taill結點
            if (q == null) {
            	//case方式設置p(p=t)節點的next指向當前節點
                if (p.casNext(null, newNode)) {
                    if (p != t) 
                    	//p不等於t更新尾結點,
                        casTail(t, newNode); //失敗了也是沒事的，由於表示有其餘線程成功更新了tail節點
                    return true;
                }
                //其餘線程搶先完成入隊，須要從新嘗試
            }
            //q不爲空，p和相等
            else if (p == q)
                p = (t != (t = tail)) ? t : head;
            else
            	// // 在兩跳以後檢查尾部更新.
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }
複製代碼

出隊操做

public E poll() {
    	//一個標號
        restartFromHead:
        //死循環方
        for (;;) {
        	// 定義p，h兩個指針 都指向head
            for (Node<E> h = head, p = h, q;;) {
            	//獲取當前p的值
                E item = p.item;
				//值不爲空，且以cas方式設置p的item賦值爲空。兩個條件成立向下執行
                if (item != null && p.casItem(item, null)) {
                	// p和h不相等則更新頭結點，不然直接返回值
                    if (p != h) 
                    	//更新頭結點，預期值是h,當p的next指向不爲空，更新值是q,爲空則是p
                        updateHead(h, ((q = p.next) != null) ? q : p);
                        //返回當前p的值
                    return item;
                }
                //若是item爲空說明已經被出隊了，而後判斷q是否null,是空則說明當前隊列爲空了。可是q = p.next賦值語句已經執行了
                else if ((q = p.next) == null) {
                //更新頭結點，預期值h=head,更新p.此時p的item是空，說明已經被出隊了
                    updateHead(h, p);
                    return null;
                }
                else if (p == q)
                    continue restartFromHead;
                else
                    p = q;
            }
        }
    }
複製代碼

出隊操做是以死循環的方式直到出隊成功。第一次出隊首先執行for (Node<E> h = head, p = h, q;;) 定義兩個指針p和h都指向head
而後定義一個item存儲p(這裏就是head)的值，而後判斷item是否爲空，此時第一次出隊時爲空的，則執行 else if ((q = p.next) == null) ,此條件不成立，由於head的next有結點。執行 else if (p == q)，此時不相等，由於上個操做已經把q賦值爲p的next結點了。因此執行最後的else語句 p = q;在次循環執行。
此時p.item不爲空條件成立且以cas方式更新p的item爲空 p.casItem(item, null)。若是都兩個條件都成立，判斷 if (p != h)此時不成立的，更新updateHead(h, ((q = p.next) != null) ? q : p); 預期值是h,更新值是q 由於不爲空。並返回item,第一次出隊成功。

總結

CoucurrentLinkedQueue的結構由頭節點和尾節點組成的，都是使用volatile修飾的。每一個節點由節點元素item和指向下一個節點的next引用組成.
入隊:先檢查插入的值是否爲空，若是是空則拋出異常。而後以死循壞的方式執行一直到入隊成功，整個過程大概就是把tail結點的next指向新結點，而後更新tail爲新結點便可。可是tail結點並非每次都是尾結點。因此每次入隊都要經過tail定位尾結點。
出隊：出隊操做就是從隊列裏返回一個最先插入的節點元素，並清空該節點對元素的引用。並非每次出隊都更新head節點，當head節點有元素時，直接彈出head節點的元素，並以cas方式設置節點的item爲null,不會更新head節點。只有當head節點沒有元素值時，出隊操做纔會更新head節點，這種作法是爲了減小cas方式更新head節點的消耗，提供出隊的效率