多線程并發編程如何在Java項目中實現

本篇文章為大家展示了多線程并發編程如何在Java項目中實現 ，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

一、多線程

1、操作系統有兩個容易混淆的概念，進程和線程。

進程：一個計算機程序的運行實例，包含了需要執行的指令；有自己的獨立地址空間，包含程序內容和數據；不同進程的地址空間是互相隔離的；進程擁有各種資源和狀態信息，包括打開的文件、子進程和信號處理。

線程：表示程序的執行流程，是CPU調度執行的基本單位；線程有自己的程序計數器、寄存器、堆棧和幀。同一進程中的線程共用相同的地址空間，同時共享進進程鎖擁有的內存和其他資源。

2、Java標準庫提供了進程和線程相關的API，進程主要包括表示進程的java.lang.Process類和創建進程的java.lang.ProcessBuilder類；

表示線程的是java.lang.Thread類，在虛擬機啟動之后，通常只有Java類的main方法這個普通線程運行，運行時可以創建和啟動新的線程；還有一類守護線程（damon thread），守護線程在后臺運行，提供程序運行時所需的服務。當虛擬機中運行的所有線程都是守護線程時，虛擬機終止運行。

3、線程間的可見性：一個線程對進程中共享的數據的修改，是否對另一個線程可見

可見性問題：

a、CPU采用時間片輪轉等不同算法來對線程進行調度

public class IdGenerator{ 
 private int value = 0; 
 public int getNext(){ 
 return value++; 
 } 
} 

 對于IdGenerator的getNext()方法，在多線程下不能保證返回值是不重復的：各個線程之間相互競爭CPU時間來獲取運行機會，CPU切換可能發生在執行間隙。

以上代碼getNext()的指令序列：CPU切換可能發生在7條指令之間，多個getNext的指令交織在一起。

aload_0 
dup 
getfield #12 
dup_x1 
iconst_1 
iadd 
putfield #12 

b、CPU緩存：

目前CPU一般采用層次結構的多級緩存的架構，有的CPU提供了L1、L2和L3三級緩存。當CPU需要讀取主存中某個位置的數據時，會一次檢查各級緩存中是否存在對應的數據。如果有，直接從緩存中讀取，這比從主存中讀取速度快很多。當CPU需要寫入時，數據先被寫入緩存中，之后再某個時間點寫回主存。所以某些時間點上，緩存中的數據與主存中的數據可能是不一致。

c、指令順序重排

出行性能考慮，編譯器在編譯時可能會對字節代碼的指令順序進行重新排列，以優化指令的執行順序，在單線程中不會有問題，但在多線程可能產生與可見性相關的問題。

二、Java內存模型（Java Memory Model）

屏蔽了CPU緩存等細節，只關注主存中的共享變量；關注對象的實例域、靜態域和數組元素；關注線程間的動作。

1、volatile關鍵詞：用來對共享變量的訪問進行同步，上一次寫入操作的結果對下一次讀取操作是肯定可見的。（在寫入volatile變量值之后，CPU緩存中的內容會被寫回內存；在讀取volatile變量時，CPU緩存中的對應內容會被置為失效，重新從主存中進行讀取），volatile不使用鎖，性能優于synchronized關鍵詞。

用來確保對一個變量的修改被正確地傳播到其他線程中。

例子：A線程是Worker，一直跑循環，B線程調用setDone(true)，A線程即停止任務

public class Worker{ 
 private volatile boolean done; 
 public void setDone(boolean done){ 
 this.done = done; 
 } 
 public void work(){ 
 while(!done){ 
  //執行任務； 
 } 
 } 
} 

例子：錯誤使用。因為沒有鎖的支持，volatile的修改不能依賴于當前值，當前值可能在其他線程中被修改。（Worker是直接賦新值與當前值無關）

public class Counter { 
 public volatile static int count = 0; 
 public static void inc() { 
 //這里延遲1毫秒，使得結果明顯 
 try { 
  Thread.sleep(1); 
 } catch (InterruptedException e) { 
 } 
 count++; 
 } 
 public static void main(String[] args) { 
 //同時啟動1000個線程，去進行i++計算，看看實際結果 
 for (int i = 0; i < 1000; i++) { 
  new Thread(new Runnable() { 
  @Override 
  public void run() { 
   Counter.inc(); 
  } 
  }).start(); 
 } 
 //這里每次運行的值都有可能不同,可能不為1000 
 System.out.println("運行結果:Counter.count=" + Counter.count); 
 } 
} 

2、final關鍵詞

final關鍵詞聲明的域的值只能被初始化一次，一般在構造方法中初始化。。（在多線程開發中，final域通常用來實現不可變對象）

當對象中的共享變量的值不可能發生變化時，在多線程中也就不需要同步機制來進行處理，故在多線程開發中應盡可能使用不可變對象。

另外，在代碼執行時，final域的值可以被保存在寄存器中，而不用從主存中頻繁重新讀取。

3、java基本類型的原子操作

1）基本類型，引用類型的復制引用是原子操作；（即一條指令完成）

2）long與double的賦值，引用是可以分割的，非原子操作；

3）要在線程間共享long或double的字段時，必須在synchronized中操作，或是聲明成volatile

三、Java提供的線程同步方式

1、synchronized關鍵字

方法或代碼塊的互斥性來完成實際上的一個原子操作。（方法或代碼塊在被一個線程調用時，其他線程處于等待狀態）

所有的Java對象都有一個與synchronzied關聯的監視器對象（monitor），允許線程在該監視器對象上進行加鎖和解鎖操作。

a、靜態方法：Java類對應的Class類的對象所關聯的監視器對象。

b、實例方法：當前對象實例所關聯的監視器對象。

c、代碼塊：代碼塊聲明中的對象所關聯的監視器對象。

注：當鎖被釋放，對共享變量的修改會寫入主存；當活得鎖，CPU緩存中的內容被置為無效。編譯器在處理synchronized方法或代碼塊，不會把其中包含的代碼移動到synchronized方法或代碼塊之外，從而避免了由于代碼重排而造成的問題。

例：以下方法getNext()和getNextV2() 都獲得了當前實例所關聯的監視器對象

public class SynchronizedIdGenerator{ 
 private int value = 0; 
 public synchronized int getNext(){ 
 return value++; 
 } 
 public int getNextV2(){ 
 synchronized(this){ 
  return value++; 
 } 
 } 
} 

2、Object類的wait、notify和notifyAll方法

生產者和消費者模式，判斷緩沖區是否滿來消費，緩沖區是否空來生產的邏輯。如果用while 和 volatile也可以做，不過本質上會讓線程處于忙等待，占用CPU時間，對性能造成影響。

wait： 將當前線程放入，該對象的等待池中，線程A調用了B對象的wait()方法，線程A進入B對象的等待池，并且釋放B的鎖。（這里，線程A必須持有B的鎖，所以調用的代碼必須在synchronized修飾下，否則直接拋出java.lang.IllegalMonitorStateException異常）。

notify：將該對象中等待池中的線程，隨機選取一個放入對象的鎖池，當當前線程結束后釋放掉鎖， 鎖池中的線程即可競爭對象的鎖來獲得執行機會。

notifyAll：將對象中等待池中的線程，全部放入鎖池。

（notify鎖喚醒的線程選擇由虛擬機實現來決定，不能保證一個對象鎖關聯的等待集合中的線程按照所期望的順序被喚醒，很可能一個線程被喚醒之后，發現他所要求的條件并沒有滿足，而重新進入等待池。因為當等待池中包含多個線程時，一般使用notifyAll方法，不過該方法會導致線程在沒有必要的情況下被喚醒，之后又馬上進入等待池，對性能有影響，不過能保證程序的正確性）

工作流程：

a、Consumer線程A 來 看產品，發現產品為空，調用產品對象的wait()，線程A進入產品對象的等待池并釋放產品的鎖。

b、Producer線程B獲得產品的鎖，執行產品的notifyAll()，Consumer線程A從產品的等待池進入鎖池，Producer線程B生產產品，然后退出釋放鎖。

c、Consumer線程A獲得產品鎖，進入執行，發現有產品，消費產品，然后退出。

例子：

public synchronized String pop(){ 
 this.notifyAll();// 喚醒對象等待池中的所有線程，可能喚醒的就是 生產者（當生產者發現產品滿，就會進入對象的等待池，這里代碼省略，基本略同） 
 while(index == -1){//如果發現沒產品，就釋放鎖，進入對象等待池 
 this.wait(); 
 }//當生產者生產完后，消費者從this.wait()方法再開始執行，第一次還會執行循環，萬一產品還是為空，則再等待，所以這里必須用while循環，不能用if 
 String good = buffer[index]; 
 buffer[index] = null; 
 index--; 
 return good;// 消費完產品，退出。 
} 

注：wait()方法有超時和不超時之分，超時的在經過一段時間，線程還在對象的等待池中，那么線程也會推出等待狀態。

3、線程狀態轉換：

已經廢棄的方法：stop、suspend、resume、destroy，這些方法在實現上時不安全的。

線程的狀態：NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING（有超時的等待）、TERMINATED。

a、方法sleep()進入的阻塞狀態，不會釋放對象的鎖（即大家一起睡，誰也別想執行代碼），所以不要讓sleep方法處在synchronized方法或代碼塊中，否則造成其他等待獲取鎖的線程長時間處于等待。

b、方法join()則是主線程等待子線程完成，再往下執行。例如main方法新建兩個線程A和B

public static void main(String[] args) throws InterruptedException { 
Thread t1 = new Thread(new ThreadTesterA()); 
Thread t2 = new Thread(new ThreadTesterB()); 
t1.start(); 
t1.join(); // 等t1執行完再往下執行 
t2.start(); 
t2.join(); // 在虛擬機執行中，這句可能被忽略 
}

c、方法interrupt()，向被調用的對象線程發起中斷請求。如線程A通過調用線程B的d的interrupt方法來發出中斷請求，線程B來處理這個請求，當然也可以忽略，這不是必須的。Object類的wait()、Thread類的join()和sleep方法都會拋出受檢異常java.lang.InterruptedException，通過interrupt方法中斷該線程會導致線程離開等待狀態。對于wait()調用來說，線程需要重新獲取監視器對象上的鎖之后才能拋出InterruptedException異常，并致以異常的處理邏輯。
可以通過Thread類的isInterrupted方法來判斷是否有中斷請求發生，通常可以利用這個方法來判斷是否退出線程（類似上面的volatitle修飾符的例子）；

Thread類還有個方法Interrupted()，該方法不但可以判斷當前線程是否被中斷，還會清楚線程內部的中斷標記，如果返回true，即曾被請求中斷，同時調用完后，清除中斷標記。

如果一個線程在某個對象的等待池，那么notify和interrupt 都可以使該線程從等待池中被移除。如果同時發生，那么看實際發生順序。如果是notify先，那照常喚醒，沒影響。如果是interrupt先，并且虛擬機選擇讓該線程中斷，那么即使nofity，也會忽略該線程，而喚醒等待池中的另一個線程。

e、yield()，嘗試讓出所占有的CPU資源，讓其他線程獲取運行機會，對操作系統上的調度器來說是一個信號，不一定立即切換線程。（在實際開發中，測試階段頻繁調用yeid方法使線程切換更頻繁，從而讓一些多線程相關的錯誤更容易暴露出來）。

四、非阻塞方式

線程之間同步機制的核心是監視對象上的鎖，競爭鎖來獲得執行代碼的機會。當一個對象獲取對象的鎖，然后其他嘗試獲取鎖的對象會處于等待狀態，這種鎖機制的實現方式很大程度限制了多線程程序的吞吐量和性能（線程阻塞），且會帶來死鎖（線程A有a對象鎖，等著獲取b對象鎖，線程B有b對象鎖，等待獲取a對象鎖）和優先級倒置（優先級低的線程獲得鎖，優先級高的只能等待對方釋放鎖）等問題。

如果能不阻塞線程，又能保證多線程程序的正確性，就能有更好的性能。

在程序中，對共享變量的使用一般遵循一定的模式，即讀取、修改和寫入三步組成。之前碰到的問題是，這三步執行中可能線程執行切換，造成非原子操作。鎖機制是把這三步變成一個原子操作。

目前CPU本身實現 將這三步 合起來 形成一個原子操作，無需線程鎖機制干預，常見的指令是“比較和替換”（compare and swap,CAS），這個指令會先比較某個內存地址的當前值是不是指定的舊指，如果是，就用新值替換，否則什么也不做，指令返回的結果是內存地址的當前值。通過CAS指令可以實現不依賴鎖機制的非阻塞算法。一般做法是把CAS指令的調用放在一個無限循環中，不斷嘗試，知道CAS指令成功完成修改。

java.util.concurrent.atomic包中提供了CAS指令。（不是所有CPU都支持CAS，在某些平臺，java.util.concurrent.atomic的實現仍然是鎖機制）

atomic包中提供的Java類分成三類：

1、支持以原子操作來進行更新的數據類型的Java類（AtomicBoolean、AtomicInteger、AtomicReference），在內存模型相關的語義上，這四個類的對象類似于volatile變量。

類中的常用方法：

a、compareAndSet：接受兩個參數，一個是期望的舊值，一個是替換的新值。

b、weakCompareAndSet：效果同compareAndSet（JSR中表示weak原子方式讀取和有條件地寫入變量但不創建任何 happen-before 排序，但在源代碼中和compareAndSet完全一樣，所以并沒有按JSR實現）

c、get和set：分別用來直接獲取和設置變量的值。

d、lazySet：與set類似，但允許編譯器把lazySet方法的調用與后面的指令進行重排，因此對值得設置操作有可能被推遲。

例：

public class AtomicIdGenerator{ 
 private final AtomicInter counter = new AtomicInteger(0); 
 public int getNext(){ 
 return counter.getAndIncrement(); 
 } 
} 
// getAndIncrement方法的內部實現方式，這也是CAS方法的一般模式，CAS方法不一定成功，所以包裝在一個無限循環中，直到成功 
public final int getAndIncrement(){ 
 for(;;){ 
 int current = get(); 
 int next = current +1; 
 if(compareAndSet(current,next)) 
  return current; 
 } 
} 

2、提供對數組類型的變量進行處理的Java類，AtomicIntegerArray、AtomicLongArray和AtomicReferenceArray類。（同上，只是放在類數組里，調用時也只是多了一個操作元素索引的參數）

3、通過反射的方式對任何對象中包含的volatitle變量使用CAS方法，AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicReferenceFieldUpdater。他們提供了一種方式把CAS的功能擴展到了任何Java類中聲明為volatitle的域上。（靈活，但語義較弱，因為對象的volatitle可能被非atomic的其他方式被修改）

public class TreeNode{ 
 private volatile TreeNode parent; 
// 靜態工廠方法 
 private static final AtomicReferenceFieldUpdater<TreeNode, TreeNode> parentUpdater = AtomicReferenceFieldUpdater.newUpdater(TreeNode.class,TreeNode.class,"parent"); 
public boolean compareAndSetParent(TreeNode expect, TreeNode update){ 
 return parentUpdater.compareAndSet(this, expect, update); 
} 
} 

注：java.util.concurrent.atomic包中的Java類屬于比較底層的實現，一般作為java.util.concurrent包中很多非阻塞的數據結構的實現基礎。

比較多的用AtomicBoolean、AtomicInteger、AtomicLong和AtomicReference。在實現線程安全的計數器時，AtomicInteger和AtomicLong類時最佳的選擇。

五、高級同步機制（比synchronized更靈活的加鎖機制）

synchronized和volatile，以及wait、notify等方法抽象層次低，在程序開發中使用比較繁瑣，易出錯。

而多線程之間的交互來說，存在某些固定的模式，如生產者-消費者和讀者-寫者模式，把這些模式抽象成高層API，使用起來會非常方便。

java.util.concurrent包為多線程提供了高層的API，滿足日常開發中的常見需求。

常用接口

1、Lock接口，表示一個鎖方法：

a、lock()，獲取所，如果無法獲取所鎖，會處于等待狀態

b、unlock()，釋放鎖。（一般放在finally代碼塊中）

c、lockInterruptibly()，與lock()類似，但允許當前線程在等待獲取鎖的過程中被中斷。（所以要處理InterruptedException）

d、tryLock()，以非阻塞方式獲取鎖，如果無法獲取鎖，則返回false。（tryLock()的另一個重載可以指定超時，如果指定超時，當無法獲取鎖，會等待而阻塞，同時線程可以被中斷）

2、ReadWriteLock接口，表示兩個鎖，讀取的共享鎖和寫入的排他鎖。（適合常見的讀者--寫者場景）

ReadWriteLock接口的readLock和writeLock方法來獲取對應的鎖的Lock接口的實現。

在多數線程讀取，少數線程寫入的情況下，可以提高多線程的性能，提高使用該數據結構的吞吐量。

如果是相反的情況，較多的線程寫入，則接口會降低性能。

3、ReentrantLock類和ReentrantReadWriteLock，分別為上面兩個接口的實現類。

他們具有重入性：即允許一個線程多次獲取同一個鎖（他們會記住上次獲取鎖并且未釋放的線程對象，和加鎖的次數，getHoldCount()）

同一個線程每次獲取鎖，加鎖數+1，每次釋放鎖，加鎖數-1，到0，則該鎖被釋放，可以被其他線程獲取。

public class LockIdGenrator{ 
//new ReentrantLock(true)是重載，使用更加公平的加鎖機制，在鎖被釋放后，會優先給等待時間最長的線程，避免一些線程長期無法獲得鎖 
 private int ReentrantLock lock = ReentrantLock(); 
 privafte int value = 0; 
 public int getNext(){ 
 lock.lock(); //進來就加鎖，沒有鎖會等待 
 try{ 
  return value++;//實際操作 
 }finally{ 
  lock.unlock();//釋放鎖 
 } 
 } 
} 

注：重入性減少了鎖在各個線程之間的等待，例如便利一個HashMap，每次next()之前加鎖，之后釋放，可以保證一個線程一口氣完成便利，而不會每次next()之后釋放鎖，然后和其他線程競爭，降低了加鎖的代價， 提供了程序整體的吞吐量。（即，讓一個線程一口氣完成任務，再把鎖傳遞給其他線程）。

4、Condition接口，Lock接口代替了synchronized，Condition接口替代了object的wait、nofity。

a、await()，使當前線程進入等待狀態，知道被喚醒或中斷。重載形式可以指定超時時間。

b、awaitNanos()，以納秒為單位等待。
c、awaitUntil()，指定超時發生的時間點，而不是經過的時間，參數為java.util.Date。

d、awaitUninterruptibly()，前面幾種會響應其他線程發出的中斷請求，他會無視，直到被喚醒。

注：與Object類的wait()相同，await()會釋放其所持有的鎖。

e、signal()和signalAll， 相當于 notify和notifyAll

Lock lock = new ReentrantLock(); 
Condition condition = lock.newCondition(); 
lock.lock(); 
try{ 
 while(/*邏輯條件不滿足*/){ 
 condition.await(); 
 } 
}finally{ 
 lock.unlock(); 
}

六、底層同步器

多線程程序中，線程之間存在多種不同的同步方式。除了Java標準庫提供的同步方式之外，程序中特有的同步方式需要由開發人員自己來實現。

常見的一種需求是 對有限個共享資源的訪問，比如多臺個人電腦，2臺打印機，當多個線程在等待同一個資源時，從公平角度出發，會用FIFO隊列。

  如果程序中的同步方式可以抽象成對有限個資源的訪問，那么可以使用java.util.concurrent.locks包中的AbstractQueuedSynchronizer類和AbstractQueuedLongSynchronizer類作為實現的基礎，前者用int類型的變量來維護內部狀態，而后者用long類型。（可以將這個變量理解為共享資源個數）

通過getState、setState、和compareAndSetState3個方法更新內部變量的值。

AbstractQueuedSynchronizer類是abstract的，需要覆蓋其中包含的部分方法，通常做法是把其作為一個Java類的內部類，外部類提供具體的同步方式，內部類則作為實現的基礎。有兩種模式，排他模式和共享模式，分別對應方法 tryAcquire()、tryRelease 和 tryAcquireShared、tryReleaseShared，在這些方法中，使用getState、setState、compareAndSetState3個方法來修改內部變量的值，以此來反應資源的狀態。

public class SimpleResourceManager{ 
 private final InnerSynchronizer synchronizer; 
 private static class InnerSynchronizer extends AbstractQueuedSynchronizer{ 
 InnerSynchronizer(int numOfResources){ 
  setState(numOfResources); 
 } 
 protected int tryAcquireShared(int acquires){ 
  for(;;){ 
  int available = getState(); 
  int remain = available - acquires; 
  if(remain <0 || comapreAndSetState(available, remain){ 
  return remain; 
  } 
  } 
 } 
 protected boolean try ReleaseShared(int releases){ 
  for(;;){ 
  int available = getState(); 
  int next = available + releases; 
  if(compareAndSetState(available,next){ 
  return true; 
  } 
  } 
 } 
 } 
 public SimpleResourceManager(int numOfResources){ 
 synchronizer = new InnerSynchronizer(numOfResources); 
 } 
 public void acquire() throws InterruptedException{ 
 synchronizer.acquireSharedInterruptibly(1); 
 } 
 pubic void release(){ 
 synchronizer.releaseShared(1); 
 } 
} 

七、高級同步對象（提高開發效率）

atomic和locks包提供的Java類可以滿足基本的互斥和同步訪問的需求，但這些Java類的抽象層次較低，使用比較復雜。
更簡單的做法是使用java.util.concurrent包中的高級同步對象。

1、信號量。

信號量一般用來數量有限的資源，每類資源有一個對象的信號量，信號量的值表示資源的可用數量。

在使用資源時，需要從該信號量上獲取許可，成功獲取許可，資源的可用數-1；完成對資源的使用，釋放許可，資源可用數+1； 當資源數為0時，需要獲取資源的線程以阻塞的方式來等待資源，或過段時間之后再來檢查資源是否可用。（上面的SimpleResourceManager類實際上時信號量的一個簡單實現）

java.util.concurrent.Semaphore類，在創建Semaphore類的對象時指定資源的可用數

a、acquire()，以阻塞方式獲取許可

b、tryAcquire()，以非阻塞方式獲取許可

c、release()，釋放許可。

d、accquireUninterruptibly()，accquire()方法獲取許可以的過程可以被中斷，如果不希望被中斷，使用此方法。

public class PrinterManager{ 
 private final Semphore semaphore; 
 private final List<Printer> printers = new ArrayList<>(): 
 public PrinterManager(Collection<&#63; extends Printer> printers){ 
  this.printers.addAll(printers); 
  //這里重載方法，第二個參數為true，以公平競爭模式，防止線程饑餓 
  this.semaphore = new Semaphore(this.printers.size(),true); 
 } 
 public Printer acquirePrinter() throws InterruptedException{ 
  semaphore.acquire(); 
  return getAvailablePrinter(); 
 } 
 public void releasePrinter(Printer printer){ 
  putBackPrinter(pinter); 
  semaphore.release(); 
 } 
 private synchronized Printer getAvailablePrinter(){ 
  printer result = printers.get(0); 
  printers.remove(0); 
  return result; 
 } 
 private synchronized void putBackPrinter(Printer printer){ 
  printers.add(printer); 
 } 
} 

2、倒數閘門

多線程協作時，一個線程等待另外的線程完成任務才能繼續進行。

java.util.concurrent.CountDownLatch類，創建該類時，指定等待完成的任務數；當一個任務完成，調用countDonw()，任務數-1。等待任務完成的線程通過await()，進入阻塞狀態，直到任務數量為0。CountDownLatch類為一次性，一旦任務數為0，再調用await()不再阻塞當前線程，直接返回。

例：

public class PageSizeSorter{ 
 // 并發性能遠遠優于HashTable的 Map實現，hashTable做任何操作都需要獲得鎖，同一時間只有有個線程能使用，而ConcurrentHashMap是分段加鎖，不同線程訪問不同的數據段，完全不受影響，忘記HashTable吧。 
 private static final ConcurrentHashMap<String , Interger> sizeMap = new ConcurrentHashMap<>(); 
 private static class GetSizeWorker implements Runnable{ 
  private final String urlString; 
  public GetSizeWorker(String urlString , CountDownLatch signal){ 
   this.urlString = urlStirng; 
   this.signal = signal; 
  } 
  public void run(){ 
   try{ 
   InputStream is = new URL(urlString).openStream(); 
   int size = IOUtils.toByteArray(is).length; 
   sizeMap.put(urlString, size); 
   }catch(IOException e){ 
   sizeMap.put(urlString, -1); 
   }finally{ 
   signal.countDown()://完成一個任務 ， 任務數-1 
   } 
  } 
 } 
 private void sort(){ 
  List<Entry<String, Integer> list = new ArrayList<sizeMap.entrySet()); 
  Collections.slort(list, new Comparator<Entry<String,Integer>>(){ 
   public int compare (Entry<String, Integer> o1, Entry<Sting , Integer> o2){ 
   return Integer.compare(o2.getValue(),o1.getValue()); 
  }; 
  System.out.println(Arrays.deepToString(list.toArray())); 
 } 
 public void sortPageSize(Collection<String> urls) throws InterruptedException{ 
  CountDownLatch sortSignal = new CountDownLatch(urls.size()); 
  for(String url: urls){ 
   new Thread(new GetSizeWorker(url, sortSignal)).start(); 
  } 
  sortSignal.await()://主線程在這里等待，任務數歸0，則繼續執行 
  sort(); 
 } 
} 

3、循環屏障

循環屏障在作用上類似倒數閘門，不過他不像倒數閘門是一次性的，可以循環使用。另外，線程之間是互相平等的，彼此都需要等待對方完成，當一個線程完成自己的任務之后，等待其他線程完成。當所有線程都完成任務之后，所有線程才可以繼續運行。

當線程之間需要再次進行互相等待時，可以復用同一個循環屏障。

類java.uti.concurrent.CyclicBarrier用來表示循環屏障，創建時指定使用該對象的線程數目，還可以指定一個Runnable接口的對象作為每次循環后執行的動作。（當最后一個線程完成任務之后，所有線程繼續執行之前，被執行。如果線程之間需要更新一些共享的內部狀態，可以利用這個Runnalbe接口的對象來處理）。

每個線程任務完成之后，通過調用await方法進行等待，當所有線程都調用await方法之后，處于等待狀態的線程都可以繼續執行。在所有線程中，只要有一個在等待中被中斷，超時或是其他錯誤，整個循環屏障會失敗，所有等待中的其他線程拋出java.uti.concurrent.BrokenBarrierException。

例：每個線程負責找一個數字區間的質數，當所有線程完成后，如果質數數目不夠，繼續擴大范圍查找

public class PrimeNumber{ 
 private static final int TOTAL_COUTN = 5000; 
 private static final int RANGE_LENGTH= 200; 
 private static final int WORKER_NUMBER = 5; 
 private static volatitle boolean done = false; 
 private static int rangeCount = 0; 
 private static final List<Long> results = new ArrayList<Long>(): 
 private static final CyclicBarrier barrier = new CyclicBarrier(WORKER_NUMBER, new Runnable(){ 
  public void run(){ 
   if(results.size() >= TOTAL_COUNT){ 
   done = true; 
   } 
  } 
 }); 
 private static class PrimeFinder implements Runnable{ 
  public void run(){ 
   while(!done){// 整個過程在一個 while循環下，await()等待，下次循環開始，會再次判斷 執行條件 
   int range = getNextRange(); 
   long start = rang * RANGE_LENGTH; 
   long end = (range + 1) * RANGE_LENGTH; 
   for(long i = start; i<end;i++){ 
    if(isPrime(i)){ 
     updateResult(i); 
    } 
   } 
   try{ 
    barrier.await(); 
   }catch (InterruptedException | BokenBarrierException e){ 
    done = true; 
   } 
   } 
  } 
 } 
 private synchronized static void updateResult(long value){ 
  results.add(value); 
 } 
 private synchronized static int getNextRange(){ 
  return rangeCount++; 
 } 
 private static boolean isPrime(long number){ 
  //找質數的代碼 
 } 
 public void calculate(){ 
  for(int i=0;i<WORKER_NUMBER;i++){ 
   new Thread(new PrimeFinder()).start(); 
  } 
  while(!done){ 
 
  } 
  //計算完成 
 } 
} 

4、對象交換器

適合于兩個線程需要進行數據交換的場景。（一個線程完成后，把結果交給另一個線程繼續處理）
java.util.concurrent.Exchanger類，提供了這種對象交換能力，兩個線程共享一個Exchanger類的對象，一個線程完成對數據的處理之后，調用Exchanger類的exchange()方法把處理之后的數據作為參數發送給另外一個線程。而exchange方法的返回結果是另外一個線程鎖提供的相同類型的對象。如果另外一個線程未完成對數據的處理，那么exchange()會使當前線程進入等待狀態，直到另外一個線程也調用了exchange方法來進行數據交換。

例：

public class SendAndReceiver{ 
 private final Exchanger<StringBuilder> exchanger = new Exchanger<StringBuilder>(); 
 private class Sender implements Runnable{ 
  public void run(){ 
   try{ 
   StringBuilder content = new StringBuilder("Hello"); 
   content = exchanger.exchange(content); 
   }catch(InterruptedException e){ 
   Thread.currentThread().interrupt(); 
   } 
  } 
 } 
 private class Receiver implements Runnable{ 
  public void run(){ 
   try{ 
   StringBuilder content = new StringBuilder("World"); 
   content = exchanger.exchange(content); 
   }catch(InterruptedException e){ 
   Thread.currentThread().interrupt(); 
   } 
  } 
 } 
 public void exchange(){ 
  new Thread(new Sender()).start(); 
  new Thread(new Receiver()).start(); 
 } 
} 

八、數據結構（多線程程序使用的高性能數據結構）

java.util.concurrent包中提供了一些適合多線程程序使用的高性能數據結構，包括隊列和集合類對象等。

1、隊列

a、BlockingQueue接口：線程安全的阻塞式隊列；當隊列已滿時，想隊列添加會阻塞；當隊列空時，取數據會阻塞。（非常適合消費者-生產者模式）

阻塞方式：put()、take()。

非阻塞方式：offer()、poll()。

實現類：基于數組的固定元素個數的ArrayBolockingQueue和基于鏈表結構的不固定元素個數的LinkedBlockQueue類。

b、BlockingDeque接口： 與BlockingQueue相似，但可以對頭尾進行添加和刪除操作的雙向隊列；方法分為兩類，分別在隊首和對尾進行操作。

實現類：標準庫值提供了一個基于鏈表的實現，LinkedBlockgingDeque。

2、集合類

在多線程程序中，如果共享變量時集合類的對象，則不適合直接使用java.util包中的集合類。這些類要么不是線程安全，要么在多線程下性能比較差。

應該使用java.util.concurrent包中的集合類。

a、ConcurrentMap接口： 繼承自java.util.Map接口

putIfAbsent()：只有在散列表不包含給定鍵時，才會把給定的值放入。

remove()：刪除條目。

replace(key,value)：把value 替換到給定的key上。

replace(key, oldvalue, newvalue)：CAS的實現。

實現類：ConcurrentHashMap：

創建時，如果可以預估可能包含的條目個數，可以優化性能。（因為動態調整所能包含的數目操作比較耗時，這個

HashMap也一樣，只是多線程下更耗時）。創建時，預估進行更新操作的線程數，這樣實現中會根據這個數把內部空間劃分為對應數量的部分。（默認是16，如果只有一個線程進行寫操作，其他都是讀取，那么把值設為1 可以提高性能）。

注：當從集合中創建出迭代器遍歷Map元素時，不一定能看到正在添加的數據，只能和集合保證弱一致性。（當然使用迭代器不會因為查看正在改變的Map，而拋出java.util.ConcurrentModifycationException）

b、CopyOnWriteArrayList接口：繼承自java.util.List接口。

顧名思義，在CopyOnWriteArrayList的實現類，所有對列表的更新操作都會新創建一個底層數組的副本，并使用副本來存儲數據；對列表更新操作加鎖，讀取操作不加鎖。

適合多讀取少修改的場景，如果更新操作多，那么不適合用，同樣迭代器只能表示創建時列表的狀態，更新后使用了新的底層數組，迭代器還是引用舊的底層數組。

九、多線程任務的執行

過去線程的執行，是先創建Thread類的想，再調用start方法啟動，這種做法要求開發人員對線程進行維護，在線程較多時，一般創建一個線程池同一管理，同時降低重復創建線程的開銷

在J2SE5.0中，java.util.concurrent包提供了豐富的用來管理線程和執行任務的實現。

1、基本接口（描述任務）

a、Callable接口：

Runnable接口受限于run方法的類型簽名，而Callable只有一個方法call()，可以有返回值，可以拋出受檢異常。

b、Future接口：

過去，需要異步線程的任務執行結果，要求主線程和任務執行線程之間進行同步和數據傳遞。

Future簡化了任務的異步執行，作為異步操作的一個抽象。調用get()方法可以獲取異步的執行結果，如果任務沒有執行完，會等待，直到任務完成或被取消，cancel()可以取消。

c、Delayed接口：

延遲執行任務，getDelay()返回當前剩余的延遲時間，如果不大于0，說明延遲時間已經過去，應該調度并執行該任務。

2、組合接口（描述任務）

a、RunnableFuture接口：繼承自Runnable接口和Future接口。

當來自Runnalbe接口中的run方法成功執行之后，相當于Future接口表示的異步任務已經完成，可以通過get()獲取運行結果。

b、ScheduledFuture接口：繼承Future接口和Delayed接口，表示一個可以調用的異步操作。

c、RunnableScheduledFuture接口：繼承自Runnable、Delayed和Future，接口中包含isPeriodic，表明該異步操作是否可以被重復執行。

3、Executor接口、ExcutorServer接口、ScheduleExecutorService接口和CompletionService接口（描述任務執行）

a、executor接口，execute()用來執行一個Runnable接口的實現對象，不同的Executor實現采取不同執行策略，但提供的任務執行功能比較弱。

b、excutorServer接口，繼承自executor；

提供了對任務的管理：submit()，可以吧Callable和Runnable作為任務提交，得到一個Future作為返回，可以獲取任務結果或取消任務。

提供批量執行：invokeAll()和invokeAny()，同時提交多個Callable；invokeAll()，會等待所有任務都執行完成，返回一個包含每個任務對應Future的列表；invokeAny()，任何一個任務成功完成，即返回該任務結果。

提供任務關閉：shutdown()、shutdownNow()來關閉服務，前者不允許新的任務提交，后者試圖終止正在運行和等待的任務，并返回已經提交單沒有被運行的任務列表。（兩個方法都不會等待服務真正關閉，只是發出關閉請求。）。

shutdownDow，通常做法是向線程發出中斷請求，所以確保提交的任務實現了正確的中斷處理邏輯。

c、ScheduleExecutorService接口，繼承自excutorServer接口：支持任務的延遲執行和定期執行，可以執行Callable或Runnable。

schedule()，調度一個任務在延遲若干時間之后執行；
scheduleAtFixedRate()：在初始延遲后，每隔一段時間循環執行；在下一次執行開始時，上一次執行可能還未結束。（同一時間，可能有多個）

scheduleWithFixedDelay：同上，只是在上一次任務執行完后，經過給定的間隔時間再開始下一次執行。（同一時間，只有一個）

以上三個方法都返回ScheduledFuture接口的實現對象。

d、CompletionService接口，共享任務執行結果。

通常在使用ExecutorService接口，通過submit提交任務，并得到一個Future接口來獲取任務結果，如果任務提交者和執行結果的使用者是程序的不同部分，那就要把Future在不同部分進行傳遞；而CompletionService就是解決這個問題，程序不同部分可以共享CompletionService，任務提交后，執行結果可以通過take（阻塞），poll（非阻塞）來獲取。

標準庫提供的實現是 ExecutorCompletionService，在創建時，需要提供一個Executor接口的實現作為參數，用來實際執行任務。

例：多線程方式下載文件

public class FileDownloader{ 
 // 線程池 
 private final ExecutorService executor = Executors.newFixedThreadPool(10); 
 public boolean download(final URL url, final Path path){ 
 Future<Path> future = executor.submit(new Callable<Path>(){ //submit提交任務 
  public Path call(){ 
   //這里就省略IOException的處理了 
   InputStream is = url.openStream(); 
   Files.copy(is, path, StandardCopyOption.REPLACE_EXISTING); 
   return path; 
  }); 
  try{ 
   return future.get() !=null &#63; true : false; 
  }<span >catch(InterruptedException | ExecutionException e){</span> 
    return false; 
  } 
 } 
 public void close(){//當不再使用FileDownloader類的對象時，應該使用close方法關閉其中包含的ExecutorService接口的實現對象，否則虛擬機不會退出，占用內存不釋放 
  executor.shutdown();// 發出關閉請求，此時不會再接受新任務 
  try{ 
   if(!executor.awaitTermination(3, TimeUnit.MINUTES)){// awaitTermination 來等待一段時間，使正在執行的任務或等待的任務有機會完成 
   executor.shutdownNow();// 如果等待時間過后還有任務沒完成，則強制結束 
   executor.awaitTermination(1, TimeUnit.MINUTES);// 再等待一段時間，使被強制結束的任務完成必要的清理工作 
   } 
  }catch(InterruptedException e){ 
   executor.shutdownNow(); 
   Thread.currentThread().interrupt(); 
  } 
 } 
} 

十、Java SE 7 新特性

對java.util.concurrent包進行更新，增加了新的輕量級任務執行框架fork/join和多階段線程同步工具。

1、輕量級任務執行框架fork/join

這個框架的目的主要是更好地利用底層平臺上的多核和多處理器來進行并行處理。

通過分治算法或map/reduce算法來解決問題。

fork/join 類比于 map/reduce。

fork操作是把一個大的問題劃分為若干個較小的問題，劃分過程一般為遞歸，直到可以直接進行計算的粒度適合的子問題；子問題在結算后，可以得到整個問題的部分解

join操作收集子結果，合并，得到完整解，也可能是 遞歸進行的。

相對一般的線程池實現，F/J框架的優勢在任務的處理方式上。在一般線程池中，一個線程由于某些原因無法運行，會等待；而在F/J，某個子問題由于等待另外一個子問題的完成而無法繼續運行，那么處理該子問題的線程會主動尋找其他尚未運行的子問題來執行。這種方式減少了等待時間，提高了性能。

為了F/J能高效，在每個子問題視線中應避免使用synchronized或其他方式進行同步，也不應使用阻塞式IO或過多訪問共享變量。在理想情況下，每個子問題都應值進行CPU計算，只使用每個問題的內部對象，唯一的同步應只發生在子問題和創建它的父問題之間。（這完全就是Hadoop的MapReduce嘛）

a、ForkJoinTask類：表示一個由F/J框架執行的任務，該類實現了Future接口，可以按照Future接口的方式來使用。（表示任務）

fork()，異步方式啟動任務的執行。

join()，等待任務完成并返回執行結果。

在創建自己的任務時，最好不要直接繼承自ForkJoinTask，而是繼承其子類，RecuriveTask或RecursiveAction，前者可以返回結果，后者不行。

b、ForkJoinPool類：表示任務執行，實現了ExecutorService接口，除了可以執行ForkJoinTask，也可以執行Callable和Runnable。（任務執行）

執行任務的兩大類：

第一類：execute、invoke或submit方法：直接提交任務。

第二類：fork()：運行ForkJoinTask在執行過程中的子任務。

一般作法是表示整個問題的ForkJoinTask用第一類提交，執行過程中產生的子任務不需要處理，ForkJoinPool會負責子任務執行。

例：查找數組中的最大值

private static class MaxValueTask extends RecursiveTask<Long>{ 
 private final long[] array; 
 private final int start; 
 private final int end; 
 MaxValueTask(long[] array, int start, int end){ 
  this.array = array; 
  this.start = start; 
  this.end = end; 
 } 
 //compute是RecursiveTask的主方法 
 protected long compute(){ 
  long max = Long.MIN_VALUE; 
  if(end - start < RANG_LENGTH){//尋找最大值 
   for(int i = start; i<end;i++{ 
   if(array[i] > max){ 
    max = array[i]; 
   } 
   } 
  }else{// 二分任務 
   int mid = (start + end) /2; 
   MaxValueTask lowTask = new MaxValueTask(array, start , mid); 
   MaxValueTask highTask = new MaxValueTask(array, mid, end); 
   lowTask.fork();// 異步啟動任務 
   highTask.fork(); 
   max = Math.max(max, lowTask.join());//等待執行結果 
   max = Math.max(max, highTask.join(); 
  } 
  return max; 
 } 
 public Long calculate(long[] array){ 
  MaxValueTask task = new MaxValueTask(array, 0 , array.length); 
  Long result = forkJoinPool.invoke(task); 
  return result; 
 } 
} 

注：這個例子是示例，但從性能上說直接對整個數組順序比較效率高，畢竟多線程所帶來的額外開銷過大。
在實際中，F/J框架發揮作用的場合很多，比如在一個目錄包含的所有文本中搜索某個關鍵字，可以每個文件創建一個子任務。

如果相關的功能可以用遞歸和分治來解決，就適合F/J。

2、多階段線程同步工具

Phaser類是Java SE 7中新增的一個使用同步工具，功能和靈活性比倒數閘門和循環屏障要強很多。

在F/J框架中的子任務之間要進行同步時，應優先考慮Phaser。

Phaser把多個線程寫作執行的任務劃分成多個階段（phase），編程時要明確各個階段的任務，每個階段都可以有任意個參與者，線程可以隨時注冊并參與到某個階段，當一個階段中所有線程都成功完成之后，Phaser的onAdvance()被調用，可以通過覆蓋添加自定義處理邏輯（類似循環屏障的使用的Runnable接口），然后Phaser類會自動進入下個階段。如此循環，知道Phaser不再包含任何參與者。

Phaser創建后，初始階段編號為0，構造函數中指定初始參與個數。

register()，bulkRegister()，動態添加一個或多個參與者。

arrive()，某個參與者完成任務后調用

arriveAndDeregister()，任務完成，取消自己的注冊。

arriveAndAwaitAdvance()，自己完成等待其他參與者完成。，進入阻塞，直到Phaser成功進入下個階段。

awaitAdvance()、awaitAdvanceInterruptibly()，等待phaser進入下個階段，參數為當前階段的編號，后者可以設置超時和處理中斷請求。

另外，Phaser的一個重要特征是多個Phaser可以組成樹形結構，Phaser提供了構造方法來指定當前對象的父對象；當一個子對象參與者>0，會自動注冊到父對象中；當=0，自動解除注冊。

例：從指定網址，下載img標簽的照片

階段

1、處理網址對應的html文本，和抽取img的鏈接；

2、創建圖片下載子線程，主線程等待；

3、子線程下載圖片，主線程等待；

4、任務完成退出

public class WebPageImageDownloader{ 
 private final Phaser phaser = new Phaser(1);//初始參與數1，代表主線程。 
 public void download(URL url, final Path path) throws IOException{ 
  String content = getContent(url);//獲得HTML文本，省略。 
  List<URL> imageUrls = extractImageUrls(content);//獲得圖片鏈接，省略。 
  for(final URL imageUrl : imageUrls){ 
   phaser.register();//子線程注冊 
   new Thread(){ 
   public void run(){ 
    phaser.arriveAndAwaitAdvance();//第二階段的等待，等待進入第三階段 
    try{ 
     InputStream is = imageUrl.openStream(); 
     File.copy(is, getSavePath(path, imageUrl), StandardCopyOption.REPLACE_EXISTING); 
    }catch(IOException e){ 
     e.printStackTrace(): 
    }finally{ 
     phaser.arriveAndDeregister();//子線程完成任務，退出。 
    } 
   } 
  }.start(); 
  } 
  phaser.arriveAndAwaitAdvance();//第二階段等待，子線程在注冊 
  phaser.arriveAndAwaitAdvance();//第三階段等待，子線程在下載 
  phaser.arriveAndDeregister();//所有線程退出。 
 } 
} 

十一、ThreadLocal類

java.lang.ThreadLocal，線程局部變量，把一個共享變量變為一個線程的私有對象。不同線程訪問一個ThreadLocal類的對象時，鎖訪問和修改的事每個線程變量各自獨立的對象。通過ThreadLocal可以快速把一個非線程安全的對象轉換成線程安全的對象。（同時也就不能達到數據傳遞的作用了）。

a、get()和set()分別用來獲取和設置當前線程中包含的對象的值。

b、remove()，刪除。

c、initialValue()，初始化值。如果沒有通過set方法設置值，第一個調用get，會通過initValue來獲取對象的初始值。

ThreadLoacl的一般用法，創建一個ThreadLocal的匿名子類并覆蓋initalValue()，把ThreadLoacl的使用封裝在另一個類中

public class ThreadLocalIdGenerator{ 
 private static final ThreadLocal<IdGenerator> idGenerator = new ThreadLocal<IdGenerator>(){ 
   protected IdGenerator initalValue(){ 
   return new IdGenerator();//IdGenerator 是個初始int value =0,然后getNext(){ return value++} 
   } 
  }; 
 public static int getNext(){ 
  return idGenerator.get().getNext(); 
 } 
} 

ThreadLoal的另外一個作用是創建線程唯一的對象，在有些情況，一個對象在代碼中各個部分都需要用到，傳統做法是把這個對象作為參數在代碼間傳遞，如果使用這個對I昂的代碼都在同一個線程，可以封裝在ThreadLocal中。

如：在多線程中，生成隨機數

java.util.Random會帶來競爭問題，java.util.concurrent.ThreadLocalRandom類提供多線程下的隨機數聲場，底層是ThreadLoacl。

上述內容就是多線程并發編程如何在Java項目中實現 ，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。