為什么80%的碼農都做不了架構師？>>> ??

同步工具類可以是任何一個對象，只要它根據其自身的狀態來協調線程控制流。阻塞隊列（BlockingQueue）可以作為同步工具類，其他類型的同步工具類還包括信號量（Semaphore），柵欄（Barrier）以及閉鎖（Latch）。在平臺類庫中還包含其他一些同步工具類的類，如果這些類還無法滿足需要，那么可以創建自己的同步工具類。

閉鎖Latch

閉鎖可以延遲線程的進度直到其到達終止狀態。閉鎖的作用相當于一扇門：在閉鎖到達結束狀態之前，這扇門一直是關著的，并且沒有任何線程能通過，當到達結束狀態時，這扇門會打開并允許所有的線程通過。當閉鎖到達結束狀態后，將不會再改變狀態，因此這扇門將永遠保持打開狀態。閉鎖可以用來確保某些活動直到其他活動都完成后才繼續執行，例如：

• 確保某個計算在其需要的所有資源都被初始化之后才繼續執行。二元閉鎖（包括兩個狀態）可以用來表示“資源R已經被初始化”，而所有需要R的操作都必須在這個閉鎖上等待。
• 確保某個服務在其依賴的所有其他服務都已經啟動之后才啟動。每個服務都有一個相關的二元閉鎖。當啟動服務S時，將首先在S以來的其他服務的閉鎖上等待，在所有依賴的服務都啟動后會釋放閉鎖S，這樣其他依賴S的服務才能繼續執行。
• 等待直到某個操作的所有參與者（例如：在多玩家游戲中的所有玩家）都就緒再繼續執行。在這種情況中，當所有的玩家都準備就緒時，閉鎖將到達結束狀態。

CountDownLatch是一種靈活的閉鎖實現，可以在上述各種情況下使用，它可以使一個或多個線程等待一組事件發生。閉鎖狀態包括一個計數器，該計數器被初始化為一個正數，表示需要等待的事件數量。countDown方法將遞減計數器，表示有一個事件已經發生了，而await方法等待計數器到達零，這表示所有需要的事件都已經發生。如果計數器的值為非零，那么await會一直阻塞直到計數器為零，或者等待中的線程中斷，或者等待超時。

在下面的TestHarness中給出了閉鎖的兩種常見用法。TestHarness創建一定數量的線程，利用它們并發的執行指定的任務。它使用兩個閉鎖，分別表示“起始門（Start?Gate）”和“結束門（End Gate）”。起始門計數器的初始值為1，而結束們計數器的初始值為工作線程數量。每個工作線程首先要做的就是在啟動門上等待，從而確保所有線程都就緒后才開始執行。而每個線程要做的最后一件事情是調用countDown方法使事件數量減1，這能使主線程高效的等待直到所有工作線程都執行完成，因此可以統計所消耗的時間：

public class TestHarness {public long timeTasks(int nThread, final Runnable task) throws InterruptedException {final CountDownLatch startGate = new CountDownLatch(1);final CountDownLatch endGate = new CountDownLatch(nThread);for (int i = 0; i < nThread; i++) {Thread t = new Thread() {public void run() {try {startGate.await();try {task.run();} finally {endGate.countDown();}} catch(InterruptedException ignored) {}}};t.start();}long start = System.nanoTime();startGate.countDown();long end = System.nanoTime();return end - start;} }

為什么要在TestHarness中使用閉鎖，而不是在線程創建后就立即啟動？或許，我們希望測試n個線程并發執行某個任務時需要的時間。如果在創建線程之后立即啟動它們，那么先啟動的線程將“領先”后啟動的線程，并且活躍線程數量會隨著時間的推移而增加或減少，競爭程度也在不斷發生變化。起始門將使得主線程能夠同時釋放所有的工作線程，而結束門則使主線程能夠等待最后一個線程執行完成，而不是順序的等待每個線程執行完成。

FutureTask

FutureTask也可以用作閉鎖。（FutureTask實現了Future語義，表示一種抽象的可生成結果的計算）。FutureTask表示的計算是通過Callable來實現的，相當于一種可生成結果的Runnable，并且可以處于以下3種狀態：等待運行（Waiting to run），正在運行（Running）和完成運行（Completed）。“執行完成”表示計算的所有可能結束方式，包括正常結束、由于取消而結束和由于異常而結束等。當FutureTask進入完成狀態以后，它會永遠停止在這個狀態上。

Future.get的行為取決于任務的狀態。如果任務已經執行完成，那么get會立即返回結果，否則get將阻塞直到任務進入完成狀態，然后返回結果或者拋出異常。FutureTask將計算結果從執行計算的線程傳遞到獲取這個結果的線程，而FutureTask的規范確保了這種傳遞過程能實現結果的安全發布。

FutureTask在Executor框架中表示異步任務，此外還可以用來表示一些時間較長的計算，這些計算可以在使用計算結果之前啟動。下面的程序就使用了FutureTask來執行一個高開銷的計算，并且計算結果將在稍后使用。通過提前啟動計算，可以減少在等待結果時需要的時間：

public class Preloader{private final FutureTask<ProductInfo> future = new FutureTask<ProductInfo>(new Callable<ProductInfo>(){public ProductInfo call() throws DataLoadException{return loadProductInfo();}});private final Thread thread = new Thread(future);public void start() {thread.start();}public ProductInfo get() throws DataLoadException, InterruptedException{try {return future.get();} catch(ExecutionException e) {Throwable cause = e.getCause();if (cause instance of DataLoadException) {throw (DataLoadException)cause;} else {throw launderThrowable(cause);}}} }

Preloader創建了一個FutureTask，其中包含從數據庫加載產品信息的任務，以及一個執行運算的線程。由于在構造函數或靜態初始化方法中啟動線程并不是一種好方法，因此提供了一個start方法來啟動線程。當程序雖有需要ProductInfo時，可以調用get方法，如果數據已經加載，那么將返回這些數據，否則將等待加載完成后再返回。

Callable表示的任務可以拋出受檢查的或未受檢查的異常，并且任何代碼都可能拋出一個Error。無論任務代碼拋出什么異常，都會被封裝到一個ExecutionException中，并在Future.get中被重新拋出。這將使調用get的代碼變得復雜，因為它不僅需要處理可能出現的ExecutionException（以及未檢查的CancellationException），而且還由于ExecutionException是作為一個Throwable類返回的，因此處理起來并不容易。

信號量Semaphore

計數信號量（Counting Semaphore）用來控制同時訪問某個特定資源的操作數量，或者同時執行某個指定操作的數量。計數信號量還可以用來實現某種資源池，或者對容器施加邊界。

Semaphore中管理著一組虛擬的許可（permit），許可的初始數量可通過構造函數來指定。在執行操作時可以首先獲得許可（只要還有剩余的許可），并在使用以后釋放即可。如果沒有許可，那么aquire將阻塞直到有許可（或者直到被中斷或者操作超時）。release方法將返回一個許可給信號量。計算信號量的一種簡化形式是二值信號量，即初始值為1的Semaphore。二值信號量可以用作互斥體（mutex），并具備不可重入的加鎖語義：誰擁有這個唯一的許可，誰就擁有了互斥鎖。

Semaphore可以用于實現資源池，例如數據庫連接池。我們可以構造一個固定長度的資源池，當池為空時，請求資源將會失敗，但你真正希望看到的行為是阻塞而不是失敗，并且當池非空時解除阻塞。如果將Semaphore的計數值初始化為池的大小，并在從池中獲取一個資源之前先調用aquire方法獲得一個許可，在將資源返回給池之后調用release釋放許可，這樣就實現了固定長度了。

同樣，你可以使用Semaphore將任何一種容器變成有界阻塞容器，如下代碼所示。信號量的計數值會初始化為容器容量的最大值。add操作在向底層容器中添加一個元素之前，首先需要獲得一個許可。如果add操作沒有添加任何元素，那么會立刻釋放許可。同樣remove操作釋放一個許可，使更多的元素能夠添加到容器中。底層的Set實現并不知道關于邊界的任何信息，這是由BoundedHashSet來處理的。

public class BoundedHashSet<T> {private final Set<T> set;private final Semaphore sem;public BoundedHashSet(int bound) {this.set = Collections.synchronizedSet(new HashSet<T>());sem = new Semaphore(bound);}public boolean add(T o) throws InterruptedException {sem.aquire();boolean wasAdded = false;try {wasAdded = set.add(o);return wasAdded;} finally {if (!wasAdded) {sem.release();}}}public boolean remove(Object o) {boolean wasRemoved = set.remove(o);if (wasRemoved) {sem.release();}return wasRemoved;} }

柵欄Barrier

我們已經看到通過閉鎖來啟動一組相關的操作，或者等待也組相關的操作結束。閉鎖是一次性對象，一旦進入終止狀態，就不能被重置。

柵欄（Barrier）類似于閉鎖，它能阻塞線程直到某個事件發生。柵欄與閉鎖的關鍵區別在于：所有線程必須同時到達柵欄位置，才能繼續執行。閉鎖也用于等待事件，而柵欄也用于等待其他線程。柵欄用于實現一些協議，例如幾個家庭決定在某個地方集合：“所有人6:00在麥當勞碰頭，到了以后要等其他人，之后再討論下一步要做的事情。”

CyclicBarrier可以使一定數量的參與方反復地在柵欄位置匯集，它在并行迭代算法中非常有用：這種算法通常將也一個問題拆分成一系列相互獨立的子問題。當線程到達柵欄位置時，將調用await方法，這個方法將阻塞直到所有線程都到達阻塞位置。如果所有線程都到達了柵欄的位置，那么柵欄將打開，此時所有線程都將被釋放，而柵欄將被重置以便下次使用。如果對await的調用超時，或者await阻塞的線程被中斷，那么柵欄就被認為是打破了，所有阻塞的await調用都將終止并拋出BrokenBarrierException。如果成功地通過柵欄，那么await將為每個線程返回一個唯一的到達索引號，我們可以利用這些索引來“選舉”產生一個領導線程，并在下一次迭代中由該領導線程執行一些特殊的工作。CyclicBarrier還可以使你將一個柵欄參數傳遞給構造函數，這是一個Runnable，當成功通過柵欄時會（在下一個子任務線程中）執行它，但在阻塞線程被釋放之前是不能執行的。

下面的例子中給出了如何通過柵欄來計算細胞的自動化模擬。在把模擬過程并行化時，為每個元素（這里為每個細胞）分配一個獨立的線程是不現實的，也因為這將產生過多的線程，而在協調這些線程上導致的開銷將降低計算性能。合理的做法是，將問題分解成一定數量的子問題，為每個子問題分配一個線程來進行求解，之后再將所有的結果合并起來。CellularAutomata將問題分解為Ncpu（可用的CPU數量）個子問題，并將每個子問題分配給一個線程。在每個步驟中，工作線程都為各自子問題中的所有細胞計算新值。當所有工作線程都到達柵欄時，柵欄會把這些新值交給數據模型。在柵欄的操作執行完成以后，工作線程將開始下一步的計算，包括調用isDone方法來判斷是否需要進行下一次迭代。

public class CellularAutomata {private final Board mainBoard;private final CyclicBarrier barrier;private final Worker[] workers;public CellularAutomata (Board board) {this.mainBoard = board;int count = Runtime.getRuntime().availableProcessors();this.barrier = new CyclicBarrier(count, new Runnable(){public void run() {mainBoard.commitNewValues();}});this.workers = new Worker[count];for (int i = 0; i < count; i++) {workers[i] = new Worker(mainBoard.getSubBoard(count, i));}}private class Worker implements Runnable {private final Board board;public Worker(Board board) {this.board = board;}public void run() {while(!board.hasConverged()) {for (int x = 0; x < board.getMaxX(); x++) {for (int y = 0; y < board.getMaxY(); y++) {board.setNewValue(x, y, computeValue(x, y));}}try {barrier.await();} catch(InterruptedException ex) {return;} catch(BrokenBarrierException ex) {return;}}}}public void start() {for (int i = 0; i < workers.length; i++) {new Thread(workers[i]).start();}mainBoard.waitForConvergence();} }

另一種形式的柵欄就是Exchanger，它是一種兩方（Two-Party）柵欄，各方在柵欄位置上交換數據。當兩方執行不對稱的操作時，Exchanger會非常有用，例如當一個線程向緩沖區寫入數據，而另一個線程從緩沖區中讀取數據。這些線程可以使用Exchanger來匯合，并將滿的緩沖區與空的緩沖區交換。當兩個線程通過Exchanger交換對象時，這種交換就把這兩個對象安全地發布給另一方。

數據交換的時機取決于應用程序的響應需求。最簡單的方案是，當緩沖區被填滿時，由填充任務進行交換，當緩沖區為空時，由清空任務進行交換。這樣會把需要交換的次數降至最低，但如果新數據的到達率不可預測，那么一些數據的處理過程就將也延遲。另一個方法是，不僅當緩沖被填滿時進行交換，并且當緩沖被填充到一定程度并保持也一定時間也以后，也進行交換。

轉載于:https://my.oschina.net/itblog/blog/775918

總結

以上是生活随笔為你收集整理的Java并发中常用同步工具类的全部內容，希望文章能夠幫你解決所遇到的問題。

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