摘要: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可以实现有限大小的链表。
使用场景
Sema