《Java並發編程的藝術》這本書特別好，和《深入了解JAVA虛擬機》有一拼，建議做java的都看看，下面全部都是復制書中的部分內容，主要目的是做個筆記，方便以后遇到問題能找到。
在Java中,所有實例域、靜態域和數組元素都存儲在堆內存中,堆內存在線程之間共享。局部變量(Local Variables),方法定義參數(Java語言規范稱之為Formal Method Parameters)和異常處理器參數(ExceptionHandler Parameters)不會在線程之間共享,它們不會有內存可見性問題,也不受內存模型的影響。
Java線程之間的通信由Java內存模型(本文簡稱為JMM)控制,JMM決定一個線程對共享變量的寫入何時對另一個線程可見。從抽象的角度來看,JMM定義了線程和主內存之間的抽象關系:線程之間的共享變量存儲在主內存(Main Memory)中,每個線程都有一個私有的本地內存(Local Memory),本地內存中存儲了該線程以讀/寫共享變量的副本。本地內存是JMM的一個抽象概念,並不真實存在。它涵蓋了緩存、寫緩沖區、寄存器以及其他的硬件和編譯器優化。
如果線程A與線程B之間要通信的話,必須要經歷下面2個步驟。
1)線程A把本地內存A中更新過的共享變量刷新到主內存中去。
2)線程B到主內存中去讀取線程A之前已更新過的共享變量。
在執行程序時,為了提高性能,編譯器和處理器常常會對指令做重排序。重排序分3種類型。
1)編譯器優化的重排序。編譯器在不改變單線程程序語義的前提下,可以重新安排語句的執行順序。
2)指令級並行的重排序。現代處理器采用了指令級並行技術(Instruction-LevelParallelism,ILP)來將多條指令重疊執行。如果不存在數據依賴性,處理器可以改變語句對應機器指令的執行順序。
3)內存系統的重排序。由於處理器使用緩存和讀/寫緩沖區,這使得加載和存儲操作看上去可能是在亂序執行。
1屬於編譯器重排序,2和3屬於處理器重排序。這些重排序可能會導致多線程程序出現內存可見性問題。對於編譯器,JMM的編譯器重排序規則會禁止特定類型的編譯器重排序(不是所有的編譯器重排序都要禁止)。對於處理器重排序,JMM的處理器重排序規則會要求Java編譯器在生成指令序列時,插入特定類型的內存屏障(Memory Barriers,Intel稱之為Memory Fence)指令,通過內存屏障指令來禁止特定類型的處理器重排序。
順序一致性
順序一致性內存模型是一個理論參考模型,在設計的時候,處理器的內存模型和編程語言的內存模型都會以順序一致性內存模型作為參照。
as-if-serial語義的意思是:不管怎么重排序(編譯器和處理器為了提高並行度),(單線程)程序的執行結果不能被改變。編譯器、runtime和處理器都必須遵守as-if-serial語義。
volatile寫的內存語義：當寫一個volatile變量時,JMM會把該線程對應的本地內存中的共享變量值刷新到主內存。
volatile讀的內存語義：當讀一個volatile變量時,JMM會把該線程對應的本地內存置為無效。線程接下來將從主內存中讀取共享變量。
volatile寫和volatile讀的內存語義做個總結。
·線程A寫一個volatile變量,實質上是線程A向接下來將要讀這個volatile變量的某個線程發出了(其對共享變量所做修改的)消息。
·線程B讀一個volatile變量,實質上是線程B接收了之前某個線程發出的(在寫這個volatile變量之前對共享變量所做修改的)消息。
·線程A寫一個volatile變量,隨后線程B讀這個volatile變量,這個過程實質上是線程A通過主內存向線程B發送消息。
happens-before
1)如果一個操作happens-before另一個操作,那么第一個操作的執行結果將對第二個操作可見,而且第一個操作的執行順序排在第二個操作之前。
2)兩個操作之間存在happens-before關系,並不意味着Java平台的具體實現必須要按照happens-before關系指定的順序來執行。如果重排序之后的執行結果,與按happens-before關系
來執行的結果一致,那么這種重排序並不非法(也就是說,JMM允許這種重排序)。
上面的1)是JMM對程序員的承諾。從程序員的角度來說,可以這樣理解happens-before關系:如果A happens-before B,那么Java內存模型將向程序員保證——A操作的結果將對B可見,且A的執行順序排在B之前。注意,這只是Java內存模型向程序員做出的保證!
上面的2)是JMM對編譯器和處理器重排序的約束原則。正如前面所言,JMM其實是在遵循一個基本原則:只要不改變程序的執行結果(指的是單線程程序和正確同步的多線程程序)編譯器和處理器怎么優化都行。JMM這么做的原因是:程序員對於這兩個操作是否真的被重排序並不關心,程序員關心的是程序執行時的語義不能被改變(即執行結果不能被改變)。因此,happens-before關系本質上和as-if-serial語義是一回事。
在JMM中,如果一個操作執行的結果需要對另一個操作可見,那么這兩個操作之間必須要存在happens-before關系。這里提到的兩個操作既可以是在一個線程之內,也可以是在不同線程之間。與程序員密切相關的happens-before規則如下。
--程序順序規則:一個線程中的每個操作,happens-before於該線程中的任意后續操作。
--監視器鎖規則:對一個鎖的解鎖,happens-before於隨后對這個鎖的加鎖。
--volatile變量規則:對一個volatile域的寫,happens-before於任意后續對這個volatile域的讀。
--傳遞性:如果A happens-before B,且B happens-before C,那么A happens-before C。
注意 兩個操作之間具有happens-before關系,並不意味着前一個操作必須要在后一個操作之前執行!happens-before僅僅要求前一個操作(執行的結果)對后一個操作可見,且前一個操作按順序排在第二個操作之前(the first is visible to and ordered before the second)。
happens-before規則
1)程序順序規則:一個線程中的每個操作,happens-before於該線程中的任意后續操作。
2)監視器鎖規則:對一個鎖的解鎖,happens-before於隨后對這個鎖的加鎖。
3)volatile變量規則:對一個volatile域的寫,happens-before於任意后續對這個volatile域的讀。
4)傳遞性:如果A happens-before B,且B happens-before C,那么A happens-before C。
5)start()規則:如果線程A執行操作ThreadB.start()(啟動線程B),那么A線程的ThreadB.start()操作happens-before於線程B中的任意操作。
6)join()規則:如果線程A執行操作ThreadB.join()並成功返回,那么線程B中的任意操作happens-before於線程A從ThreadB.join()操作成功返回。
延遲加載的方案：
public class SafeDoubleCheckedLocking { private volatile static Instance instance; public static Instance getInstance() { if (instance == null) { synchronized (SafeDoubleCheckedLocking.class) { if (instance == null) instance = new Instance();// instance為volatile,現在沒問題了 } } return instance; } }
public class InstanceFactory { private static class InstanceHolder { public static Instance instance = new Instance(); } public static Instance getInstance() { return InstanceHolder.instance ; // 這里將導致InstanceHolder類被初始化 } }
隊列同步器的實現分析
從實現角度分析同步器是如何完成線程同步的,主要包括:同步隊列、獨占式同步狀態獲取與釋放、共享式同步狀態獲取與釋放以及超時獲取同步狀態等同步器的核心數據結構與模板方法。
同步隊列：
同步器依賴內部的同步隊列(一個FIFO雙向隊列)來完成同步狀態的管理,當前線程獲取同步狀態失敗時,同步器會將當前線程以及等待狀態等信息構造成為一個節點(Node)並將其加入同步隊列,同時會阻塞當前線程,當同步狀態釋放時,會把首節點中的線程喚醒,使其再次嘗試獲取同步狀態。
同步隊列中的節點(Node)用來保存獲取同步狀態失敗的線程引用、等待狀態以及前驅和后繼節點。節點的屬性類型與名稱以及描述如表所示：
節點是構成同步隊列(等待隊列,在5.6節中將會介紹)的基礎,同步器擁有首節點(head)和尾節點(tail),沒有成功獲取同步狀態的線程將會成為節點加入該隊列的尾部,同步隊列的基本結構如圖所示。
在圖中,同步器包含了兩個節點類型的引用,一個指向頭節點,而另一個指向尾節點。試想一下,當一個線程成功地獲取了同步狀態(或者鎖),其他線程將無法獲取到同步狀態,轉而被構造成為節點並加入到同步隊列中,而這個加入隊列的過程必須要保證線程安全,因此同步器提供了一個基於CAS的設置尾節點的方法:compareAndSetTail(Node expect,Nodeupdate),它需要傳遞當前線程“認為”的尾節點和當前節點,只有設置成功后,當前節點才正式與之前的尾節點建立關聯。
同步隊列遵循FIFO,首節點是獲取同步狀態成功的節點,首節點的線程在釋放同步狀態時,將會喚醒后繼節點,而后繼節點將會在獲取同步狀態成功時將自己設置為首節點。設置首節點是通過獲取同步狀態成功的線程來完成的,由於只有一個線程能夠成功獲取到同步狀態,因此設置頭節點的方法並不需要使用CAS來保證,它只需要將首節點設置成為原首節點的后繼節點並斷開原首節點的next引用即可。
獨占式同步狀態獲取與釋放
通過調用同步器的acquire(int arg)方法可以獲取同步狀態,該方法對中斷不敏感,也就是由於線程獲取同步狀態失敗后進入同步隊列中,后續對線程進行中斷操作時,線程不會從同步隊列中移出,該方法代碼：
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
上述代碼主要完成了同步狀態獲取、節點構造、加入同步隊列以及在同步隊列中自旋等待的相關工作,其主要邏輯是:首先調用自定義同步器實現的tryAcquire(int arg)方法,該方法保證線程安全的獲取同步狀態,如果同步狀態獲取失敗,則構造同步節點(獨占式Node.EXCLUSIVE,同一時刻只能有一個線程成功獲取同步狀態)並通過addWaiter(Node node)方法將該節點加入到同步隊列的尾部,最后調用acquireQueued(Node node,int arg)方法,使得該節點以“死循環”的方式獲取同步狀態。如果獲取不到則阻塞節點中的線程,而被阻塞線程的喚醒主要依靠前驅節點的出隊或阻塞線程被中斷來實現。
下面分析一下相關工作。首先是節點的構造以及加入同步隊列
private Node addWaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // 快速嘗試在尾部添加 Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; } private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
上述代碼通過使用compareAndSetTail(Node expect,Node update)方法來確保節點能夠被線程安全添加。試想一下:如果使用一個普通的LinkedList來維護節點之間的關系,那么當一個線程獲取了同步狀態,而其他多個線程由於調用tryAcquire(int arg)方法獲取同步狀態失敗而並發地被添加到LinkedList時,LinkedList將難以保證Node的正確添加,最終的結果可能是節點的數量有偏差,而且順序也是混亂的。
在enq(final Node node)方法中,同步器通過“死循環”來保證節點的正確添加,在“死循環”中只有通過CAS將節點設置成為尾節點之后,當前線程才能從該方法返回,否則,當前線程不斷地嘗試設置。可以看出,enq(final Node node)方法將並發添加節點的請求通過CAS變得“串行化”了。
節點進入同步隊列之后,就進入了一個自旋的過程,每個節點(或者說每個線程)都在自省地觀察,當條件滿足,獲取到了同步狀態,就可以從這個自旋過程中退出,否則依舊留在這個自旋過程中(並會阻塞節點的線程)
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return interrupted; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) cancelAcquire(node); } }
在acquireQueued(final Node node,int arg)方法中,當前線程在“死循環”中嘗試獲取同步狀態,而只有前驅節點是頭節點才能夠嘗試獲取同步狀態,這是為什么?原因有兩個,如下。
第一,頭節點是成功獲取到同步狀態的節點,而頭節點的線程釋放了同步狀態之后,將會喚醒其后繼節點,后繼節點的線程被喚醒后需要檢查自己的前驅節點是否是頭節點。
第二,維護同步隊列的FIFO原則。該方法中,節點自旋獲取同步狀態的行為如圖
由於非首節點線程前驅節點出隊或者被中斷而從等待狀態返回,隨后檢查自己的前驅是否是頭節點,如果是則嘗試獲取同步狀態。可以看到節點和節點之間在循環檢查的過程中基本不相互通信,而是簡單地判斷自己的前驅是否為頭節點,這樣就使得節點的釋放規則符合FIFO,並且也便於對過早通知的處理(過早通知是指前驅節點不是頭節點的線程由於中斷而被喚醒)。
獨占式同步狀態獲取流程,也就是acquire(int arg)方法調用流程
前驅節點為頭節點且能夠獲取同步狀態的判斷條件和線程進入等待狀態是獲取同步狀態的自旋過程。當同步狀態獲取成功之后,當前線程從acquire(int arg)方法返回,如果對於鎖這種並發組件而言,代表着當前線程獲取了鎖。
當前線程獲取同步狀態並執行了相應邏輯之后,就需要釋放同步狀態,使得后續節點能夠繼續獲取同步狀態。通過調用同步器的release(int arg)方法可以釋放同步狀態,該方法在釋放了同步狀態之后,會喚醒其后繼節點(進而使后繼節點重新嘗試獲取同步狀態)
public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }
該方法執行時,會喚醒頭節點的后繼節點線程,unparkSuccessor(Node node)方法使用LockSupport(在后面的章節會專門介紹)來喚醒處於等待狀態的線程。
分析了獨占式同步狀態獲取和釋放過程后,適當做個總結:在獲取同步狀態時,同步器維護一個同步隊列,獲取狀態失敗的線程都會被加入到隊列中並在隊列中進行自旋;移出隊列(或停止自旋)的條件是前驅節點為頭節點且成功獲取了同步狀態。在釋放同步狀態時,同步器調用tryRelease(int arg)方法釋放同步狀態,然后喚醒頭節點的后繼節點。
共享式同步狀態獲取與釋放
通過調用同步器的doAcquireNanos(int arg,long nanosTimeout)方法可以超時獲取同步狀態,即在指定的時間段內獲取同步狀態,如果獲取到同步狀態則返回true,否則,返回false。該方法提供了傳統Java同步操作(比如synchronized關鍵字)所不具備的特性。
在分析該方法的實現前,先介紹一下響應中斷的同步狀態獲取過程。在Java 5之前,當一個線程獲取不到鎖而被阻塞在synchronized之外時,對該線程進行中斷操作,此時該線程的中斷標志位會被修改,但線程依舊會阻塞在synchronized上,等待着獲取鎖。在Java 5中,同步器提供了acquireInterruptibly(int arg)方法,這個方法在等待獲取同步狀態時,如果當前線程被中斷,會立刻返回,並拋出InterruptedException。
超時獲取同步狀態過程可以被視作響應中斷獲取同步狀態過程的“增強版”,doAcquireNanos(int arg,long nanosTimeout)方法在支持響應中斷的基礎上,增加了超時獲取的特性。針對超時獲取,主要需要計算出需要睡眠的時間間隔nanosTimeout,為了防止過早通知,nanosTimeout計算公式為:nanosTimeout-=now-lastTime,其中now為當前喚醒時間,lastTime為上次喚醒時間,如果nanosTimeout大於0則表示超時時間未到,需要繼續睡眠nanosTimeout納秒,反之,表示已經超時。
private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException { long lastTime = System.nanoTime(); final Node node = addWaiter(Node.EXCLUSIVE); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return true; } if (nanosTimeout <= 0) return false; if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); long now = System.nanoTime(); // 計算時間,當前時間now減去睡眠之前的時間lastTime得到已經睡眠 // 的時間delta,然后被原有超時時間nanosTimeout減去,得到了 // 還應該睡眠的時間 nanosTimeout -= now - lastTime; lastTime = now; if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
該方法在自旋過程中,當節點的前驅節點為頭節點時嘗試獲取同步狀態,如果獲取成功則從該方法返回,這個過程和獨占式同步獲取的過程類似,但是在同步狀態獲取失敗的處理上有所不同。如果當前線程獲取同步狀態失敗,則判斷是否超時(nanosTimeout小於等於0表示已經超時),如果沒有超時,重新計算超時間隔nanosTimeout,然后使當前線程等待nanosTimeout納秒(當已到設置的超時時間,該線程會從LockSupport.parkNanos(Object blocker,long nanos)方法返回)。
如果nanosTimeout小於等於spinForTimeoutThreshold(1000納秒)時,將不會使該線程進行超時等待,而是進入快速的自旋過程。原因在於,非常短的超時等待無法做到十分精確,如果這時再進行超時等待,相反會讓nanosTimeout的超時從整體上表現得反而不精確。因此,在超時非常短的場景下,同步器會進入無條件的快速自旋。
獨占式超時獲取同步狀態doAcquireNanos(int arg,long nanosTimeout)和獨占式獲取同步狀態acquire(int args)在流程上非常相似,其主要區別在於未獲取到同步狀態時的處理邏輯。acquire(int args)在未獲取到同步狀態時,將會使當前線程一直處於等待狀態,而doAcquireNanos(int arg,long nanosTimeout)會使當前線程等待nanosTimeout納秒,如果當前線程在nanosTimeout納秒內沒有獲取到同步狀態,將會從等待邏輯中自動返回。
ConcurrentLinkedQueue
ConcurrentLinkedQueue由head節點和tail節點組成,每個節點(Node)由節點元素(item)和指向下一個節點(next)的引用組成,節點與節點之間就是通過這個next關聯起來,從而組成一張鏈表結構的隊列。默認情況下head節點存儲的元素為空,tail節點等於head節點。
private transient volatile Node<E> tail = head;
入隊列
入隊列就是將入隊節點添加到隊列的尾部。為了方便理解入隊時隊列的變化,以及head節點和tail節點的變化,這里以一個示例來展開介紹。
·添加元素1。隊列更新head節點的next節點為元素1節點。又因為tail節點默認情況下等於head節點,所以它們的next節點都指向元素1節點。
·添加元素2。隊列首先設置元素1節點的next節點為元素2節點,然后更新tail節點指向元素2節點。
·添加元素3,設置tail節點的next節點為元素3節點。
·添加元素4,設置元素3的next節點為元素4節點,然后將tail節點指向元素4節點。
通過調試入隊過程並觀察head節點和tail節點的變化,發現入隊主要做兩件事情:第一是將入隊節點設置成當前隊列尾節點的下一個節點;第二是更新tail節點,如果tail節點的next節點不為空,則將入隊節點設置成tail節點,如果tail節點的next節點為空,則將入隊節點設置成tail的next節點,所以tail節點不總是尾節點(理解這一點對於我們研究源碼會非常有幫助)。
通過對上面的分析,我們從單線程入隊的角度理解了入隊過程,但是多個線程同時進行入隊的情況就變得更加復雜了,因為可能會出現其他線程插隊的情況。如果有一個線程正在入隊,那么它必須先獲取尾節點,然后設置尾節點的下一個節點為入隊節點,但這時可能有另外一個線程插隊了,那么隊列的尾節點就會發生變化,這時當前線程要暫停入隊操作,然后重新獲取尾節點。讓我們再通過源碼來詳細分析一下它是如何使用CAS算法來入隊的。
public boolean offer(E e) { if (e == null) throw new NullPointerException(); // 入隊前,創建一個入隊節點 Node<E> n = new Node<E>(e); retry: // 死循環,入隊不成功反復入隊。 for (;;) { // 創建一個指向tail節點的引用 Node<E> t = tail; // p用來表示隊列的尾節點,默認情況下等於tail節點。 Node<E> p = t; for (int hops = 0;; hops++) { // 獲得p節點的下一個節點。 Node<E> next = succ(p); // next節點不為空,說明p不是尾節點,需要更新p后在將它指向next節點 if (next != null) { // 循環了兩次及其以上,並且當前節點還是不等於尾節點 if (hops > HOPS && t != tail) continue retry; p = next; } // 如果p是尾節點,則設置p節點的next節點為入隊節點。 else if (p.casNext(null, n)) { /* * 如果tail節點有大於等於1個next節點,則將入隊節點設置成tail節點, 更新失敗了也沒關系,因為失敗了表示有其他線程成功更新了tail節點 */ if (hops >= HOPS) casTail(t, n); // 更新tail節點,允許失敗 return true; } // p有next節點,表示p的next節點是尾節點,則重新設置p節點 else { p = succ(p); } } } }
從源代碼角度來看,整個入隊過程主要做兩件事情:第一是定位出尾節點;第二是使用CAS算法將入隊節點設置成尾節點的next節點,如不成功則重試。
定位尾節點
tail節點並不總是尾節點,所以每次入隊都必須先通過tail節點來找到尾節點。尾節點可能是tail節點,也可能是tail節點的next節點。代碼中循環體中的第一個if就是判斷tail是否有next節點,有則表示next節點可能是尾節點。獲取tail節點的next節點需要注意的是p節點等於p的next節點的情況,只有一種可能就是p節點和p的next節點都等於空,表示這個隊列剛初始化,正准備添加節點,所以需要返回head節點。獲取p節點的next節點代碼
final Node<E> succ(Node<E> p) { Node<E> next = p.getNext(); return (p == next) head : next; }
設置入隊節點為尾節點
p.casNext(null,n)方法用於將入隊節點設置為當前隊列尾節點的next節點,如果p是null,表示p是當前隊列的尾節點,如果不為null,表示有其他線程更新了尾節點,則需要重新獲取當前隊列的尾節點。
HOPS的設計意圖
讓tail節點永遠作為隊列的尾節點,這樣實現代碼量非常少,而且邏輯清晰和易懂。但是,這么做有個缺點,每次都需要使用循環CAS更新tail節點。如果能減少CAS更新tail節點的次數,就能提高入隊的效率,所以doug lea使用hops變量來控制並減少tail節點的更新頻率,並不是每次節點入隊后都將tail節點更新成尾節點,而是當tail節點和尾節點的距離大於等於常量HOPS的值(默認等於1)時才更新tail節點,tail和尾節點的距離越長,使用CAS更新tail節點的次數就會越少,但是距離越長帶來的負面效果就是每次入隊時定位尾節點的時間就越長,因為循環體需要多循環一次來定位出尾節點,但是這樣仍然能提高入隊的效率,因為從本質上來看它通過增加對volatile變量的讀操作來減少對volatile變量的寫操作,而對volatile變量的寫操作開銷要遠遠大於讀操作,所以入隊效率會有所提升。
private static final int HOPS = 1;
注意 入隊方法永遠返回true,所以不要通過返回值判斷入隊是否成功。
出隊列
出隊列的就是從隊列里返回一個節點元素,並清空該節點對元素的引用。讓我們通過每個節點出隊的快照來觀察一下head節點的變化。
從圖中可知,並不是每次出隊時都更新head節點,當head節點里有元素時,直接彈出head節點里的元素,而不會更新head節點。只有當head節點里沒有元素時,出隊操作才會更新head節點。這種做法也是通過hops變量來減少使用CAS更新head節點的消耗,從而提高出隊效率。讓我們再通過源碼來深入分析下出隊過程。
public E poll() { Node<E> h = head; // p表示頭節點,需要出隊的節點Node<E> p = h; for (int hops = 0;; hops++) { // 獲取p節點的元素 E item = p.getItem(); // 如果p節點的元素不為空,使用CAS設置p節點引用的元素為null, // 如果成功則返回p節點的元素。 if (item != null && p.casItem(item, null)) { if (hops >= HOPS) { // 將p節點下一個節點設置成head節點 Node<E> q = p.getNext(); updateHead(h, (q != null) q : p); } return item; } // 如果頭節點的元素為空或頭節點發生了變化,這說明頭節點已經被另外 // 一個線程修改了。那么獲取p節點的下一個節點 Node<E> next = succ(p); // 如果p的下一個節點也為空,說明這個隊列已經空了 if (next == null) { // 更新頭節點。 updateHead(h, p); break; } // 如果下一個元素不為空,則將頭節點的下一個節點設置成頭節點 p = next; } return null; }
首先獲取頭節點的元素,然后判斷頭節點元素是否為空,如果為空,表示另外一個線程已經進行了一次出隊操作將該節點的元素取走,如果不為空,則使用CAS的方式將頭節點的引用設置成null,如果CAS成功,則直接返回頭節點的元素,如果不成功,表示另外一個線程已經進行了一次出隊操作更新了head節點,導致元素發生了變化,需要重新獲取頭節點。