第三章 Java內存模型

3.1 Java內存模型的基礎

• 通信
  在共享內存的模型里，通過寫-讀內存中的公共狀態進行隱式通信；在消息傳遞的并發模型里，線程之間必須通過發送消息來進行顯示的通信。
• 同步
  在共享內存并發模型里，同步是顯示進行的，程序員必須顯示指定某個方法或者某段代碼需要在線程之間互斥執行；在消息傳遞的并發模型里，由于消息的發送必須在接收之前，因此同步是隱式進行的。

在Java中，所有實例域、靜態域和數組元素都存儲在堆內存中，堆內存在線程之間共享；局部變量、方法定義參數和異常處理器參數不會在線程之間共享。
從抽象角度來看，JMM定義了線程和主內存之間的抽象關系：線程之間的共享變量存儲在主內存中，每個線程都有一個私有的本地內存，本地內存涵蓋了緩存、寫緩沖區、寄存器以及其它的硬件和編譯器優化。
JMM通過控制主內存與每個線程的本地內存之間的交互，來為Java程序員提供內存可見性保證。

重排序

指編譯器和處理器為了優化程序性能而對指令序列進行重新排序的一種手段：

• 編譯器優化的重排序：編譯器在不改變單線程程序語義的前提下，重新安排語句的執行順序。
• 處理器的指令級并行的重排序：如果不存在數據依賴性，處理器可以改變語句對應機器指令的執行順序。
• 內存系統的重排序：由于處理器使用緩存和讀/寫緩沖區，這使得加載和存儲操作看上去可能是在亂序執行。

JMM的編譯器重新排序規則會禁止特定類型的編譯器重排序，對于處理器重排序，JMM的處理器重排序規則會要求Java編譯器在生成指令時，插入特定類型的內存屏障。
現代的處理器使用寫緩沖區臨時保存向內存寫入的數據，但每個處理器上的寫緩沖區，僅僅對它所在的處理器可見。
由于寫緩沖區僅對自己的處理器可見，它會導致處理器執行內存操作的順序可能會與內存實際的操作執行順序不一致，由于現代的處理器都會使用寫緩沖區，因此現代的處理器都會允許對寫-讀操作進行重排序，但不允許對存在數據依賴的操作做重排序。

happens-before簡介

用來闡述操作之間的內存可見性，如果一個操作執行的結果需要對另一個操作可見，那么這兩個操作必須要存在happens-before關系，這兩個操作既可以在一個線程之內，也可以在不同線程之間，但并不等于前一個操作必須要在后一個操作之前執行。

數據依賴性

編譯器和處理器不會改變存在數據依賴關系的兩個操作的執行順序，但是僅針對單個處理器中執行的指令序列和單個線程中執行的操作。

as-if-serial

無論怎么重排序，單線程程序的執行結果不能改變。

在單線程中，對存在控制依賴的操作重排序，不會改變執行結果；但在多線程程序中，對存在控制依賴的操作重排序，可能會改變程序的執行結果。

順序一致性

順序一致性是一個理論參考模型，在設計的時候，處理器的內存模型和編程語言的內存模型都會以順序一致性內存作為參照。
如果程序是正確同步的，程序的執行將具有順序一致性：即程序的執行結果與該程序在順序一致性內存模型中的執行結果相同。
如果程序是正確同步的，程序的執行將具有順序一致性：即程序的執行結果與該程序在順序一致性內存模型中的執行結果相同。
順序一致模型有兩大特性：

• 一個線程中的所有操作必須按照程序的順序來執行。
• 所有線程都只能看到一個單一的操作執行順序，在順序一致性內存模型中，每個操作都必須原子執行且立刻對所有線程可見。

對于未同步或未正確同步的多線程程序，JMM只提供最小安全性：線程執行時讀取到的值，要么是之前某個線程寫入的值，要么是默認值。
JMM不保證未同步程序的執行結果與該程序在順序一致性模型中的執行結果一致。
未同步程序在兩個模型中的執行特征有如下差異：

• 順序一致性模型保證單線程內的操作會按程序的順序執行，而JMM不保證單線程內的操作會按程序的順序執行。
• 順序一致性模型保證所有線程只能看到一致的操作執行順序，而JMM不保證所有線程能看到一致的操作執行順序。
• JMM不保證對64位的long/double型變量的寫操作具有原子性，而順序一致性模型保證對所有內存讀/寫操作都具有原子性。

第四章 Java并發編程基礎

• 現代操作系統調度的最小單元是線程，也叫輕量級進程，在一個進程里可以創建多個線程，這些線程都擁有各自的計數器、堆棧和局部變量等特性，并且能夠訪問共享的內存變量。
• 設置線程優先級時，針對頻繁阻塞（休眠或者I/O操作）的線程需要設置較高優先級，而偏重計算（需要較多CPU時間或者偏運算）的線程則設置較低的優先級，確保處理器不會被獨占。
• 線程在運行的生命周期中可能處于以下6種不同的狀態：
• New：初始狀態，線程被創建，但是沒有調用start()方法。
• Runnable：運行狀態，Java線程將操作系統中的就緒和運行兩種狀態統稱為“運行中”。
• Blocked：阻塞狀態，表示線程阻塞于鎖。
• Waiting：等待狀態，表示線程進入等待狀態，進入該狀態表示當前線程需要等待其它線程做出一些指定動作（通知或中斷）。
• Time_Waiting：超時等待狀態，可以在指定的時間自行返回。
• Terminated：終止狀態，表示當前線程已經執行完畢。
• 中斷可以理解為線程的一個標識位屬性，它標識一個運行中的線程是否被其它線程進行了中斷操作。中斷好比其他線程對該線程打了一個招呼，其他線程通過調用該線程的interrupt()方法對其進行中斷操作。
• 線程通過檢查自身是否被中斷來進行響應，線程通過方法isInterrupt來進行判斷是否被中斷，也可以調用靜態方法Thread.interrupt對當前線程的中斷標識位進行復位，如果該線程已經處于終止狀態，即使該線程被中斷過，在調用該線程對象的isInterrupt時依舊返回false。
• 在拋出InterruptedException異常之前，Java虛擬機會先將該線程的中斷標識位清除。
• 中斷狀態是線程的一個標識位，而中斷操作是一種簡便的線程間交互方式，而這種交互方式最適合用來取消或停止任務，除了中斷之外，還可以利用一個boolean變量來控制是否需要停止任務并終止該線程。
• Java支持多個線程同時訪問一個對象或者對象的成員變量，由于每個線程可以擁有這個變量的拷貝，所以在程序的執行過程中，一個線程看到的變量并不一定是最新的。
• volatile可以用來修飾字段，就是告知程序任何對該變量的訪問需要從共享內存中獲取，而對它的改變必須同步刷新回共享內存，它能保證所有線程對變量訪問的可見性。
• synchronized可以修飾方法或者以同步塊的形式來進行使用，它主要確保多個線程在同一時刻，只能有一個線程處于方法或者同步塊中，它保證了線程對變量訪問的可見性和排他性。
• 任意線程對Object（Object由synchronized保護）的訪問，首先要獲得Object的監視器，如果獲取失敗，線程進入同步隊列，線程狀態變為Blocked，當訪問Object的前驅（獲得了鎖的線程）釋放了鎖，則該釋放操作喚醒阻塞在同步隊列中的線程，使其重新嘗試對監視器的獲取。
• 等待/通知的相關方法：
• notify()：通知一個在對象上等待的線程，使其從wait()方法返回，而返回的前提是該線程獲取到了對象上的鎖。
• notifyAll()：通知所有等待在該對象上的鎖。
• wait()：調用該方法的線程進入Waiting狀態，只有等待另外線程的通知或被中斷才會返回，調用wait()方法后，會釋放對象的鎖。
• wait(long)：超時等待一段時間，如果沒有通知就返回。
• wait(long, int)：對于超時時間更精細粒度的控制，可以達到納秒。
• 兩個線程通過對象來完成交互，而對象上的wait和notify/notifyAll()的關系就如同開關信號一樣，用來完成等待方和通知方之間的交互工作。
• 等待/通知的經典范式：
• 等待方
  (1) 獲取對象的鎖。
  (2) 如果條件不滿足，那么調用對象的wait()方法，被通知后仍要檢查條件。
  (3) 條件滿足則執行對應的邏輯。

synchronized(對象) {while(條件不滿足) {對象.wait();}對應的處理邏輯; }

• 通知方
  (1) 獲得對象的鎖
  (2) 改變條件
  (3) 通知所有等待在該對象上的線程。

synchronized(對象) {改變條件;對象.notifyAll(); }

• 管道輸入/輸出流用于線程之間的數據傳輸，而傳輸的媒介為內存，主要包括了以下4種實現：PipedOutputStream、PipeInputStream、PipedReader、PipedWriter，前兩種面向字節，后兩種面向字符。
• 如果一個線程A執行了Thread.join()，其含義是：當前線程A等待Thread線程終止后，才從Thread.join返回，線程Thread除了提供join()方法外，還提供了join(long millis)和join(long millis, int nanos)兩個具備超時特性的方法，如果在給定的超時時間內沒有終止，那么將會從超時方法中返回。
• ThreadLocal，即線程變量，是一個以ThreadLocal對象為鍵、任意對象為值的存儲結構，這個結構被附帶在線程上，也就是說一個線程可以根據一個ThreadLocal對象查詢到綁定在這個線程上的一個值，可以通過set(T)方法來設置一個值，在當前線程下再通過get()方法獲取到原先設置的值。

第五章 Java中的鎖

5.1 Lock接口

• 鎖是用來控制多個線程訪問共享資源的方式，雖然它缺少了隱式獲取釋放鎖的便捷性，但是卻擁有了鎖獲取與釋放的可操作性、可中斷地獲取鎖以及超時獲取鎖等多種synchronized關鍵字不具備的同步特性。

• 在finally塊中釋放鎖，目的是保證在獲取到鎖之后，最終能夠被釋放。

• Lock接口提供的synchronized關鍵字不具備的主要特性

• 嘗試非阻塞地獲取鎖：當前線程嘗試獲取鎖，如果這一時刻沒有被其它線程獲取到，則成功獲取并持有鎖。

• 能被中斷地獲取鎖：與synchronized不同，獲取到鎖的線程能夠響應中斷，當獲取到鎖的線程被中斷時，中斷異常將會被拋出，同時鎖會被釋放。

• 在指定的截止時間之前獲取鎖：如果截止時間到了仍舊無法獲取鎖，則返回。

• Lock的API

• void lock()：獲取鎖，調用該方法當前線程將會獲取鎖，當鎖獲得后，從該方法返回。

• void lockInterruptibly()：可中斷地獲取鎖，該方法會響應中斷，即在鎖的獲取中可以中斷當前線程。

• boolean tryLock()：嘗試非阻塞地獲取鎖，調用該方法后立刻返回，如果能夠獲取則返回true，否則返回false。

• boolean tryLock(long time, TimeUnit unit) throws InterruptedException：當前線程在超時時間內獲得了鎖；當前線程在超時時間內被中斷；超時時間結束，返回false。

• void unlock()：釋放鎖。

• Condition newCondition()：獲取等待/通知組件，該組件和當前的鎖綁定，當前線程只有獲得了鎖，才能調用該組件的wait()方法，而調用后，當前線程將釋放鎖。

5.2 隊列同步器

5.2.1 隊列同步器接口

• 隊列同步器AbstractQueuedSynchronizer，是用來構建鎖或者其它同步組件的基礎框架，它使用了一個int成員變量表示同步狀態，通過內置的FIFO隊列來完成資源獲取線程的排隊工作。
• 同步器是實現鎖的關鍵，在鎖的實現中聚合同步器，利用同步器實現鎖的語義。可以理解二者之間的關系：鎖是面向使用者的，它定義了使用者與鎖交互的接口，隱藏了實現細節；同步器面向鎖的實現者，它簡化了鎖的實現方式，屏蔽了同步狀態管理、線程的排隊、等待與喚醒等底層操作。鎖和同步器很好地隔離了使用者和實現者所需關注地領域。
• 同步器的設計是基于模板方法模式，使用者需要繼承同步器并重寫指定的方法，隨后將同步器組合在自定義同步組件的實現中，并調用同步器的模板方法，而這些模板方法將會調用使用者重載的方法。
• 重寫同步器指定的方法時，需要使用同步器提供的3個方法來訪問或者修改同步狀態：
• getState()：獲取當前同步狀態。
• setState(int newState)：設置當前同步狀態。
• compareAndSetState(int except, int update)：使用CAS設置當前狀態，該方法能夠保證狀態設置的原始性。
• 同步器提供的模板方法基本上分為以下3類：
• 獨占式獲取與釋放同步狀態
• 共享式獲取與釋放同步狀態
• 查詢同步隊列中的等待線程情況。

5.2.2 隊列同步器的實現分析

5.2.2.1 同步隊列

• 同步器依賴內部的同步隊列來完成同步狀態的管理，當前線程獲取同步狀態失敗時，同步器會將當前線程以及等待狀態等信息構造稱為一個節點，并將其加入同步隊列，同時會阻塞當前線程，當同步狀態釋放時，會把首節點中的線程喚醒，使其再次嘗試獲取同步狀態。
• 同步器中包含了兩個節點類型的引用，一個指向頭節點，而另一個指向尾節點。
• 當一個線程成功地獲取了同步狀態，其他線程將無法獲取到同步狀態，轉而被構造成為節點并加入到同步隊列當中，而這個加入到隊列地過程必須要保證線程安全，因此同步器提供了一個基于CAS的設置尾節點的方法。
• 同步隊列遵循FIFO，首節點是獲取同步狀態成功的節點，首節點的線程在釋放同步狀態時，將會喚醒后繼節點，而后繼節點將會在獲取同步狀態成功時將自己設置為首節點。

5.2.2.2 獨占式同步狀態獲取與釋放

• 通過調用同步器的acquire(int arg)方法可以獲取同步狀態，該方法對中斷不敏感，即由于線程獲取同步狀態失敗而進入同步隊列后，后續對線程進行中斷操作時，線程不會從同步隊列中移除。

public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

它的主要邏輯是：

• （1）調用自定義同步器實現的tryAcquire方法，該方法保證線程安全的獲取同步狀態，這個方法需要隊列同步器的實現者來重寫。
• （2）如果同步狀態獲取失敗，則構造同步節點（獨占式Node.EXCLUSIVE）并通過addWaiter(Node node)方法將該節點加入到同步隊列的尾部。

private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred;//1.確保節點能夠線程安全地被添加if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}//2.通過死循環來確保節點的正確添加，在"死循環"中只有通過`CAS`將節點設置為尾節點之后，當前線程才能從該方法返回，否則當前線程不斷地進行嘗試。enq(node);return node;}private Node enq(final Node node) {for (;;) {Node t = tail;if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;} else {node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}}

• （3）最后調用acquireQueued(Node node, int arg)方法，使得該節點以死循環的方式獲取同步狀態。

final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {//1.得到當前節點的前驅節點final Node p = node.predecessor();//2.如果當前節點的前驅節點是頭節點，只有在這種情況下獲取同步狀態成功if (p == head && tryAcquire(arg)) {//3.將當前節點設為頭節點setHead(node);p.next = null; // help GCfailed = false;return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}

• 可以看到，當前線程在“死循環”中嘗試獲取同步狀態，而只有前驅節點是頭節點才能夠嘗試獲取同步狀態，這是由于：

  • 頭節點是成功獲取到同步狀態的節點，而頭節點的線程釋放了同步狀態后，將會喚醒其后繼節點，后繼節點的線程被喚醒后需要檢查自己的前驅節點是否是頭節點。
  • 維護同步隊列的FIFO原則，通過簡單地判斷自己的前驅是否為頭節點，這樣就使得節點的釋放規則符合FIFO，并且也便于對過早通知的處理（過早通知是指前驅節點不是頭節點的線程由于中斷而被喚醒）

• 當同步狀態獲取成功之后，當前線程從acquire(int arg)方法返回，如果對于鎖這種并發組件而言，代表著當前線程獲取了鎖。

• 通過調用同步器的release(int arg)方法可以釋放同步狀態，該方法執行時，會喚醒頭節點的后繼節點線程，unparkSuccessor(Node node)方法使用LockSupport來喚醒處于等待狀態的線程。

public final boolean release(int arg) {if (tryRelease(arg)) {Node h = head;if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}return false;}

• （4）如果獲取不到，則阻塞節點中的線程，而被阻塞線程的喚醒主要依靠前驅節點的出隊或阻塞線程被中斷來實現。

總結：
1.在獲取同步狀態時，同步器維護一個同步隊列，獲取狀態失敗的線程都會被加入到隊列中進行自旋；
2.移出隊列（或停止自旋）的條件是前驅節點為頭節點且成功獲取了同步狀態。
3.在釋放同步狀態時，同步器調用tryRelease(int arg)方法來釋放同步狀態，然后喚醒頭節點的后繼節點。

5.2.2.3 共享式同步狀態獲取與釋放

• 共享式獲取和獨占式獲取最主要的區別在于同一時刻能夠有多個線程同時獲取到同步狀態。
• 通過調用同步器的acquireShared(int arg)方法可以共享式地獲取同步狀態：

public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0)doAcquireShared(arg);}private void doAcquireShared(int arg) {final Node node = addWaiter(Node.SHARED);boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCif (interrupted)selfInterrupt();failed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)cancelAcquire(node);}}

tryAcquireShared返回int類型，如果同步狀態獲取成功，那么返回值大于等于0，否則進入自旋狀態；成功獲取到同步狀態并退出自旋狀態的條件是當前節點的前驅節點為頭節點，并且返回值大于等于0.

• 共享式獲取，通過調用releaseShared(int arg)方法釋放同步狀態，tryReleaseShared必須要確保同步狀態線程安全釋放，一般是通過循環或CAS來保證的，因為釋放同步狀態的操作會同時來自多個線程。

public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;}

5.2.2.4 獨占式超時獲取同步狀態

• 通過調用同步器的doAcquireNanos(int arg, long nanosTimeout)方法可以超時獲取同步狀態，即在指定的時間段內獲取同步狀態。
• 在此之前，一個線程如果獲取不到鎖而被阻塞在synchronized之外，對該線程進行中斷操作，此時線程中斷的標志位會被修改，但線程依舊會阻塞在synchronized上；如果通過acquireInterruptibly(int arg)方法獲取，如果在等待過程中被中斷，會立刻返回，并拋出InterruptedException異常。

private boolean doAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {if (nanosTimeout <= 0L)return false;//1.計算出截止時間.final long deadline = System.nanoTime() + nanosTimeout;//2.加入節點final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {for (;;) {//3.取出前驅節點final Node p = node.predecessor();//4.如果獲取成功則直接返回if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return true;}nanosTimeout = deadline - System.nanoTime();//5.如果到了超時時間，則直接返回if (nanosTimeout <= 0L)return false;if (shouldParkAfterFailedAcquire(p, node) &&nanosTimeout > spinForTimeoutThreshold)LockSupport.parkNanos(this, nanosTimeout);//6.如果在自旋過程中被中斷，那么拋出異常返回if (Thread.interrupted())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}}

通過上面的代碼可以知道，它和獨占式獲取的區別在于未獲取到同步狀態時的處理邏輯：獨占式獲取在獲取不到是會一直自旋等待；而超時獲取則會使當前線程等待nanosTimeout納秒，如果當前線程在這個時間內沒有獲取到同步狀態，將會從等待邏輯中自動返回。

5.2.2.5 自定義同步組件 - TwinsLock

TwinsLock只允許至多兩個線程同時訪問，超過兩個線程的訪問將會被阻塞。

public class TwinsLock implements Lock {private final Sync sync = new Sync(2);private static final class Sync extends AbstractQueuedSynchronizer {Sync(int count) {//初始值為2.setState(count);}@Overrideprotected int tryAcquireShared(int arg) {for(;;) {//1.獲得當前的狀態.int current = getState();//2.newCount表示剩余可獲取同步狀態的線程數int newCount = current - arg;//3.如果小于0,那么返回獲取同步狀態失敗;否則通過CAS確保設置的正確性.if (newCount < 0 || compareAndSetState(current, newCount)) {//4.當返回值大于等于0表示獲取同步狀態成功.return newCount;}}}@Overrideprotected boolean tryReleaseShared(int arg) {for (;;) {int current = getState();//將可獲取同步狀態的線程數加1.int newCount = current + current;if (compareAndSetState(current, newCount)) {return true;}}}}@Overridepublic void lock() {sync.acquireShared(1);}@Overridepublic void unlock() {sync.releaseShared(1);}@Overridepublic boolean tryLock() {return false;}@Overridepublic boolean tryLock(long time, @NonNull TimeUnit unit) throws InterruptedException {return false;}@Overridepublic void lockInterruptibly() throws InterruptedException {}@NonNull@Overridepublic Condition newCondition() {return null;} }

測試用例：

public static void createTwinsLock() {final Lock lock = new TwinsLock();class TwinsLockThread extends Thread {@Overridepublic void run() {Log.d(TAG, "TwinsLockThread, run=" + Thread.currentThread().getName());while (true) {lock.lock();try {Thread.sleep(1000);Log.d(TAG, "TwinsLockThread, name=" + Thread.currentThread().getName());Thread.sleep(1000);} catch (Exception e) {e.printStackTrace();} finally {Log.d(TAG, "TwinsLockThread, unlock=" + Thread.currentThread().getName());lock.unlock();}}}}for (int i = 0; i < 10; i++) {Thread thread = new TwinsLockThread();thread.start();}}

5.3 重入鎖

• 重入鎖ReentrantLock表示該鎖能夠支持一個線程對資源的重復加鎖。
• 如果在絕對時間上，先對鎖獲取的請求一定先被滿足，那么這個鎖是公平的，公平地獲取鎖，也就是等待時間最長的線程最優先地獲取鎖。

5.3.1 實現重進入

重進入需要解決兩個問題：

• 線程再次獲取鎖，鎖需要去識別獲取鎖地線程是否為當前占據鎖的線程，如果是，則再次獲取成功。
• 鎖的最終釋放，線程重復n次獲取了鎖，隨后在第n次釋放該鎖后，其它線程能夠獲取到該鎖。

5.3.2 公平與非公平鎖的區別

• 公平與否是針對獲取鎖而言的，如果一個鎖是公平的，那么鎖的獲取順序就應該符合請求的絕對時間順序，即FIFO。
• 公平鎖的區別在于加入了同步隊列中當前節點是否有前驅節點的判斷，如果該方法返回true，表示有線程比當前線程更早地請求獲取鎖，因此需要等待前驅線程獲取并釋放鎖之后才能繼續獲取鎖；而對于非公平鎖，只要CAS設置同步狀態成功即可。
• 因此，公平鎖每次都是從同步隊列中的第一個節點獲取到鎖，而非公平鎖出現了一個線程連續獲取鎖的情況。
• 非公平鎖可能使線程饑餓，但其極少的線程切換，保證了更大的吞吐量。

5.4 讀寫鎖

• 之前提到的鎖都是排它鎖，這些鎖在同一時刻只允許一個線程進行訪問，而讀寫鎖在同一時刻可以允許多個讀線程訪問，但是在寫線程訪問時，所有的讀線程和其他寫線程均被阻塞。讀寫鎖維護了一對鎖，一個讀鎖和一個寫鎖，通過分離讀鎖和寫鎖，使得并發性有很大提升。
• 并發包提供的讀寫鎖的實現是ReentrantReadWrireLock，它支持公平性選擇、重進入、鎖降級（寫鎖能夠降級為讀鎖）。

ReadWriteLock僅定義了獲取讀鎖和寫鎖的兩個方法，即readLock和writeLock，而其實現ReentrantReadWriteLock：

• getReadLockCount：返回當前讀鎖被獲取的次數。
• getReadHoldCount：返回當前線程獲取讀鎖的次數。
• isWriteLocked：判斷寫鎖是否被獲取。
• getWriteHoldCount：返回當前線程獲取寫鎖的次數。

下面是一個讀寫鎖的簡單用例：

public class ReadWriteCache {static Map<String, Object> map = new HashMap<>();static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();static Lock r = rwl.readLock();static Lock w = rwl.writeLock();public static Object get(String key) {r.lock();try {return map.get(key);} finally {r.unlock();}}public static Object put(String key, Object value) {w.lock();try {return map.put(key, value);} finally {w.unlock();}}public static void clear() {w.lock();try {map.clear();} finally {w.unlock();}} }

5.4.2 讀寫鎖的實現分析

• 讀寫狀態的設計
  讀寫鎖需要在同步狀態（一個整形變量，高16表示讀，低16表示寫）上維護多個讀線程和一個寫線程的狀態，使得該狀態的設計成為讀寫鎖實現的關鍵。
• 寫鎖的獲取與釋放
  寫鎖是一個支持重進入的排它鎖，如果當前線程已經獲取了寫鎖，則增加寫狀態。如果當前線程在獲取寫鎖時，讀鎖已經被獲取，則當前線程進入等待狀態。
  原因在于：讀寫鎖要確保寫鎖的操作對讀鎖可見，如果允許讀鎖在已經被獲取的情況下對寫鎖的獲取，那么正在運行的其它讀線程就無法感知到當前寫線程的操作。
• 讀鎖的獲取與釋放
  讀鎖是一個支持重進入的共享鎖，它能被多個線程同時獲取，在沒有其它寫線程訪問（或者寫狀態為0）時，讀鎖總是被成功地獲取，而所做的也只是（線程安全）增加讀狀態。
• 鎖降級
  鎖降級是指把持住（當前擁有的）寫鎖，再獲取到讀鎖，隨后釋放（先前擁有的）寫鎖的過程。

5.6 Condition接口

Condition定義了等待/通知兩種類型的方法，當前線程調用這些方法時，需要提前獲取到Condition對象關聯的鎖，Condition是依賴Lock對象的。
當調用await()方法后，當前線程會釋放鎖并在此等待，而其他線程調用Condition對象的signal方法，通知當前線程后，當前線程才從await方法返回，并且在返回前已經獲取了鎖。
獲取一個Condition必須通過Lock的newCondition方法，下面是一個有界隊列的示例：

public class BoundedQueue<T> {private Object[] items;private int addIndex, removeIndex, count;private Lock lock = new ReentrantLock();private Condition notEmpty = lock.newCondition();private Condition notFull = lock.newCondition();public BoundedQueue(int size) {items = new Object[size];}public void add(T t) throws InterruptedException {lock.lock();try {while (count == items.length) { //如果當前隊列內的個數等于最大長度,那么釋放鎖.notFull.await();}if (++addIndex == items.length) { //如果已經到了尾部,那么從頭開始.addIndex = 0;}++count;notEmpty.signal(); //通知阻塞在"空"條件上的線程.} finally {lock.unlock();}}public T remove() throws InterruptedException {lock.lock();try {while (count == 0) {notEmpty.await(); //如果當前隊列的個數等于0,那么釋放鎖.}Object x = items[removeIndex];if (++removeIndex == items.length) {removeIndex = 0;}--count;notFull.signal(); //通知阻塞在"滿"條件上的線程.return (T) x;} finally {lock.unlock();}} }

Condition的方法：

• await()：當前線程進入等待狀態直到被通知signal或中斷，當前線程進入運行狀態且從await返回的情況：

• 其他線程調用該Condition的signal或signalAll方法。

• 其它線程中斷當前線程（interrupt）。

• 如果當前等待線程從await方法返回，那么表明當前線程已經獲取了Condition對象所對應的鎖。

• awaitUninerruptibly：對中斷不敏感

• long await Nanos(long)：加入了超時的判斷，返回值是（nanosTimeout - 實際耗時），如果返回值是0或者負數，那么可以認定為超時。

• boolean awaitUntil(Data)：直到某個固定時間。

• signal：喚醒一個等待在Condition上的線程。

• signalAll：喚醒所有等待在Condition上的線程。

5.6.2 Condition的實現

ConditionObject是AbstractQueuedSynchronizer的內部類，每個Condition對象都包含著一個隊列。

1.等待隊列

在隊列中的每個節點都包含了一個線程的引用，該線程就是在Condition對象上等待的線程，同步隊列和等待隊列中節點的類型都是同步器的靜態內部類AbstractQueuedSynchronizer.Node。
由于Condition的實現是同步器的內部類，因此每個Condition實例都能夠訪問同步器提供的方法，相當于每個Condition都擁有所屬同步器的引用。
當調用await方法時，將會以當前線程構造節點，并將節點從尾部加入到等待隊列，也就是將同步隊列移動到Condition隊列當中。

2.等待

調用該方法的前提是當前線程必須獲取了鎖，也就是同步隊列中的首節點，它不是直接加入到等待隊列當中，而是通過addConditionWaiter()方法把當前線程構造成一個新的節點并將其加入到等待隊列當中。

3.通知

調用該方法的前提是當前線程必須獲取了鎖，接著獲取等待隊列的首節點，將其移動到同步隊列并使用LockSupport喚醒節點中的線程。
被喚醒的線程，將從await方法中的while中返回，進而調用同步器的acquireQueued方法加入到獲取同步狀態的競爭中。
Condition的signalAll方法，相當于對等待隊列中的每個節點均執行一次signal方法，效果就是將等待隊列中所有節點全部移動到同步隊列中，并喚醒每個節點。

六、Java并發容器和框架

6.1 ConcurrentHashMap

ConcurrentHashMap是線程安全并且高效的HashMap，其它的類似容器有以下缺點：

• HashMap在并發執行put操作時，會導致Entry鏈表形成環形數據結構，就會產生死循環獲取Entry。
• HashTable使用synchronized來保證線程安全，但在線程競爭激烈的情況下HashTable的效率非常低下。
  ConcurrentHashMap高效的原因在于它采用鎖分段技術，首先將數據分成一段一段地存儲，然后給每段數據配一把鎖，當一個線程占用鎖并且訪問一段數據的時候，其他段的數據也能被其他線程訪問。

6.1.2 ConcurrentHashMap的結構

ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成：

• Segment是一種可重入鎖，在ConcurrentHashMap里面扮演鎖的角色；
• HashEntry則用于存儲鍵值對數據。

一個ConcurrentHashMap里包含一個Segment數組，它的結構和HashMap類似，是一種數組和鏈表結構。
一個Segment里包含一個HashEntry數組，每個HashEntry是一個鏈表結構的元素，每個Segment守護著一個HashEntry里的元素，當對HashEntry數組的數據進行修改時，必須首先獲得與它對應的Segment鎖。

6.1.5 ConcurrentHashMap的操作

get
get的高效在于整個get過程中不需要加鎖，除非讀到的值是空才會加鎖重讀。原因是它的get方法將要使用的共享變量都設為volatile，能夠在線程間保持可見性，能夠被多線程同時讀，并且不會讀到過期的值，例如用于統計當前Segment大小的count字段和用于存儲值的HashEntry的value。
put
put方法里需要對共享變量進行寫入操作，所以為了線程安全，在操作共享變量之前必須加鎖，put首先定位到Segment，然后在Segment里進行插入操作。
size
先嘗試2次通過不鎖住Segment的方式來統計各個Segment的大小，如果統計的過程中，容器的count發生了變化，則再用加鎖的方式來統計所有Segment的大小。

6.2 ConcurrentLinkedQueue

ConcurrentLinkedQueue是一個基于鏈接節點的無界線程安全隊列，它采用先進先出的規則對節點進行排序，它采用CAS算法來實現。

6.2.1 入隊列

入隊主要做兩件事情：

• 將入隊節點設置成當前隊列尾節點的下一個節點。
• 更新tail節點，如果tail節點的next節點不為空，則將入隊節點設置成tail節點；如果tail節點的next節點為空，則將入隊節點設置成tail的next節點。

在多線程情況下，如果有一個線程正在入隊，那么它必須先獲取尾節點，然后設置尾節點的下一個節點為入隊節點，但這時可能有另外一個線程插隊了，那么隊列的尾節點就會發生變化，這時第一個線程要暫停入隊操作，然后重新獲取尾節點。
整個入隊操作主要做兩件事：

• 定位出尾節點。
• 使用CAS算法將入隊節點設置成尾節點的next節點，如不成功則重試。

6.3 阻塞隊列

6.3.1 阻塞隊列

阻塞隊列是一個支持兩個附加操作的隊列，這兩個附加的操作支持阻塞的插入和移除方法：

• 當隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。
• 當隊列空時，獲取元素的線程會等待隊列為空。

在阻塞隊列不可用時，附加操作提供了4種處理方式：拋出異常、返回特殊值、一直阻塞、超時退出。每種方式通過調用不同的方法來實現。
Java里面提供了7種阻塞隊列。

6.4 Fork/Join框架

用于并行執行任務的框架，是把一個大任務分割成若干個小任務，最終匯總每個小任務結果后得到大人物結果的框架。
Fork/Join使用兩個類來完成事情：

• ForkJoinTask：它提供了fork()和join()操作的機制，通常情況下，我們繼承它的子類：有返回結果的RecursiveTask和沒有返回結果的RecursiveAction。
• ForkJoinPool：ForkJoinTask需要通過ForkJoinPool來添加。
  ForkJoinTask在執行的時候可能會拋出異常，但是我們沒有辦法在主線程里直接捕獲異常，所以ForkJoinTask提供了isCompletedAbnormally()方法來檢查任務是否已經拋出異常或已經取消了。
  ForkJoinPool由ForkJoinTask數組和ForkJoinWorkerThread數組組成，ForkJoinTask數組負責將存放程序提交給ForkJoinPool的任務，而ForkJoinWorkerThread數組負責執行這些任務。

七、Java中的13個原子操作類

Atomic包里提供了：原子更新基本類型、原子更新數組、原子更新引用和原子更新屬性。

7.1 原子更新基本類型：

• AtomicBoolean
• AtomicInteger
• AtomicLong

基本方法：

• int addAndGet(int delta)：以原子方式將輸入的值與當前的值相加，并返回結果。
• boolean compareAndSet(int expect, int update)：如果當前的數值等于預期值，則以原子方式將該值設置為輸入的值。
• int getAndIncrement()：以原子方式加1，并返回自增前的值。
• void lazySet(int newValue)：最終會設置成newValue，可能會導致其他線程在之后的一小段時間內還是讀到舊值。
• int getAndSet(int newValue)：以原子方式設置為newValue的值，并返回舊值。

7.2 原子更新引用類型

• AtomicIntegerArray
• AtomicLongArray
• AtomicReferenceArray

基本方法：

• int addAndGet(int i, int delta)：以原子方式將輸入值和索引i的元素相加。
• boolean compareAndSet(int i, int expect, int update)：如果當前值等于預期值，則以原子方式將數組位置i的元素設置成update值。

7.3 原子更新引用類型

用于原子更新多個變量，提供了3種類型：

• AtomicReference：原子更新引用類型。
• AtomicReferenceFieldUpdater：原子更新引用類型里的字段。
• AtomicMarkableReference：原子更新帶有標記位的引用類型。

7.4 原子更新字段類

• AtomicIntegerFieldUpdater：原子更新整形的字段的更新器。
• AtomicLongFieldUpdater：原子更新長整形字段的更新器。
• AtomicStampedReference：原子更新帶有版本號的引用類型。

原子地更新字段需要兩步：

• 因為原子更新字段類都是抽象類，每次使用的時候必須使用靜態方法newUpdater創建一個更新器，并且需要設置想要更新的類和屬性。
• 更新類的字段必須使用public volatile來修飾。

八、Java中的并發工具類

九、Java中的線程池

線程池的優點：降低資源消耗，提高響應速度，提高線程的可管理性。

9.1 線程池的實現原理

線程池的處理流程如下：

• 判斷核心線程池是否已滿，如果不是，則創建一個新的工作線程來執行任務；如果已滿，則進入下個流程。
• 判斷工作隊列是否已滿，如果不是，則將提交的任務存儲在工作隊列里；如果已滿，則進入下個流程。
• 判斷線程池的線程是否都處于工作狀態，如果沒有，則創建一個新的工作線程；如果已滿，則交給飽和策略來處理。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task. The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread. If it fails, we know we are shut down or saturated* and so reject the task.*/int c = ctl.get();if (workerCountOf(c) < corePoolSize) { //1.添加進入核心線程.if (addWorker(command, true))return;c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) { //2.添加進入隊列.int recheck = ctl.get();if (!isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}else if (!addWorker(command, false)) //3.添加進入非核心線程.reject(command);}private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int rs = runStateOf(ctl.get());if (rs < SHUTDOWN ||(rs == SHUTDOWN && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

在以上的三步中，除了加入隊列不用獲取全局鎖以外，其它兩種情況都需要獲取，為了盡可能地避免獲取全局鎖，在ThreadPoolExecutor完成預熱之后（當前運行的線程數大于corePoolSize），幾乎所有的execute方法調用都是加入到隊列當中。

9.2 線程池的使用

9.2.1 線程池的創建

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

• corePoolSize：當提交一個任務到線程池時，線程池會創建一個線程來執行任務，即使其它空閑的基本線程能夠執行新任務也會創建。
• runnableTaskQueue：用于保存等待執行的任務的阻塞隊列，可以選擇：
• ArrayBlockingQueue：基于數組結構的有界阻塞隊列。
• LinkedBlockingQueue：基于鏈表結構的阻塞隊列，吞吐量高于前者。
• SynchronousQueue：不存儲元素的阻塞隊列，每個插入操作必須等待另一個線程調用了移除操作，靜態工廠方法Executors.newCachedThreadPool使用了這個隊列。
• PriorityBlockingQueue：一個具有優先級的無限阻塞隊列。
• maxPoolSize：允許創建的最大線程數。
• ThreadFactory：用于設置創建線程的工廠。
• RejectExecutionHandler：飽和策略。
• keepAliveTime：線程池的工作線程空閑后，保持存活的時間。
• TimeUnit：線程保持活動的單位。

9.2.2 向線程池提交任務

• execute(Runnable runnable)：提交不需要返回值的任務。
• Future<Object> future = executor.submit(haveReturnValuetask)：用于提交需要返回值的任務，線程池會返回一個future類型任務，可以用它來判斷任務是否執行成功，并且可以通過get方法來獲取返回值，get方法會阻塞當前線程直到任務完成。

9.2.3 關閉線程池

• shutdownNow：首先將線程池的狀態設為STOP，然后嘗試停止所有的正在執行或暫停任務的線程，并返回等待執行任務的列表。
• shutdown：將線程池的狀態置為SHUTDOWN，然后中斷所有沒有正在執行任務的線程。

十、Executor框架

（1）在上層，Java多線程程序通常把應用分解為若干個任務，然后使用用戶級的調度器（Executor框架）將這些任務映射為固定數量的線程。
（2）在HotSpot VM的線程模型中，Java線程再被一對一映射為本地操作系統線程，Java線程啟動時會創建一個本地操作系統線程，當該線程終止時，這個操作系統線程也會被回收。
（3）操作系統會調度所有線程并將它們分配給可用的CPU。

Executor框架

由三個部分組成：

• 任務，即Runnable接口或Callable接口。
• 任務的執行，包括核心接口Executor，以及繼承自Executor的ExecutorService，還有它的兩個關鍵類ThreadPoolExecutor（用來執行任務）和ScheduledThreadPoolExecutor（可以在給定的延遲后運行命令，或者定期執行命令）。
• 異步計算的結果，包括接口Future和實現類FutureTask。

10.2 ThreadPoolExecutor詳解

通過工具類Executors，可以創建以下三種類型的ThreadPoolExecutor，調用靜態創建方法之后，會返回ExecutorService

• FixedThreadPool
  可重用固定線程數的線程池；如果當前運行的線程數少于corePoolSize，則創建新線程來執行任務；如果等于corePoolSize，將任務加入到無界隊列LinkedBlockingQueue當中；多余的空閑線程將會被立即終止。
• SingleThreadPool
  單個woker線程的executor；corePoolSize和maximumPoolSize為1；采用無界隊列作為工作隊列。
• CacheThreadPool
  采用沒有容量的SynchronousQueue作為線程池的工作隊列，其corePoolSize為0，maximumPool是無界的；其中的空閑線程最多等待60s。
  如果主線程提交任務的速度高于maximumPool中線程處理任務的速度時，CacheThreadPool會不斷創建新線程，極端情況下，CacheThreadPool會因為創建過多線程而耗盡CPU資源。

10.3 ScheduledThreadPoolExecutor詳解

用來在給定的延遲之后執行任務，或者定期執行任務，并且可以在指定的構造函數中指定多個對應的后臺線程數。
它采用DelayQueue這個無界隊列作為工作隊列，其執行分為兩個部分：

• 當調用ScheduledThreadPoolExecutor的scheduleAtFixedRate()或者scheduleWithFIxedDelay，它會向DelayQueue中添加ScheduledFutureTask。
• 線程池中的線程從DelayQueue中獲取ScheduledFutureTask。

作者：澤毛
鏈接：https://www.jianshu.com/p/0d77a717c52a
來源：簡書
簡書著作權歸作者所有，任何形式的轉載都請聯系作者獲得授權并注明出處。

總結

以上是生活随笔為你收集整理的多线程知识梳理(1) - 并发编程的艺术笔记的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。