---
title: AQS AQS
category: Java
tag:
- Java
---
- AQSAQS
- `CountDownLatch` `CyclicBarrier`
- `Semaphore`
- ......
## AQS
AQS `AbstractQueuedSynchronizer` `java.util.concurrent.locks`
AQS
```java
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
}
```
AQS AQS `ReentrantLock``Semaphore` `ReentrantReadWriteLock``SynchronousQueue``FutureTask`(jdk1.7) AQS
## AQS
> AQS
AQS
### AQS
AQS AQS **CLH **
> CLH(Craig,Landin,and Hagersten)AQS CLH Node
AQS(`AbstractQueuedSynchronizer`)
AQS int FIFO AQS CAS
```java
private volatile int state;//volatile
```
`protected` `getState()``setState()``compareAndSetState()`
```java
//
protected final int getState() {
return state;
}
//
protected final void setState(int newState) {
state = newState;
}
//CASupdateexpect
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
```
### AQS
AQS
**1)Exclusive**
`ReentrantLock``ReentrantLock` `ReentrantLock`
- ****
- **** CAS
> `ReentrantLock` https://www.javadoop.com/post/AbstractQueuedSynchronizer-2
** `ReentrantLock` **
`ReentrantLock` `boolean` true
```java
/** Synchronizer providing all implementation mechanics */
private final Sync sync;
public ReentrantLock() {
//
sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
```
`ReentrantLock` `lock`
```java
static final class FairSync extends Sync {
final void lock() {
acquire(1);
}
// AbstractQueuedSynchronizer.acquire(int arg)
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// 1.
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;
}
}
```
`lock`
```java
static final class NonfairSync extends Sync {
final void lock() {
// 2. CAS
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
// AbstractQueuedSynchronizer.acquire(int arg)
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
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;
}
```
1. lock CAS
2. CAS `tryAcquire` `tryAcquire` state == 0 CAS
CAS
**2)Share**
`Semaphore/CountDownLatch``Semaphore``CountDownLatCh` `CyclicBarrier``ReadWriteLock`
`ReentrantReadWriteLock` `ReentrantReadWriteLock`
state /AQS
### AQS
1. `AbstractQueuedSynchronizer` state
2. AQS
**AQS AQS **
```java
protected boolean tryAcquire(int)//truefalse
protected boolean tryRelease(int)//truefalse
protected boolean tryAcquireShared(int)//0
protected boolean tryReleaseShared(int)//truefalse
protected boolean isHeldExclusively()//condition
```
**** `protected`
[Java8 yyds!](https://mp.weixin.qq.com/s/zpScSCktFpnSWHWIQem2jg)
AQS `final`
`ReentrantLock` state 0A `lock()` `tryAcquire()` `state+1` `tryAcquire()` A `unlock()` `state=`0A state state
`CountDownLatch` N state N N N ` countDown()` state CAS(Compare and Swap) 1( `state=0` ) `unpark()` `await()`
`tryAcquire-tryRelease``tryAcquireShared-tryReleaseShared` AQS `ReentrantReadWriteLock`
AQS
- [JavaAQS](https://www.cnblogs.com/waterystone/p/4920797.html)
- [Java-AQS](https://www.cnblogs.com/chengxiao/p/7141160.html)
## Semaphore()
`synchronized` `ReentrantLock` `Semaphore`()
```java
/**
*
* @author Snailclimb
* @date 2018930
* @Description:
*/
public class SemaphoreExample1 {
//
private static final int threadCount = 550;
public static void main(String[] args) throws InterruptedException {
//
ExecutorService threadPool = Executors.newFixedThreadPool(300);
//
final Semaphore semaphore = new Semaphore(20);
for (int i = 0; i < threadCount; i++) {
final int threadnum = i;
threadPool.execute(() -> {// Lambda
try {
semaphore.acquire();// 20/1=20
test(threadnum);
semaphore.release();//
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
});
}
threadPool.shutdown();
System.out.println("finish");
}
public static void test(int threadnum) throws InterruptedException {
Thread.sleep(1000);//
System.out.println("threadnum:" + threadnum);
Thread.sleep(1000);//
}
}
```
`acquire()` `release` `acquire()` `Semaphore` `Semaphore`
```java
semaphore.acquire(5);// 520/5=4
test(threadnum);
semaphore.release(5);// 5
```
`acquire()` `tryAcquire()` false
`Semaphore`
- **** `acquire()` FIFO
- ****
`Semaphore`
```java
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
```
****
[issue645 ](https://github.com/Snailclimb/JavaGuide/issues/645) `Semaphore` `CountDownLatch` AQS state `permits` `permits` Park, state 0 state 0 , `release()` `release()` state 1
`permits`
## CountDownLatch
`CountDownLatch` `count`
`CountDownLatch` , AQS `state` `count` `countDown()` ,`tryReleaseShared` CAS `state`, `state` 0 `await()` `state` 0`await()` `await()` `CountDownLatch` CAS `state == 0` `state == 0` `await()`
### CountDownLatch
**1 n **
`CountDownLatch` n `new CountDownLatch(n)` 1 `countdownlatch.countDown()` 0 `CountDownLatch await()`
**2**
`CountDownLatch` 1 `new CountDownLatch(1)` `coundownlatch.await()` `countDown()` 0
### CountDownLatch
```java
/**
*
* @author SnailClimb
* @date 2018101
* @Description: CountDownLatch
*/
public class CountDownLatchExample1 {
//
private static final int threadCount = 550;
public static void main(String[] args) throws InterruptedException {
//
ExecutorService threadPool = Executors.newFixedThreadPool(300);
final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
final int threadnum = i;
threadPool.execute(() -> {// Lambda
try {
test(threadnum);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
countDownLatch.countDown();//
}
});
}
countDownLatch.await();
threadPool.shutdown();
System.out.println("finish");
}
public static void test(int threadnum) throws InterruptedException {
Thread.sleep(1000);//
System.out.println("threadnum:" + threadnum);
Thread.sleep(1000);//
}
}
```
550 550 `System.out.println("finish");`
`CountDownLatch` `CountDownLatch.await()`
N `CountDownLatch` `CountDownLatch.countDown()` count 1 N count 0 `await()`
`CountDownLatch` `await()` for
```java
for (int i = 0; i < threadCount-1; i++) {
.......
}
```
`count` 0
### CountDownLatch
`CountDownLatch` `CountDownLatch`
### CountDownLatch
- `CountDownLatch`
- `CountDownLatch` `CyclicBarrier`
- `CountDownLatch`
## CyclicBarrier()
`CyclicBarrier` `CountDownLatch` `CountDownLatch` `CountDownLatch`
> `CountDownLatch` AQS `CycliBarrier` `ReentrantLock`(`ReentrantLock` AQS ) `Condition`
`CyclicBarrier` CyclicBarrier
`CyclicBarrier` `CyclicBarrier(int parties)` `await()` `CyclicBarrier`
```java
public CyclicBarrier(int parties) {
this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties {
try {
test(threadNum);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
});
}
threadPool.shutdown();
}
public static void test(int threadnum) throws InterruptedException, BrokenBarrierException {
System.out.println("threadnum:" + threadnum + "is ready");
try {
/**60*/
cyclicBarrier.await(60, TimeUnit.SECONDS);
} catch (Exception e) {
System.out.println("-----CyclicBarrierException------");
}
System.out.println("threadnum:" + threadnum + "is finish");
}
}
```
```
threadnum:0is ready
threadnum:1is ready
threadnum:2is ready
threadnum:3is ready
threadnum:4is ready
threadnum:4is finish
threadnum:0is finish
threadnum:1is finish
threadnum:2is finish
threadnum:3is finish
threadnum:5is ready
threadnum:6is ready
threadnum:7is ready
threadnum:8is ready
threadnum:9is ready
threadnum:9is finish
threadnum:5is finish
threadnum:8is finish
threadnum:7is finish
threadnum:6is finish
......
```
5 `await()`
`CyclicBarrier` `CyclicBarrier(int parties, Runnable barrierAction)` `barrierAction`
```java
/**
*
* @author SnailClimb
* @date 2018101
* @Description: CyclicBarrier Runnable
*/
public class CyclicBarrierExample3 {
//
private static final int threadCount = 550;
//
private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
System.out.println("------------");
});
public static void main(String[] args) throws InterruptedException {
//
ExecutorService threadPool = Executors.newFixedThreadPool(10);
for (int i = 0; i < threadCount; i++) {
final int threadNum = i;
Thread.sleep(1000);
threadPool.execute(() -> {
try {
test(threadNum);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
});
}
threadPool.shutdown();
}
public static void test(int threadnum) throws InterruptedException, BrokenBarrierException {
System.out.println("threadnum:" + threadnum + "is ready");
cyclicBarrier.await();
System.out.println("threadnum:" + threadnum + "is finish");
}
}
```
```
threadnum:0is ready
threadnum:1is ready
threadnum:2is ready
threadnum:3is ready
threadnum:4is ready
------------
threadnum:4is finish
threadnum:0is finish
threadnum:2is finish
threadnum:1is finish
threadnum:3is finish
threadnum:5is ready
threadnum:6is ready
threadnum:7is ready
threadnum:8is ready
threadnum:9is ready
------------
threadnum:9is finish
threadnum:5is finish
threadnum:6is finish
threadnum:8is finish
threadnum:7is finish
......
```
### CyclicBarrier
`CyclicBarrier` `await()` `dowait(false, 0L)` `await()` `parties`
```java
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
```
`dowait(false, 0L)`
```java
// count await count 5
private int count;
/**
* Main barrier code, covering the various policies.
*/
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
//
lock.lock();
try {
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
//
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// cout1
int index = --count;
// count 0 await
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// count parties
//
//
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// We're about to finish waiting even if we had not
// been interrupted, so this interrupt is deemed to
// "belong" to subsequent execution.
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
if (timed && nanos CountDownLatch: A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.(CountDownLatch: )
> CyclicBarrier : A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.(CyclicBarrier : )
`CountDownLatch` N `CyclicBarrier`
`CountDownLatch` `CyclicBarrier`
### ReentrantLock ReentrantReadWriteLock
`ReentrantLock` `synchronized` `ReentrantReadWriteLock`