帶你熟悉3種AQS的線程併發工具的用法

来源:https://www.cnblogs.com/huaweiyun/archive/2023/01/31/17079017.html
-Advertisement-
Play Games

摘要：AQS 的全稱為（AbstractQueuedSynchronizer），AQS 是一個用來構建鎖和同步器的框架，使用 AQS 能簡單且高效地構造出應用廣泛的大量的同步器。 本文分享自華為雲社區《【高併發】AQS中的CountDownLatch、Semaphore與CyclicBarrier核 ...


摘要：AQS 的全稱為（AbstractQueuedSynchronizer），AQS 是一個用來構建鎖和同步器的框架，使用 AQS 能簡單且高效地構造出應用廣泛的大量的同步器。

本文分享自華為雲社區《【高併發】AQS中的CountDownLatch、Semaphore與CyclicBarrier核心用法》，作者： 冰 河。

AQS 的全稱為（AbstractQueuedSynchronizer），AQS 是一個用來構建鎖和同步器的框架，使用 AQS 能簡單且高效地構造出應用廣泛的大量的同步器。本文主要講述AQS中的CountDownLatch、Semaphore與CyclicBarrier核心用法。

CountDownLatch

概述

同步輔助類，通過它可以阻塞當前線程。也就是說，能夠實現一個線程或者多個線程一直等待，直到其他線程執行的操作完成。使用一個給定的計數器進行初始化，該計數器的操作是原子操作，即同時只能有一個線程操作該計數器。

調用該類await()方法的線程會一直阻塞，直到其他線程調用該類的countDown()方法，使當前計數器的值變為0為止。每次調用該類的countDown()方法，當前計數器的值就會減1。當計數器的值減為0的時候，所有因調用await()方法而處於等待狀態的線程就會繼續往下執行。這種操作只能出現一次，因為該類中的計數器不能被重置。如果需要一個可以重置計數次數的版本，可以考慮使用CyclicBarrier類。

CountDownLatch支持給定時間的等待，超過一定的時間不再等待，使用時只需要在await()方法中傳入需要等待的時間即可。此時，await()方法的方法簽名如下：

public boolean await(long timeout, TimeUnit unit)

使用場景

在某些業務場景中，程式執行需要等待某個條件完成後才能繼續執行後續的操作。典型的應用為並行計算：當某個處理的運算量很大時，可以將該運算任務拆分成多個子任務，等待所有的子任務都完成之後，父任務再拿到所有子任務的運算結果進行彙總。

代碼示例

調用ExecutorService類的shutdown()方法，並不會第一時間內把所有線程全部都銷毀掉，而是讓當前已有的線程全部執行完，之後，再把線程池銷毀掉。

示例代碼如下：

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);
 }
}

支持給定時間等待的示例代碼如下：

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可以實現有限大小的鏈表。

使用場景

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;
@Slf4j
public 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);
 }
}

 

點擊關註，第一時間瞭解華為雲新鮮技術~


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 1.編程初步
    程式設計基礎 基礎知識 什麼是程式？ 為進行某項活動的步驟，電腦的程式，為得到某種結果，通過電腦語言表達的指令序列。 什麼是程式設計？ 計算思維，是運用電腦科學的基礎概念進行問題求解，系統設計，以及人類行為理解等涵蓋電腦科學之廣度的一系列思維活動。 計算思維的特點： 1.滿足電腦程式執行的 ...
  • Gin框架實戰——HTML渲染
    Gin框架實戰——HTML渲染 最近使用Go的Gin框架做了個簡單的前端網頁，記錄一下細節~ 1.載入靜態文件 由於網頁需要使用css、圖片等渲染，而靜態文件必須先聲明：否則模板中調用載入不出來，這個很重要，即使你把文件放到對應路徑下，html中也寫了相應的路徑，但是開啟go服務端的網頁，會顯示不出 ...
  • python教程：shutil高級文件操作
    1、shutil高級文件操作模塊 shutil模塊提供了大量的文件的高級操作。特別針對文件拷貝和刪除，主要功能為目錄和文件操作以及壓縮操作。對單個文件的操作也可參見os模塊。 2、shutil模塊的拷貝方法 >>> import shutil >>> shutil.chown('test.txt', ...
  • DDS結構的FPGA實現
    一、DDS工作原理 以正弦信號為例，DDS大概就是將M個點的一個周期的正弦序列存入ROM中，序列數據的地址就是正弦信號的相位； 通過修改頻率控制字(Fword)來改變每隔多少個地址取ROM里的數據進行輸出。頻率控制字越大，從ROM取出的數據點就越少，點數越少，輸出一個周期信號的時間就越短，從而改變了 ...
  • SpringBoot項目動態定時任務之 ScheduledTaskRegistrar（解決方案一）
    在做SpringBoot項目的過程中，有時客戶會提出按照指定時間執行一次業務的需求。 在單一使用ScheduledTaskRegistrar類解決定時任務問題的時候，可能會達不到預期的動態調整定時任務的效果。 ...
  • 【manim】之圓規動畫
    概要 前端時間做尺規作圖相關的動畫的時候，封裝了一個圓規的動畫，順便研究了下 manim 庫的動畫函數。 manim 本身就是做動畫的庫，所以，基於它封裝自定義的動畫非常方便。 動畫原理 對於單個的元素，manim本身就提供了非常多的動畫函數。 比如：創建/消除的動畫，移動元素的動畫，旋轉元素的動畫 ...
  • 記一次線上FGC問題排查
    本文記錄一次線上 GC 問題的排查過程與思路，希望對各位讀者有所幫助。過程中也走了一些彎路，現在有時間沉澱下來思考並總結出來分享給大家，希望對大家今後排查線上 GC 問題有幫助。 ...
  • 垃圾收集器必問系列—ZGC
    本文已收錄至Github，推薦閱讀 👉 Java隨想錄 微信公眾號：Java隨想錄 CSDN： 碼農BookSea 人的一切痛苦，本質上都是對自己的無能的憤怒。——王小波 ZGC有人稱它為Zero GC，其實“Z”並非什麼專業名詞的縮寫，這款收集器的名字就叫作Z Garbage Collector ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類