Java并发——结合CountDownLatch源码、Semaphore源码及ReentrantLock源码来看AQS原理

时间:2022-11-08 17:53:23

前言:

  如果说J.U.C包下的核心是什么?那我想答案只有一个就是AQS。那么AQS是什么呢?接下来让我们一起揭开AQS的神秘面纱

AQS是什么?

  AQS是AbstractQueuedSynchronizer的简称。为什么说它是核心呢?是因为它提供了一个基于FIFO的队列和state变量来构建锁和其他同步装置的基础框架。下面是其底层的数据结构。

Java并发——结合CountDownLatch源码、Semaphore源码及ReentrantLock源码来看AQS原理

AQS的特点

  1、其内使用Node实现FIFO(FirstInFirstOut)队列。可用于构建锁或者其他同步装置的基础框架
  2、且利用了一个int类表示状态。在AQS中维护了一个volatile int state,通常表示有线程访问资源的状态,当state>1的时候表示线程重入的数量,主要有三个方法控制:getState(),setState(),CompareAndSetState()。后面的源码分析多用到这几个方法
  3、使用方法是继承,子类通过继承并通过实现它的方法管理其状态(acquire和release)的方法操纵状态。
  4、同时实现排它锁和共享锁模式。实际上AQS功能主要分为两类:独占(只有一个线程能执行)和共享(多个线程同时执行),它的子类要么使用独占功能要么使用共享功能,而ReentrantLock是通过两个内部类来实现独占和共享

CountDownLatch如何借助AQS实现计数功能?

Java并发——结合CountDownLatch源码、Semaphore源码及ReentrantLock源码来看AQS原理

  先来说一下CountDownLatch,CountDownLatch是一个同步辅助类,通过它可以来完成类似阻塞当前线程的功能,即一个或多个线程一起等待,直到其他线程执行的操作完成。要实现上面的功能,CountDownLatch是通过一个给定的原子操作的计数器来实现。调用该类的await()方法的线程会一直处于阻塞状态,直到其他线程调用countDown()方法使得计数器的值变为0之后线程才会执行,这个计数器是不能被重置的。通常这个类会用在程序执行需要等待某个条件完成的场景,比如说并行计算,可将一个数据量很大的计算拆分成一个个子任务,当子任务完成之后,再将最终的结果汇总。每次访问CountDownLatch只能有一个线程,但是这个线程在使用完countDown()方法之后能多个线程能继续运行,而调用await()方法的线程就一定要计数器为0才会运行

  下面来分析CountDownLatch的源码以及如何使用AQS框架

public class CountDownLatch {
/**
* CountDownLatch 实现同步控制
* 底层是使用AQS的state来代表count
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
//初始化内部类实际上是设置AQS的state
Sync(int count) {
setState(count);
} int getCount() {
return getState();
}
//尝试获取共享是看当前的state是否为0
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
/*尝试释放共享锁则是递减计数直到state==0就返回false代表资源已经释放完全否则就会使用CAS来让state减一*/
protected boolean tryReleaseShared(int releases) { for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
} private final Sync sync; /**
* 初始化CountDownLatch,实际上是初始化内部类,实际上是设置AQS的state,count不能小于0
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
} /**
* 这里实际上是调用了AQS里的acquireSharedInterruptibly方法,完成的功能就是先去查看线程是否被中断,中断则抛出异常,没有被中断就会尝试获取共享资源。
* 注意在syn内部类中重写了tryAcquireShared,也就是当state为0就返回1,这时候就会将当前线程放入AQS的队列中去,也就是这时候线程可以不再阻塞而是尝试去获取锁
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
} /**
* 原理同上面方法,但是加了一个时间参数来设置等待的时间
*/
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
} /**
* 这里传入参数为1,同样上面内部类一样重写了AQS的tryReleaseShared方法,使用这个重写的方法来让计数器原子操作的减一
*/
public void countDown() {
sync.releaseShared(1);
} /**
* 就是获取AQS的state
*/
public long getCount() {
return sync.getCount();
} /**
* 转换成字符串的方法
*/
public String toString() {
return super.toString() + "[Count = " + sync.getCount() + "]";
}
}

  由上面代码可看见CountDownLatch实现了AQS的共享锁,原理是操作state来实现计数,并且重写了tryAcquireShared(),tryReleaseShared()等方法

Semaphore是如何借助AQS实现控制并发访问线程个数?

  Semaphore的功能类似于操作系统的信号量,可以很方便的控制某个资源同时被几个线程访问,即做并发访问控制,与CountDownLatch类似,同样是实现获取和释放两个方法。Semaphore的使用场景:常用于仅能提供访问的资源,比如数据库的连接数最大只有30,而应用程序的并发数可能远远大于30,这时候就可以使用Semaphore来控制同时访问的线程数。当Semaphore控制线程数到1的时候就和我们单线程一样了。同样Semaphore说是信号量的意思,我们这里就可以把它理解为十字路口的红绿灯,可以控制车流量(这里是控制线程数)

下面来分析Semaphore的源码以及如何使用AQS框

public class Semaphore implements java.io.Serializable {
    private static final long serialVersionUID = -3222578661600680210L;
/** 所有机制都通过AbstractQueuedSynchronizer子类实现 */
private final Sync sync; /**
* 同样是通过内部类来实现AQS主要功能,使用state来表示许可证数量
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L; Sync(int permits) {
setState(permits);
} final int getPermits() {
return getState();
}
/*
* 不公平的获取方式,会有一个抢占锁的情况,即线程执行顺序会乱
*/
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
/*
* 释放资源
*/
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
} final void reducePermits(int reductions) {
for (;;) {
int current = getState();
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
if (compareAndSetState(current, next))
return;
}
} final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
} /**
* 不公平的sync版本,使用的就是sync定义的不公平锁
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L; NonfairSync(int permits) {
super(permits);
} protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
} /**
* 公平版本,获取锁的线程顺序就是线程启动的顺序。具体是使用hasQueuedPredecessors()方法判断“当前线程”是不是CLH队列中的第一个线程。
* 若不是的话,则返回-1,是就设置获取许可证,并检查许可证数量是否足够
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L; FairSync(int permits) {
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;
}
}
} /**
* 默认使用不公平的版本,如果需要公平的,则需要两个参数
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
} public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
} /**
* 分析同CountDownLatch中的类似方法,具体的实现都是内部类中的获取方法,这里是获取一个许可
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
} /**
*功能同上,但是这里不会检测线程是否被中断
*/
public void acquireUninterruptibly() {
sync.acquireShared(1);
} /**
* 尝试获取
*/
public boolean tryAcquire() {
return sync.nonfairTryAcquireShared(1) >= 0;
} /**
* 在一段时间内一直尝试获取许可
*/
public boolean tryAcquire(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
} /**
* 当前线程释放一个许可证
*/
public void release() {
sync.releaseShared(1);
} /**
* 可以规定一个线程获得许可证的数量
*/
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
} public void acquireUninterruptibly(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireShared(permits);
} public boolean tryAcquire(int permits) {
if (permits < 0) throw new IllegalArgumentException();
return sync.nonfairTryAcquireShared(permits) >= 0;
} public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
} /**
* 同样可以规定一个线程释放许可证的数量
*/
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
} /**
* 当前的许可还剩几个
*/
public int availablePermits() {
return sync.getPermits();
} /**
* 销毁所有许可
*/
public int drainPermits() {
return sync.drainPermits();
} protected void reducePermits(int reduction) {
if (reduction < 0) throw new IllegalArgumentException();
sync.reducePermits(reduction);
} public final boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
} public final int getQueueLength() {
return sync.getQueueLength();
} protected Collection<Thread> getQueuedThreads() {
return sync.getQueuedThreads();
}
public String toString() {
return super.toString() + "[Permits = " + sync.getPermits() + "]";
}}

  上面对于Semaphore的一些重要内部类和常用方法进行了解释,与CountDownLatch很类似,实现的都是共享的功能,即Semaphore允许得到许可证的线程同时执行,而CountDownLatch允许调用countDown()方法的线程同时执行。并且都是通过内部类实现的。相信看到这里,你能越来越看见AQS为什么被称作JUC包的核心。下面就来介绍一下ReentrantLock

ReentrantLock是如何借助AQS实现锁机制

  ReentrantLock是可重入锁,前面博客中写到synchronized实现的锁也是可重入的。不过synchronized是基于JVM指令实现,而ReentrantLock是使用Java代码实现的。ReentrantLock重点就是需要我们手动声明加锁和释放锁,如果手工忘记释放锁,很有可能就会导致死锁,即资源永远都被锁住,其他线程无法得到,当前线程也释放不出去。ReentrantLock实现的是自旋锁,通过循环调用CAS操作实现加锁,避免了线程进入内核态的阻塞状态,所以性能较好。ReentrantLock内部同样实现了公平锁和非公平锁。事实上Synchronized能做的ReentrantLock都能做,但是反过来就不一样了、

  经过前面的源码分析我们发现核心的都在当前类的内部类里,而当前类的一些方法不过是使用的内部类以及AQS的方法罢了,所以下面我们就来分析ReentrantLock中的三个内部类。

public class ReentrantLock implements Lock, java.io.Serializable {
private static final long serialVersionUID = 7373984872572414699L;
/** 同步的机制都是通过内部类来实现的 */
private final Sync sync; /**
* 在ReentrantLock中state表示的是线程重入锁的次数,当state为0时才能释放锁
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L; /**
* 这个抽象方法提供给公平锁和不公平锁来单独实现,父类不实现
*/
abstract void lock(); /**
* 首先得到当前线程,而后获取state,如果state为0,也就是没有线程获得当前锁,那么就设置当前线程拥有当前锁的独占访问权,并且返回true。
* 如果state不为0,那么就看当前线程是否是已经获得过锁的线程,如果是就让state+=acquire,acquire一般是1,即表示线程重入并且返回true。
* 上面两个条件都不满足就代表是锁被其他线程获取了,当前线程获取不到,所以返回false
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
/**
* 先判断当前线程等不等于拥有锁的线程,不等于就会抛异常,也就是释放不了。
* 等于之后就看state-releases是否为0,当为0的时候就代表释放完全。
* 可以设置锁的状态为没有线程拥有,从而让锁能被其他线程竞争,否则就设置state,代表线程重入该锁,并且线程还没释放完全。
*/
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
/*
*该方法检验当前线程是否是锁的独占者
*/
protected final boolean isHeldExclusively() {
return getExclusiveOwnerThread() == Thread.currentThread();
}
/*
*该方法是创建一个条件锁,本文不做具体分析
*/
final ConditionObject newCondition() {
return new ConditionObject();
} // Methods relayed from outer class final Thread getOwner() {
return getState() == 0 ? null : getExclusiveOwnerThread();
} final int getHoldCount() {
return isHeldExclusively() ? getState() : 0;
} final boolean isLocked() {
return getState() != 0;
} /**
* 使得该类从流中能重构实例,并且会重置为解锁状态
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
setState(0);
}
} /**
* Sync 的不公平版本
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L; /**
* 将state从0更新到1成功的话就让当前线程获取锁,否则就会尝试获得锁和获取当前节点的前一节点,并判断这一个节点是否为头节点,即当前线程是不是头节点的直接后继。
* 如果两个中有一个失败则线程中断,进入阻塞状态
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
} protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
} /**
* Sync 的公平版本
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
/**
* 尝试获得锁和获取当前节点的前一节点,并判断这一个节点是否为头节点,即当前线程是不是头节点的直接后继,如果两个中有一个失败则线程中断,进入阻塞状态。
* 也就是一定按照队列中线程的顺序来实现
*/
final void lock() {
acquire(1);
} /**
* 跟不公平的版本相比其实是在state为0的时候检查当前线程是不是在队列的头部节点的直接后继,来达到公平的概念
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
}

  ReentrantLock和上面两个类最不同的莫过于ReentrantLock使用的是独占功能,即每次只能有一个线程来获取ReentrantLock类。ReentrantLock类下还有很多方法,这里就不一一介绍,但是本质都是内部类中的实现以及AQS的一些调用

总结:

  AQS只是一个基础的框架,里面最核心的就是维护了state变量和CHL队列,而其他的类全部都是通过继承的方法进行扩展,虽然没有直接说源码,但是通过上面三个主要类的源码分析再去看AQS已经不是难事。继承主要改变的就是获取和释放的方法,通过这两个方法来对state和队列进行操作达到我们能够进行的并发控制的功能,事实上J.U.C包下的类和能够实现的功能远不止这三个,后面会选择重点的来介绍。

Java并发——结合CountDownLatch源码、Semaphore源码及ReentrantLock源码来看AQS原理的更多相关文章

  1. java并发初探CountDownLatch

    java并发初探CountDownLatch CountDownLatch是同步工具类能够允许一个或者多个线程等待直到其他线程完成操作. 当前前程A调用CountDownLatch的await方法进入 ...

  2. java并发中CountDownLatch的使用

    文章目录 主线程等待子线程全都结束之后再开始运行 等待所有线程都准备好再一起执行 停止CountdownLatch的await java并发中CountDownLatch的使用 在java并发中,控制 ...

  3. Java并发编程基础三板斧之Semaphore

    引言 最近可以进行个税申报了,还没有申报的同学可以赶紧去试试哦.不过我反正是从上午到下午一直都没有成功的进行申报,一进行申报 就返回"当前访问人数过多,请稍后再试".为什么有些人就 ...

  4. JAVA并发,CountDownLatch使用

    该文章转自:http://www.itzhai.com/the-introduction-and-use-of-a-countdownlatch.html CountDownLatch 1.类介绍 一 ...

  5. Java并发编程-CountDownLatch

    基于AQS的前世今生,来学习并发工具类CountDownLatch.本文将从CountDownLatch的应用场景.源码原理解析来学习这个并发工具类. 1. 应用场景 CountDownLatch是并 ...

  6. java并发编程学习:用 Semaphore (信号量)控制并发资源

    并发编程这方面以前关注得比较少,恶补一下,推荐一个好的网站:并发编程网 - ifeve.com,上面全是各种大牛原创或编译的并发编程文章. 今天先来学习Semaphore(信号量),字面上看,根本不知 ...

  7. 【java并发】线程同步工具Semaphore的使用

    Semaphore通常用于限制可以访问某些资源(物理或逻辑的)的线程数目,我们可以自己设定最大访问量.它有两个很常用的方法是acquire()和release(),分别是获得许可和释放许可.  官方J ...

  8. java并发编程CountDownLatch

    /** * CountDownLatch用法 * CountDownLatch类位于java.util.concurrent包下,利用它可以实现类似计数器的功能.比如有一个任务A, * 它要等待其他4 ...

  9. Java并发案例04---生产者消费者问题03--使用ReentrantLock

    /** * 面试题:写一个固定容量同步容器,拥有put和get方法,以及getCount方法, * 能够支持2个生产者线程以及10个消费者线程的阻塞调用 * * 使用wait和notify/notif ...

随机推荐

  1. sicily vector有序插入

    实现了简单的vector有序插入,这个题目值得注意的点是1.当vector为空时,需要判断再排除 2.迭代器的使用是此段代码的特点 int insertVector(vector<int> ...

  2. HTML 学习笔记 CSS&lpar;选择器4&rpar;

    CSS 后代选择器 后代选择器(descendant selector)又称为包含选择器.后代选择器可以选择作为某元素后代的元素. 根据上下文选择元素 我们可以定义后代选择器来创建一些规则,使这些规则 ...

  3. python中random模块使用

  4. asp&period;net下载文件的几种方法

    最近做东西遇到了下载相关的问题.在这里总结一下自己处理的方法. 1.以字节流的形式向页面输出数据以下载Excel为例子. string path=Server.MapPath("文件路径&q ...

  5. docker遇到超时

    1. 直接通过docker拉取镜像遇到的问题:熟悉的timeout!!! [root@localhost ~]# docker search mysql INDEX NAME DESCRIPTION ...

  6. 利用vi编辑器创建和编辑正文文件(二)

    末行模式下的命令 1.       w:写文件,将编辑的内容保存到文件系统. 2.       w!:如果只读文件,强制写入系统. 3.       q!:退出vi,但文件内容修改的话,系统要提示是否 ...

  7. bzoj3879

    题解: 后缀数组 然后把读入的内容去重,按照rank排序 然后用单调栈处理一下 代码: #include<bits/stdc++.h> using namespace std; typed ...

  8. CSS------Filter属性的使用方法

    转载: http://www.w3cplus.com/css3/ten-effects-with-css3-filter

  9. Metasploit 学习

    知识准备:CCNA/CCNP基础计算机知识框架:操作系统.汇编.数据库.网络.安全 木马.灰鸽子.口令破解.用后门拷贝电脑文件 渗透测试工程师 penetration test engineer &l ...

  10. 如何搭建struts2框架

    一.首先,下载5个Struts2核心jar包: commons-logging-1.1.1.jar freemarker-2.3.15.jar ognl-2.7.3.jar struts2-core- ...