高併發之——淺談AQS中的CountDownLatch、Semaphore與CyclicBarrie
CountDownLatch
概述
同步輔助類，經過它能夠阻塞當前線程。也就是說，可以實現一個線程或者多個線程一直等待，直到其餘線程執行的操做完成。使用一個給定的計數器進行初始化，該計數器的操做是原子操做，即同時只能有一個線程操做該計數器。java
調用該類await()方法的線程會一直阻塞，直到其餘線程調用該類的countDown()方法，使當前計數器的值變爲0爲止。每次調用該類的countDown()方法，當前計數器的值就會減1。當計數器的值減爲0的時候，全部因調用await()方法而處於等待狀態的線程就會繼續往下執行。這種操做只能出現一次，由於該類中的計數器不能被重置。若是須要一個能夠重置計數次數的版本，能夠考慮使用CyclicBarrier類。數據庫
CountDownLatch支持給定時間的等待，超過必定的時間再也不等待，使用時只須要在countDown()方法中傳入須要等待的時間便可。此時，countDown()方法的方法簽名以下：多線程
public boolean await(long timeout, TimeUnit unit)
使用場景
在某些業務場景中，程序執行須要等待某個條件完成後才能繼續執行後續的操做。典型的應用爲並行計算：當某個處理的運算量很大時，能夠將該運算任務拆分紅多個子任務，等待全部的子任務都完成以後，父任務再拿到全部子任務的運算結果進行彙總。併發
代碼示例
調用ExecutorService類的shutdown()方法，並不會第一時間內把全部線程所有都銷燬掉，而是讓當前已有的線程所有執行完，以後，再把線程池銷燬掉。ide
示例代碼以下：高併發
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await();
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
Thread.sleep(100);
log.info("{}", threadNum);
Thread.sleep(100);
}
}
支持給定時間等待的示例代碼以下：ui
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
test(threadNum);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
});
}
countDownLatch.await(10, TimeUnit.MICROSECONDS);
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
Thread.sleep(100);
log.info("{}", threadNum);
}
}
Semaphore
概述
控制同一時間併發線程的數目。可以完成對於信號量的控制，能夠控制某個資源可被同時訪問的個數。線程
提供了兩個核心方法——acquire()方法和release()方法。acquire()方法表示獲取一個許可，若是沒有則等待，release()方法則是在操做完成後釋放對應的許可。Semaphore維護了當前訪問的個數，經過提供同步機制來控制同時訪問的個數。Semaphore能夠實現有限大小的鏈表。code
使用場景
Semaphore經常使用於僅能提供有限訪問的資源，好比：數據庫鏈接數。資源
代碼示例
每次獲取並釋放一個許可，示例代碼以下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
semaphore.acquire(); //獲取一個許可
test(threadNum);
semaphore.release(); //釋放一個許可
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
每次獲取並釋放多個許可，示例代碼以下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
semaphore.acquire(3); //獲取多個許可
test(threadNum);
semaphore.release(3); //釋放多個許可
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
假設有這樣一個場景，併發過高了，即便使用Semaphore進行控制，處理起來也比較棘手。假設系統當前容許的最高併發數是3，超過3後就須要丟棄，使用Semaphore也能實現這樣的場景，示例代碼以下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
private static final int threadCount = 200;
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < threadCount; i++){
final int threadNum = i;
exec.execute(() -> {
try {
//嘗試獲取一個許可，也能夠嘗試獲取多個許可，
//支持嘗試獲取許可超時設置，超時後再也不等待後續線程的執行
//具體能夠參見Semaphore的源碼
if (semaphore.tryAcquire()) {
test(threadNum);
semaphore.release(); //釋放一個許可
}
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
log.info("finish");
exec.shutdown();
}
private static void test(int threadNum) throws InterruptedException {
log.info("{}", threadNum);
Thread.sleep(1000);
}
}
CyclicBarrier
概述
是一個同步輔助類，容許一組線程相互等待，直到到達某個公共的屏障點，經過它能夠完成多個線程之間相互等待，只有當每一個線程都準備就緒後，才能各自繼續往下執行後面的操做。
與CountDownLatch有類似的地方，都是使用計數器實現，當某個線程調用了CyclicBarrier的await()方法後，該線程就進入了等待狀態，並且計數器執行加1操做，當計數器的值達到了設置的初始值，調用await()方法進入等待狀態的線程會被喚醒，繼續執行各自後續的操做。CyclicBarrier在釋放等待線程後能夠重用，因此，CyclicBarrier又被稱爲循環屏障。
使用場景
能夠用於多線程計算數據，最後合併計算結果的場景。
CyclicBarrier與CountDownLatch的區別
-
CountDownLatch的計數器只能使用一次，而CyclicBarrier的計數器可使用reset()方法進行重置，而且能夠循環使用
-
CountDownLatch主要實現1個或n個線程須要等待其餘線程完成某項操做以後，才能繼續往下執行，描述的是1個或n個線程等待其餘線程的關係。而CyclicBarrier主要實現了多個線程之間相互等待，直到全部的線程都知足了條件以後，才能繼續執行後續的操做，描述的是各個線程內部相互等待的關係。
-
CyclicBarrier可以處理更復雜的場景，若是計算髮生錯誤，能夠重置計數器讓線程從新執行一次。
-
CyclicBarrier中提供了不少有用的方法，好比：能夠經過getNumberWaiting()方法獲取阻塞的線程數量，經過isBroken()方法判斷阻塞的線程是否被中斷。
代碼示例
示例代碼以下。
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++){
final int threadNum = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
private static void race(int threadNum) throws Exception{
Thread.sleep(1000);
log.info("{} is ready", threadNum);
cyclicBarrier.await();
log.info("{} continue", threadNum);
}
}
設置等待超時示例代碼以下：
package io.binghe.concurrency.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++){
final int threadNum = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
private static void race(int threadNum) throws Exception{
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try{
cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
}catch (BrokenBarrierException | TimeoutException e){
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}
在聲明CyclicBarrier的時候，還能夠指定一個Runnable，當線程達到屏障的時候，能夠優先執行Runnable中的方法。
示例代碼以下：
package io.binghe.concurrency.example.aqs;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CyclicBarrier;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;@Slf4jpublic class CyclicBarrierExample { private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> { log.info("callback is running"); }); public static void main(String[] args) throws Exception { ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++){ final int threadNum = i; Thread.sleep(1000); executorService.execute(() -> { try { race(threadNum); } catch (Exception e) { e.printStackTrace(); } }); } executorService.shutdown(); } private static void race(int threadNum) throws Exception{ Thread.sleep(1000); log.info("{} is ready", threadNum); cyclicBarrier.await(); log.info("{} continue", threadNum); }}
相關文章
- 1. 高併發之——淺談AQS中的CountDownLatch、Semaphore與CyclicBarrier
- 2. java高併發7.1 AQS--CountDownLatch, Semaphore, CyclicBarrier
- 3. Java併發（6）- CountDownLatch、Semaphore與AQS
- 4. AQS中的 CountdownLatch、CyclicBarrier、 Semaphore
- 5. J.U.C之AQS之 CountDownLatch、Semaphore、CyclicBarrier
- 6. 併發編程與高併發解決方案學習(J.U.C之AQS-CountDownLatch、Semaphore)
- 7. 高併發之——淺談AQS中的ReentrantLock、ReentrantReadWriteLock、StampedLock與Condition
- 8. CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS
- 9. 【併發編程】--CountDownLatch、CyclicBarrier、Semaphore
- 10. Java併發之CountDownLatch、Semaphore和CyclicBarrier
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • Redis悲觀鎖解決高併發搶紅包的問題 - 紅包項目實戰
- • 漫談MySQL的鎖機制
- • 互聯網組織的未來：剖析GitHub員工的任性之源
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 融合阿里雲，牛客助您找到心儀好工作
- 2. 解決jdbc(jdbctemplate)在測試類時不報錯在TomCatb部署後報錯
- 3. 解決PyCharm GoLand IntelliJ 等 JetBrains 系列 IDE無法輸入中文
- 4. vue+ant design中關於圖片請求不顯示的問題。
- 5. insufficient memory && Native memory allocation (malloc) failed
- 6. 解決IDEA用Maven創建的Web工程不能創建Java Class文件的問題
- 7. [已解決] Error: Cannot download ‘https://start.spring.io/starter.zip?
- 8. 在idea讓java文件夾正常使用
- 9. Eclipse啓動提示「subversive connector discovery」
- 10. 帥某-技巧-快速轉帖博主文章（article_content）
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 高併發之——淺談AQS中的CountDownLatch、Semaphore與CyclicBarrier
- 2. java高併發7.1 AQS--CountDownLatch, Semaphore, CyclicBarrier
- 3. Java併發（6）- CountDownLatch、Semaphore與AQS
- 4. AQS中的 CountdownLatch、CyclicBarrier、 Semaphore
- 5. J.U.C之AQS之 CountDownLatch、Semaphore、CyclicBarrier
- 6. 併發編程與高併發解決方案學習(J.U.C之AQS-CountDownLatch、Semaphore)
- 7. 高併發之——淺談AQS中的ReentrantLock、ReentrantReadWriteLock、StampedLock與Condition
- 8. CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS
- 9. 【併發編程】--CountDownLatch、CyclicBarrier、Semaphore
- 10. Java併發之CountDownLatch、Semaphore和CyclicBarrier