AQS的同步组件有哪些呢?

摘要:aqs 的全称为 abstract queued synchronizer,是在 j.u.c(java.util.concurrent)下子包中的类。
aqs 的全称为 abstract queued synchronizer,是在 j.u.c(java.util.concurrent)下子包中的类。
一、aqs 的设计如下
(1)使用 node 实现 fifo 队列,可以用于构建锁或者其他同步装置的基础框架。
(2)利用了一个 int 类型表示状态
在 aqs 类中,有一个叫做 state 的成员变量。
基于 aqs 有一个同步组件 reentrantlock,在 reentrantlock 中,state 表示获取锁的线程数。如果 state=0, 则表示还没有线程获取锁;如果 state=1, 则表示有线程获取了锁;如果 state>1,则表示重入锁的数量。
(3)使用方法是继承
设计上基于模板方法,使用时需要继承 aqs,并覆写其中的方法
(4)子类通过继承并通过实现它的方法管理其状态 {acquire 和 release} 的方法操纵状态
(5)可以同时实现排它锁和共享锁模式(独占、共享)
站在使用者的角度,aqs 的功能主要分为两类:独占模式和共享模式。它的所有子类中要么实现并使用了它的独占功能的 api,要么使用了共享锁的功能,而不会同时使用两套 api。即便是它最有名的子类 ——reentrantreadwritelock,也是通过两个内部类 ——readlock(读锁)和 writelock(写锁)两套 api 来实现的。
二、aqs 内部实现的大体思路
首先,aqs 内部维护了一个 clh 队列来管理锁,线程会首先尝试获取锁,如果失败,就将当前线程以及等待等信息封装成一个 node 节点,加入到同步队列 syncqueue,接着会不断循环尝试获取锁,获取锁的条件是当前节点为 head 的直接后继节点才会尝试获取锁,如果失败,就会阻塞自己,直到自己被唤醒。
而持有锁的线程释放锁的时候,会唤醒队列中的后继线程。基于这些基础的设计和思路,jdk 提供了许多基于 aqs 的子类,比如:countdownlatch、semaphore、cyclicbarrier、reentrantlock、condition、futuretask 等
三、aqs 同步组件
countdownlatch:闭锁,通过一个计数,来保证线程是否一直阻塞
semaphore:控制同一时间并发线程的数目
cyclicbarrier:与 countdownlatch 类似,都能阻塞进程;
reentrantlock:可重入锁
condition: 在使用时需要 reentrantlock
futuretask: 对比 runnable 和 callable
1.countdownlatch
同步辅助类,通过它可以阻塞当前线程。也就是说,能够实现一个线程或者多个线程一直等待,直到其他线程执行的操作完成。使用一个给定的计数器进行初始化,该计数器的操作是原子操作,即同时只能有一个线程操作该计数器。
调用该类 await () 方法的线程会一直阻塞,直到其他线程调用该类的 countdown () 方法,使当前计数器的值变为 0 为止。
每次调用该类的 countdown () 方法,当前计数器的值就会减 1。当计数器的值减为 0 的时候,所有因调用 await () 方法而处于等待状态的线程就会继续往下执行。这种操作只能出现一次,因为该类中的计数器不能被重置。如果需要一个可以重置计数次数的版本,可以考虑使用 cyclicbarrier 类。
countdownlatch 支持给定时间的等待,超过一定的时间不再等待,使用时只需要在 await () 方法中传入需要等待的时间即可。此时,await () 方法的方法签名如下:
public boolean await(long timeout, timeunit unit)countdownlatch 使用场景 在某些业务场景中,程序执行需要等待某个条件完成后才能继续执行后续的操作。典型的应用为并行计算:当某个处理的运算量很大时,可以将该运算任务拆分成多个子任务,等待所有的子任务都完成之后,父任务再拿到所有子任务的运算结果进行汇总。
调用 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;@slf4jpublic 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 { 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;@slf4jpublic 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 { 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); }}  
2.semaphore
控制同一时间并发线程的数目。能够完成对于信号量的控制,可以控制某个资源可被同时访问的个数。 提供了两个核心方法 ——acquire () 方法和 release () 方法。acquire () 方法表示获取一个许可,如果没有则等待,release () 方法则是在操作完成后释放对应的许可。
semaphore 维护了当前访问的个数,通过提供同步机制来控制同时访问的个数。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;@slf4jpublic 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 { 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;@slf4jpublic 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 { 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;@slf4jpublic 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 { 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); }}  
3.cyclicbarrier
是一个同步辅助类,允许一组线程相互等待,直到到达某个公共的屏障点,通过它可以完成多个线程之间相互等待,只有当每个线程都准备就绪后,才能各自继续往下执行后面的操作。
与 countdownlatch 有相似的地方,都是使用计数器实现,当某个线程调用了 cyclicbarrier 的 await () 方法后,该线程就进入了等待状态,而且计数器执行加 1 操作,当计数器的值达到了设置的初始值,调用 await () 方法进入等待状态的线程会被唤醒,继续执行各自后续的操作。
cyclicbarrier 在释放等待线程后可以重用,所以,cyclicbarrier 又被称为循环屏障。
cyclicbarrier 使用场景
可以用于多线程计算数据,最后合并计算结果的场景
cyclicbarrier 与 countdownlatch 的区别
(1)countdownlatch 的计数器只能使用一次,而 cyclicbarrier 的计数器可以使用 reset () 方法进行重置,并且可以循环使用
(2)countdownlatch 主要实现 1 个或 n 个线程需要等待其他线程完成某项操作之后,才能继续往下执行,描述的是 1 个或 n 个线程等待其他线程的关系。而 cyclicbarrier 主要实现了多个线程之间相互等待,直到所有的线程都满足了条件之后,才能继续执行后续的操作,描述的是各个线程内部相互等待的关系。
(3)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;@slf4jpublic 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 { 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.*;@slf4jpublic 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 { 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 { 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); }}  
4.reentrantlock 与锁
java 中主要分为两类锁,一类是 synchronized 修饰的锁,另外一类就是 j.u.c 中提供的锁。j.u.c 中提供的核心锁就是 reentrantlock。 reentrantlock(可重入锁)与 synchronized 区别:
(1)可重入性
二者都是同一个线程进入 1 次,锁的计数器就自增 1,需要等到锁的计数器下降为 0 时,才能释放锁。
(2)锁的实现
synchronized 是基于 jvm 实现的,而 reentrantlock 是 jdk 实现的
(3)性能的区别
synchronized 优化之前性能比 reentrantlock 差很多,但是自从 synchronized 引入了偏向锁,轻量级锁也就是自旋锁后,性能就差不多了。
(4)功能区别
便利性:synchronized 使用起来比较方便,并且由编译器保证加锁和释放锁;reentrantlock 需要手工声明加锁和释放锁,最好是在 finally 代码块中声明释放锁。
锁的灵活度和细粒度:在这点上 reentrantlock 会优于 synchronized
reentrantlock 独有的功能如下: (1)reentrantlock 可指定是公平锁还是非公平锁。而 synchronized 只能是非公平锁。所谓的公平锁就是先等待的线程先获得锁。
(2)提供了一个 condition 类,可以分组唤醒需要唤醒的线程。而 synchronized 只能随机唤醒一个线程,或者唤醒全部的线程
(3)提供能够中断等待锁的线程的机制,lock.lockinterruptibly ()。reentrantlock 实现是一种自旋锁,通过循环调用 cas 操作来实现加锁,性能上比较好是因为避免了使线程进入内核态的阻塞状态。 synchronized 能做的事情 reentrantlock 都能做,而 reentrantlock 有些能做的事情,synchronized 不能做。
在性能上,reentrantlock 不会比 synchronized 差。 synchronized 的优势:
(1)不用手动释放锁,jvm 自动处理,如果出现异常,jvm 也会自动释放锁
(2)jvm 用 synchronized 进行管理锁定请求和释放时,jvm 在生成线程转储时能够锁定信息,这些对调试非常有价值,因为它们能标识死锁或者其他异常行为的来源。而 reentrantlock 只是普通的类,jvm 不知道具体哪个线程拥有 lock 对象。
(3)synchronized 可以在所有 jvm 版本中工作,reentrantlock 在某些 1.5 之前版本的 jvm 中可能不支持 reentrantlock 中的部分方法说明:
boolean trylock (): 仅在调用时锁定未被另一个线程保持的情况下才获取锁定
boolean trylock (long, timeunit): 如果锁定在给定的等待时间内没有被另一个线程保持,且当前线程没有被中断,则获取这个锁定。
void lockinterruptibly (): 如果当前线程没有被中断,就获取锁定;如果被中断,就抛出异常
boolean islocked (): 查询此锁定是否由任意线程保持
boolean isheldbycurrentthread (): 查询当前线程是否保持锁定状态;
boolean isfair (): 判断是否是公平锁
boolean hasqueuedthread (thread):查询指定线程是否在等待获取此锁定
boolean hasqueuedthreads (): 查询是否有线程正在等待获取此锁定
boolean getholdcount (): 查询当前线程保持锁定的个数
示例代码如下:
package io.binghe.concurrency.example.lock;import lombok.extern.slf4j.slf4j;import java.util.concurrent.countdownlatch;import java.util.concurrent.executorservice;import java.util.concurrent.executors;import java.util.concurrent.semaphore;import java.util.concurrent.locks.lock;import java.util.concurrent.locks.reentrantlock;@slf4jpublic class lockexample { //请求总数 public static int clienttotal = 5000; //同时并发执行的线程数 public static int threadtotal = 200; public static int count = 0; private static final lock lock = new reentrantlock(); public static void main(string[] args) throws interruptedexception { executorservice executorservice = executors.newcachedthreadpool(); final semaphore semaphore = new semaphore(threadtotal); final countdownlatch countdownlatch = new countdownlatch(clienttotal); for(int i = 0; i { try{ semaphore.acquire(); add(); semaphore.release(); }catch (exception e){ log.error(exception, e); } countdownlatch.countdown(); }); } countdownlatch.await(); executorservice.shutdown(); log.info(count:{}, count); } private static void add(){ lock.lock(); try{ count ++; }finally { lock.unlock(); } }}  
5.reentrantreadwritelock
在没有任何读写锁的时候,才可以取得写锁。如果一直有读锁存在,则无法执行写锁,这就会导致写锁饥饿。
示例代码如下:
package io.binghe.concurrency.example.lock;import lombok.extern.slf4j.slf4j;import java.util.map;import java.util.set;import java.util.treemap;import java.util.concurrent.locks.lock;import java.util.concurrent.locks.reentrantlock;import java.util.concurrent.locks.reentrantreadwritelock;@slf4jpublic class lockexample { private final map map = new treemap(); private final reentrantreadwritelock lock = new reentrantreadwritelock(); private final lock readlock = lock.readlock(); private final lock writelock = lock.writelock(); public data get(string key){ readlock.lock(); try{ return map.get(key); }finally { readlock.unlock(); } } public set getallkeys(){ readlock.lock(); try{ return map.keyset(); }finally { readlock.unlock(); } } public data put(string key, data value){ writelock.lock(); try{ return map.put(key, value); }finally { writelock.unlock(); } } class data{ }}  
6.stampedlock
控制锁三种模式:写、读、乐观读。 stampedlock 的状态由版本和模式两个部分组成,锁获取方法返回的是一个数字作为票据,用相应的锁状态来表示并控制相关的访问,数字 0 表示没有写锁被授权访问。
在读锁上分为悲观锁和乐观锁,乐观读就是在读操作很多,写操作很少的情况下,可以乐观的认为写入和读取同时发生的几率很小。
因此,不悲观的使用完全的读取锁定。程序可以查看读取资料之后,是否遭到写入进行了变更,再采取后续的措施,这样的改进可以大幅度提升程序的吞吐量。
总之,在读线程越来越多的场景下,stampedlock 大幅度提升了程序的吞吐量。 stampedlock 源码中的案例如下,这里加上了注释
class point {private double x, y;private final stampedlock sl = new stampedlock();void move(double deltax, double deltay) { // an exclusively locked methodlong stamp = sl.writelock();try {x += deltax;y += deltay;} finally {sl.unlockwrite(stamp);}}//下面看看乐观读锁案例double distancefromorigin() { // a read-only methodlong stamp = sl.tryoptimisticread(); //获得一个乐观读锁double currentx = x, currenty = y; //将两个字段读入本地局部变量if (!sl.validate(stamp)) { //检查发出乐观读锁后同时是否有其他写锁发生?stamp = sl.readlock(); //如果没有,我们再次获得一个读悲观锁try {currentx = x; // 将两个字段读入本地局部变量currenty = y; // 将两个字段读入本地局部变量} finally {sl.unlockread(stamp);}}return math.sqrt(currentx * currentx + currenty * currenty);}//下面是悲观读锁案例void moveifatorigin(double newx, double newy) { // upgrade// could instead start with optimistic, not read modelong stamp = sl.readlock();try {while (x == 0.0 && y == 0.0) { //循环,检查当前状态是否符合long ws = sl.tryconverttowritelock(stamp); //将读锁转为写锁if (ws != 0l) { //这是确认转为写锁是否成功stamp = ws; //如果成功 替换票据x = newx; //进行状态改变y = newy; //进行状态改变break;} else { //如果不能成功转换为写锁sl.unlockread(stamp); //我们显式释放读锁stamp = sl.writelock(); //显式直接进行写锁 然后再通过循环再试}}} finally {sl.unlock(stamp); //释放读锁或写锁}}}示例代码如下:package io.binghe.concurrency.example.lock;import lombok.extern.slf4j.slf4j;import java.util.concurrent.countdownlatch;import java.util.concurrent.executorservice;import java.util.concurrent.executors;import java.util.concurrent.semaphore;import java.util.concurrent.locks.stampedlock;@slf4jpublic class lockexample { //请求总数 public static int clienttotal = 5000; //同时并发执行的线程数 public static int threadtotal = 200; public static int count = 0; private static final stampedlock lock = new stampedlock(); public static void main(string[] args) throws interruptedexception { executorservice executorservice = executors.newcachedthreadpool(); final semaphore semaphore = new semaphore(threadtotal); final countdownlatch countdownlatch = new countdownlatch(clienttotal); for(int i = 0; i { try{ semaphore.acquire(); add(); semaphore.release(); }catch (exception e){ log.error(exception, e); } countdownlatch.countdown(); }); } countdownlatch.await(); executorservice.shutdown(); log.info(count:{}, count); } private static void add(){//加锁时返回一个long类型的票据 long stamp = lock.writelock(); try{ count ++; }finally { //释放锁的时候带上加锁时返回的票据 lock.unlock(stamp); } }}总结:
(1)当只有少量竞争者时,synchronized 是一个很好的通用锁实现
(2)竞争者不少,但是线程的增长趋势是可预估的,此时,reentrantlock 是一个很好的通用锁实现
(3)synchronized 不会引发死锁,其他的锁使用不当可能会引发死锁。  
7.condition
condition 是一个多线程间协调通信的工具类,condition 除了实现 wait 和 notify 的功能以外,它的好处在于一个 lock 可以创建多个 condition,可以选择性的通知 wait 的线程 特点:
(1)condition 的前提是 lock,由 aqs 中 newcondition () 方法 创建 condition 的对象
(2)condition await 方法表示线程从 aqs 中移除,并释放线程获取的锁,并进入 condition 等待队列中等待,等待被 signal
(3)condition signal 方法表示唤醒对应 condition 等待队列中的线程节点,并加入 aqs 中,准备去获取锁。
示例代码如下:
package io.binghe.concurrency.example.lock;import lombok.extern.slf4j.slf4j;import java.util.concurrent.locks.condition;import java.util.concurrent.locks.reentrantlock;@slf4jpublic class lockexample { public static void main(string[] args) { reentrantlock reentrantlock = new reentrantlock(); condition condition = reentrantlock.newcondition(); new thread(() -> { try { reentrantlock.lock(); log.info(wait signal); // 1 condition.await(); } catch (interruptedexception e) { e.printstacktrace(); } log.info(get signal); // 4 reentrantlock.unlock(); }).start(); new thread(() -> { reentrantlock.lock(); log.info(get lock); // 2 try { thread.sleep(3000); } catch (interruptedexception e) { e.printstacktrace(); } condition.signalall(); log.info(send signal ~ ); // 3 reentrantlock.unlock(); }).start(); }}


接触器上的A1和A2接什么线 接触器线圈是a1进a2出吗
鼎盛合——汽车轮胎数显胎压计方案
因为芯片短缺,特斯拉的加州工厂宣布停产
Atmel高性能手写笔方案具多重感测功能
扩声系统有什么组成_扩声系统分类
AQS的同步组件有哪些呢?
浅谈PULSAR模拟电压可变衰减器,它都有哪些特征
HTC回应出售上海手机工厂 手机的生产制造不受影响
通用KTF滑块系列直线位移传感器的功能特性分析
在政府支持IPv6升级改造的方案中有哪五项重点任务?
乌拉圭央行推出了六个月的数字纸币发行试点计划
GPRS DTU快速应用指南
Wolfram语言与Mathematica 13.2 版本(6)
人工智能数据标注的应用
祝贺!国家高分中心商业遥感交易服务平台正式试运行
安森美半导体LED驱动器方案为互联照明增添智能
MIT推新型机器算法,可破译消失已久的古语言
一次电池与二次电池的有哪些异同点?
音频/视频接口的EMI/EMC抑制
Mini USB在USB中的重要性