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