Table of contents
- 1. Introduction to AQS
-
2. AQS core design
- 2.1 Core components
- 2.2 How AQS works
- 2.3 Key AQS Methods
-
3. The relationship between ReentrantLock and AQS
- 3.1 ReentrantLock's structure
- 3.2 How to use AQS's state
-
4. Analysis of key process of AQS
- 4.1 The acquisition process of exclusive locks
- 4.2 The release process of exclusive locks
-
5. Fair lock and unfair lock
- 5.1 Unfair lock (default)
- 5.2 Fair lock
- 6. Custom implementation: Simplified version ReentrantLock
- 7. Condition implementation principle
- 8. AQS application scenarios
- 9. Summary
1. Introduction to AQS
AbstractQueuedSynchronizer (AQS for short) is one of the most core basic components in Java concurrency package(). It provides a general framework for most synchronization classes in Java (such as ReentrantLock, Semaphore, CountDownLatch, etc.). Understanding how AQS works is essential to gain a deep understanding of Java concurrent programming.
The function of AQS is to solve the implementation problem of synchronizers. It decomposes complex synchronizer implementations into simple framework methods. Developers only need to implement a small number of specific methods to quickly build reliable synchronizers.
2. AQS core design
2.1 Core components
AQS mainly consists of the following parts:
- Synchronous state (state): Use variables of type volatile int to represent the available status of the resource
- FIFO Waiting Queue: A queue implemented using a two-way linked list, used to manage threads waiting to obtain resources
- Exclusive/sharing mode: Supports two modes: exclusive lock (such as ReentrantLock) and shared lock (such as CountDownLatch)
- Conditional variables: Provides a conditional wait/notify mechanism through the ConditionObject class, similar to ()/notify()
2.2 How AQS works
AQS encapsulates some common synchronization operations inside the framework through the template method mode, and handes the characteristics of a specific synchronizer (such as the judgment of whether the resource is available) to subclasses for implementation. AQS provides the following basic operations:
- Resource acquisition: The thread tries to obtain the resource. If it cannot be retrieved, it will be wrapped as Node and added to the waiting queue and blocked
- Resource release: After the thread holding the resource releases the resource, it will wake up the next thread in the waiting queue.
- Thread blocking and wake-up: Implemented through LockSupport's park/unpark mechanism
2.3 Key AQS Methods
AQS defines a set of methods that require subclass implementation:
- tryAcquire(int): Try to obtain resources in exclusive mode
- tryRelease(int): Try to release resources in exclusive mode
- tryAcquireShared(int): Try to obtain resources in shared mode
- tryReleaseShared(int): Try to release resources in shared mode
- isHeldExclusively(): Determine whether the resource is exclusive to the current thread
3. The relationship between ReentrantLock and AQS
ReentrantLock is a reentrant lock implemented based on AQS. It implements the basic functions of locks through the internal class Sync (inherited from AQS), and implements fair locks and non-fair locks through two subclasses, FairSync and NonfairSync.
3.1 ReentrantLock's structure
public class ReentrantLock implements Lock {
private final Sync sync;
abstract static class Sync extends AbstractQueuedSynchronizer {
// Basic operation of implementing lock
}
// Fair lock implementation
static final class FairSync extends Sync { ... }
// Unfair lock implementation
static final class NonfairSync extends Sync { ... }
}3.2 How to use AQS's state
ReentrantLock uses the state field of AQS to indicate the number of locks held:
- state = 0: means that the lock is not held
- state > 0: indicates that the lock is held, and the value indicates the number of reentries
4. Analysis of key process of AQS
4.1 The acquisition process of exclusive locks
When a thread calls the () method, the following process is actually executed:
- First call AQS acquire(1) method:
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(), arg))
selfInterrupt();
}
-
tryAcquire attempts to acquire the lock, which is implemented by ReentrantLock's Sync subclass:
- If state=0, try to set state to 1 using CAS and set the current thread to the thread holding the lock
- If the current thread already holds a lock, add the state value to achieve reentering
- Return false in other cases
-
If tryAcquire fails, addWaiter is called to encapsulate the current thread into Node and add it to the end of the waiting queue:
private Node addWaiter(Node mode) {
Node node = new Node((), mode);
// Try to quickly add to the end of the queue
Node pred = tail;
if (pred != null) {
= pred;
if (compareAndSetTail(pred, node)) {
= node;
return node;
}
}
// Failed to quickly add, enter the complete enqueue method
enq(node);
return node;
}- Then execute the acquireQueued method, allowing the node to continuously try to acquire the lock in the queue until it is successful or interrupted:
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// Get the front-drive node
final Node p = ();
// If the front driver is the head node, it means it is the current node's turn to try to acquire the lock
if (p == head && tryAcquire(arg)) {
// Acquisition successfully, set the current node as the header node
setHead(node);
= null; // help GC
failed = false;
return interrupted;
}
// Determine whether the current thread should be blocked
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}4.2 The release process of exclusive locks
When the thread calls the () method, the following process will be executed:
- First call AQS release(1) method:
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && != 0)
unparkSuccessor(h);
return true;
}
return false;
}
-
tryRelease attempts to release the lock, which is implemented by ReentrantLock's Sync class:
- Check whether the current thread is a thread holding the lock
- Reduce state value
- If state becomes 0, clear the thread holding the lock and return true
-
If tryRelease returns true, indicating that the lock has been completely released, unparkSuccessor is called to wake up the next thread in the waiting queue:
private void unparkSuccessor(Node node) {
// Get the waiting status of the current node
int ws = ;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// Find the next node that needs to be woken up
Node s = ;
if (s == null || > 0) {
s = null;
//Follow the node to wake up from the tail forward
for (Node t = tail; t != null && t != node; t = )
if ( <= 0)
s = t;
}
// Wake up the found node
if (s != null)
();
}5. Fair lock and unfair lock
ReentrantLock supports two modes: fair lock and unfair lock:
5.1 Unfair lock (default)
The tryAcquire implementation of unfair locks:
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = ();
int c = getState();
if (c == 0) {
// Unfair lock directly attempts to acquire the lock without checking the queue
if (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;
}5.2 Fair lock
The tryAcquire implementation of fair lock:
protected final boolean tryAcquire(int acquires) {
final Thread current = ();
int c = getState();
if (c == 0) {
// Fair lock will first call hasQueuedPredecessors to check whether there are predecessor nodes waiting
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;
}The main difference between a fair lock and an unfair lock is whether the waiting queue is considered when acquiring the lock. A fair lock will check whether there are threads queued in the waiting queue, while a fair lock will try to obtain it directly, regardless of the waiting order.
6. Custom implementation: Simplified version ReentrantLock
To understand the AQS principle more deeply, we can implement a simplified version of ReentrantLock:
public class SimpleReentrantLock implements Lock {
private final Sync sync;
/**
* Create an unfair lock by default
*/
public SimpleReentrantLock() {
sync = new NonfairSync();
}
/**
* Create fair locks or non-fair locks based on parameters
*/
public SimpleReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
/**
* Synchronizer implementation inherited from AQS
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* Unfair way to acquire locks
*/
final boolean unfairTryAcquire(int acquires) {
// Get the current thread
final Thread current = ();
// Get the current state
int c = getState();
// state is 0 means that the lock is not held
if (c == 0) {
// Use CAS to try to set state from 0 to 1
if (compareAndSetState(0, acquires)) {
// Acquisition of the lock successfully, set the thread currently holding the lock to the current thread
setExclusiveOwnerThread(current);
return true;
}
}
// If the current thread is a thread holding the lock, it can be reentered
else if (current == getExclusiveOwnerThread()) {
// Increase the state value to achieve reentry count
int nextC = c + acquires;
// Check for overflow
if (nextC < 0) {
throw new Error("Maximum lock count exceeded");
}
// Set the new state value, CAS is not needed here because the current thread already holds the lock
setState(nextC);
return true;
}
// Acquisition of lock failed
return false;
}
/**
* Release the lock
*/
@Override
protected final boolean tryRelease(int releases) {
// Check whether the current thread is a thread holding the lock
if (() != getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
}
// Reduce the state value
int c = getState() - releases;
// Determine whether the lock is fully released
boolean free = (c == 0);
if (free) {
// Completely release the lock and clear the thread holding the lock
setExclusiveOwnerThread(null);
}
// Update state value
setState(c);
return free;
}
/**
* Determine whether the current thread holds a lock
*/
@Override
protected boolean isHeldExclusively() {
return getExclusiveOwnerThread() == ();
}
/**
* Create condition variables
*/
Condition newCondition() {
return new ConditionObject();
}
/**
* Acquire the lock's holding count
*/
public int getHoldCount() {
return isHeldExclusively() ? getState() : 0;
}
}
/**
* Implementation of fair locks
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
@Override
protected boolean tryAcquire(int acquires) {
final Thread current = ();
int c = getState();
if (c == 0) {
// Fairness manifestation: first check whether there are predecessor nodes in the queue waiting
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;
}
}
/**
* Implementation of unfair locks
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* The acquisition implementation of unfair locks
*/
@Override
protected boolean tryAcquire(int acquires) {
return unfairTryAcquire(acquires);
}
}
// Methods to implement Lock interface
@Override
public void lock() {
(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
(1);
}
@Override
public boolean tryLock() {
return (1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return (1, (time));
}
@Override
public void unlock() {
(1);
}
@Override
public Condition newCondition() {
return ();
}
/**
* Check whether the current lock is held by a thread
*/
public boolean isLocked() {
return ();
}
/**
* Check whether the current thread holds a lock
*/
public boolean isHeldByCurrentThread() {
return ();
}
/**
* Get the current lock holding count
*/
public int getHoldCount() {
return ();
}
}7. Condition implementation principle
AQS provides a ConditionObject internal class to implement the Condition interface and supports a conditional wait/notify mechanism similar to wait/notify:
- Conditional Queue: Each Condition maintains a separate condition queue, independent of the AQS synchronization queue
- await operation: Add the current thread to the condition queue and release the held lock
- signal operation: Transfer the threads in the conditional queue to the synchronization queue and wait for the lock to be reacquisitioned
public final void await() throws InterruptedException {
if (())
throw new InterruptedException();
// Add to conditional queue
Node node = addConditionWaiter();
// Completely release the lock
int savedState = fullyRelease(node);
int interruptMode = 0;
//Cyclically check whether the node has been transferred to the synchronization queue
while (!isOnSyncQueue(node)) {
// Block the current thread
(this);
// Check interrupt
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// Recompetition lock
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if ( != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}8. AQS application scenarios
AQS is widely used in various synchronizers in Java concurrency packages:
- ReentrantLock: Reentrant exclusive lock
- Semaphore: Semaphore, controlling the number of threads that access specific resources at the same time
- CountDownLatch: Blocking, allowing one or more threads to wait for a set of operations to complete
- ReentrantReadWriteLock: Read and write lock, allowing multiple threads to read at the same time, but only one thread can write
- CyclicBarrier: Loop fence, allowing a group of threads to wait for each other to reach a common point
9. Summary
AQS is the most core basic component in the Java concurrency framework. It achieves efficient thread synchronization through the following mechanisms:
- Status Management: Use volatile variables and CAS operations to ensure thread safety
- Queue Management: Use CLH queues to efficiently manage waiting threads
- Blocking primitive: Use LockSupport to achieve thread blocking and wake-up
- Template method pattern: Separate general logic from specific logic to improve scalability
Understanding how AQS works not only helps to better use synchronizers in Java concurrent packages, but also helps us implement our own efficient synchronizers when necessary. AQS breaks down complex synchronizer problems into small amounts of basic methods through simple design, allowing developers to quickly implement various synchronizers. ReentrantLock provides more functions than synchronized, such as interruptibility, timeout waiting, fair selection, etc.