Java同步組件概況

• CountDownLatch : 是閉鎖,通過一個計數來保證線程是否一直阻塞
• Semaphore: 控制同一時間,并發線程數量
• CyclicBarrier:字面意思是回環柵欄,通過它可以實現讓一組線程等待至某個狀態之后再全部同時執行。
• ReentrantLock:是一個重入鎖,一個線程獲得了鎖之后仍然可以反復加鎖,不會出現自己阻塞自己的情況。
• Condition:配合ReentrantLock,實現等待/通知模型
• FutureTask:FutureTask實現了接口Future，同Future一樣，代表異步計算的結果。

CountDownLatch 同步輔助類

CountDownLatch類位于java.util.concurrent包,利用它可以實現類似計數器的功能,比如有一個任務A,它要等待其它4個任務執行完畢后才能執行,此時就可以使用CountDownLatch來實現這種功能。

假設計數器的值是3,線程A調用await()方法后,當前線程就進入了等待狀態,之后其它線程中執行CountDownLatch,計數器就會減1,當計數器從3變成0,線程A繼續執行,CountDownLatch這個類可以阻塞當前線程,保證線程在某種條件下,繼續執行。

構造器中的計數值(count)實際上就是閉鎖需要等待的線程數量,這個值只能被設置一次,而且CountDownLatch沒有提供任何機會修改這個計數值。

CountDownLatch代碼案例

package com.rumenz.task;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class CountDownLatchTest {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();CountDownLatch countDownLatch=new CountDownLatch(2);executorService.execute(()->{try{Thread.sleep(3000);System.out.println("任務一完成");}catch (Exception e){e.printStackTrace();}countDownLatch.countDown();});executorService.execute(()->{try{Thread.sleep(5000);System.out.println("任務二完成");}catch (Exception e){e.printStackTrace();}countDownLatch.countDown();});countDownLatch.await();//所有子任務執行完后才會執行System.out.println("主線程開始工作.....");executorService.shutdown();}
}任務一完成
任務二完成
主線程開始工作.....

CountDownlatch指定時間完成任務,如果在規定時間內完成,則等待之前的等待線程(countDownLatch.await())繼續執行

countDownLatch.await(int timeout,TimeUnit timeUnit);設置，第一個參數沒超時時間，第二個參數為時間單位。

package com.rumenz.task;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;public class CountDownLatchTest {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();CountDownLatch countDownLatch=new CountDownLatch(2);executorService.execute(()->{try{Thread.sleep(3000);System.out.println("任務一完成");}catch (Exception e){e.printStackTrace();}countDownLatch.countDown();});executorService.execute(()->{try{Thread.sleep(5000);System.out.println("任務二完成");}catch (Exception e){e.printStackTrace();}countDownLatch.countDown();});//這里只等3秒countDownLatch.await(3, TimeUnit.SECONDS);//所有子任務執行完后才會執行System.out.println("主線程開始工作.....");executorService.shutdown();}
}
//任務一完成
//主線程開始工作.....
//任務二完成

Semaphore控制線程數量

Semaphore經常用于限制獲取某種資源的線程數量,其內部是基于AQS的共享模式,AQS的狀態可以表示許可證的數量,許可證數量不夠線程被掛起;而一旦有一個線程釋放資源,那么可喚醒等待隊列中的線程繼續執行。

Semaphore翻譯過來就是信號量,Semaphore可以阻塞進程并控制同時訪問的線程數,通過acquire()獲取一個許可,如果沒有就等待,而release()釋放一個許可,Semaphore有點類似鎖。

CountDownLatch和Semaphore在使用時,通過和線程池配合使用。
Semaphore適合控制并發,CountDownLatch比較適合保證線程執行完后再執行其它處理,因此模擬并發兩者結合最好。

Semaohore應用場景

Semaphore適合做流量控制,特別是共享的有限資源,比如數據庫連接,假如有一個需求，要讀取幾萬個文件的數據，因為都是IO密集型任務，我們可以啟動幾十個線程并發的讀取，但是如果讀到內存后，還需要存儲到數據庫中，而數據庫的連接數只有10個，這時我們必須控制只有十個線程同時獲取數據庫連接保存數據，否則會報錯無法獲取數據庫連接。這個時候，我們就可以使用Semaphore來做流控。

package com.rumenz.task;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;public class SemaphoreExample1 {private static Integer clientTotal=30;private static Integer threadTotal=3;public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();Semaphore semaphore=new Semaphore(threadTotal);for (int i = 0; i < clientTotal; i++) {final Integer j=i;executorService.execute(()->{try{semaphore.acquire(); // 獲取一個許可update(j);semaphore.release(); // 釋放一個許可}catch (Exception e) {e.printStackTrace();}});}executorService.shutdown();}private static void update(Integer j) throws Exception {System.out.println(j);Thread.sleep(2000);}
}

每2秒打印3個數字。

package com.rumenz.task;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;public class SemaphoreExample1 {private static Integer clientTotal=30;private static Integer threadTotal=3;public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();Semaphore semaphore=new Semaphore(threadTotal);for (int i = 0; i < clientTotal; i++) {final Integer j=i;executorService.execute(()->{try{semaphore.acquire(3); // 獲取多個許可update(j);semaphore.release(3); // 釋放多個許可}catch (Exception e) {e.printStackTrace();}});}executorService.shutdown();}private static void update(Integer j) throws Exception {System.out.println(j);Thread.sleep(2000);}
}

每2秒打印一個數字。

tryAcquire

嘗試獲取許可，如果獲取不成功，則放棄操作，tryAcquire方法提供幾個重載

• tryAcquire() : boolean
• tryAcquire(int permits) : boolean 嘗試獲取指定數量的許可
• tryAcquire(int permits,long timeout,TimeUnit timeUnit) : boolean
• tryAcquire(long timeout,TimeUnit timeUnit) : boolean 嘗試獲取許可的時候可以等待一段時間，在指定時間內未獲取到許可則放棄

Semaphore源碼分析

Semaphore有兩種模式，公平模式和非公平模式。公平模式就是調用acquire的順序就是獲取許可證的順序，遵循FIFO；而非公平模式是搶占式的，也就是有可能一個新的獲取線程恰好在一個許可證釋放時得到了這個許可證，而前面還有等待的線程。

// 非公平模式
public Semaphore(int permits) {sync = new NonfairSync(permits);
}
// fair=true為公平模式，false=非公平模式
public Semaphore(int permits, boolean fair) {sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

---

public class Semaphore implements java.io.Serializable {/** 只指定許可量，構造不公平模式*/public Semaphore(int permits) {sync = new NonfairSync(permits);}/** 指定許可量，并指定模式*/public Semaphore(int permits, boolean fair) {sync = fair ? new FairSync(permits) : new NonfairSync(permits);}//Semaphore內部基于AQS的共享模式，所以實現都委托給了Sync類。 abstract static class Sync extends AbstractQueuedSynchronizer {}/*** NonFair version*/static final class NonfairSync extends Sync {private static final long serialVersionUID = -2694183684443567898L;NonfairSync(int permits) {// 可以看到調用了setState方法，也就是說AQS中的資源就是許可證的數量。super(permits);}protected int tryAcquireShared(int acquires) {return nonfairTryAcquireShared(acquires);}}/*** Fair version*/static final class FairSync extends Sync {private static final long serialVersionUID = 2014338818796000944L;FairSync(int permits) {// 可以看到調用了setState方法，也就是說AQS中的資源就是許可證的數量。super(permits);}protected int tryAcquireShared(int acquires) {for (;;) {if (hasQueuedPredecessors())return -1;int available = getState();int remaining = available - acquires;if (remaining < 0 ||compareAndSetState(available, remaining))return remaining;}}}}

關注微信公眾號:【入門小站】,解鎖更多知識點

總結

以上是生活随笔為你收集整理的Java同步组件之CountDownLatch,Semaphore的全部內容，希望文章能夠幫你解決所遇到的問題。

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