The preface
AQS can be said to be one of the JAVA source code must read. At the same time, it is also one of the high frequency knowledge points of the Interview. Know and understand it, JAVA junior high school to become a senior engineer essential knowledge. AQS is short for AbstractQueuedSynchronizer, it is also the JUC package under many non-native lock core.
1. AQS Basic overview
AQS is an improved locking mechanism based on CLH queue algorithm. The main logic is that there is a linked Queue inside the AQS. The Queue Node class is an inner class of the AQS, forming a Sync Queue similar to the following (remember the noun)
As you can see, a Node maintains a Thread object and an int waitStatus in addition to the index of the front and back nodes. A Thread object is the Thread object itself in a competing queue. WaitStatus represents the state of the current contention node, which is ignored here.
The node at the Head of the queue that the Head points to is the node that acquired the lock. Release the lock is out of the queue, the subsequent node will become the head of the queue, that is, to obtain the lock
Each ReentrantLock instance has one and only one AQS instance, and each AQS instance maintains a Sync Queue. So, when multiple threads in our business code compete on the same ReentrantLock instance, they are actually enqueuing and unqueuing the same Sync Queue.
Two: AQS call entry —-ReentrantLock
When we use ReentrantLock, the code usually looks like this:
ReentrantLock lock = new ReentrantLock();
Runnable runnable = new Runnable() {
public void run() {
lock.lock();
Sys.out(Thread.currentThread().name() + "Grab the lock."); lock.unlock(); }}; Thread t1 = new Thread(runnable); Thread t2 = new Thread(runnable); t2.start(); t1.start();Copy the codeVisible threads T1 and T2 compete for the lock. Lock. Lock () locks. Unlock () releases the lock
public void lock() {
sync.lock();
}
Copy the codeSync is an abstract class within ReentrantLock that inherits AQS
abstract static class Sync extends AbstractQueuedSynchronizer {
}
Copy the codeThe concrete implementation of the abstract class is two other inner classes, NonFairSync FairSync, which stand for unfair and fair locks, respectively.
Let’s look at the inheritance diagram systematically
The sync instance in ReentrantLock is initialized in the constructor by default,
public ReentrantLock() {
sync = new NonfairSync();
}
Copy the codeLet’s look at the implementation logic of the default NonFairSync.
NonFairSync
When we call reentrantLock. lock(), we call nonfairsync.lock () directly.
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
Copy the codeCompareAndSetState (0, 1) through CAS. Set the int state of AQS to 1. AQS determines whether the current lock has been acquired by the thread based on whether the state is 0. Returns true, then acquiring a lock is successful, the exclusive thread setExclusiveOwnerThreaThread. Set the current lock currentThread ()); Otherwise, acquire(1) tries to acquire D (lock. This method spins and blocks until the lock is successfully acquired. Here, the parameter 1 is passed in to refer to the version field of the optimistic lock, and here, it also records the number of times the reentrant lock has been acquired repeatedly. The lock must be released the same number of times to finally release it.
AQS acquire() logic
AbstractQueuedSynchronizer.acquire()
public final void acquire(int arg) {
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeNote that this method ignores thread interrupts. Nonfairacquire (arg) is a method that AQS has left to its subimplementation class. The implementation of NonFairSync calls sync.nonfairtryacquire () directly.
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;
}
Copy the codeIt can be seen that the method first tries to acquire the initial lock when state = 0. If it fails, it can determine whether the current lock is acquired by the current thread. If so, add the parameters passed in acquire to the state field. In this case, this field is used to record the number of times the lock was acquired repeatedly. Return false on failure to get
AcquireQueued (addWaiter(Node.exclusive), arG) ¶ acquireQueued(addWaiter(Node.exclusive), arg) ¶ Look at the addWaiter
Private Node addWaiter(Node mode) {1: Node Node = new Node(thread.currentThread (), mode); 2: // Try the fast path of enq; Backup to full enq on failure 3: Node pred = tail; 4:if(pred ! = null) {5: node.prev = pred; 6:if(compareAndSetTail(pred, node)) {7: pred.next = node; 8:returnnode; 9:} 10:} 11: enq(node); 12:returnnode; 13:}Copy the codeInitialize a new Node with the passed node.exclusive lock parameter and the current thread instance. Lines 4-10 are entered when there are competing threads in the Sync Queue. Is empty and goes to ENq (node), which is the operation to enqueue the contention thread when the contention is empty. Then return the current contended thread node to Sync Queue as shown in the figure
AcquireQueued is the main node contention lock method. AcquireQueued acquireQueued is the main node contention lock method.
Final Boolean acquireQueued(final Node Node, int arg) {final Boolean failed =true; Try {// The interrupt flag of the current competing thread Boolean interrupted =false; // Spin contention lock, if not contention lock, the thread is not interrupted // the loop is always herefor(;;) Final Node p = node.predecessor(); // If the precursor is the head, try tryAcquire // this logic we have seen before, if the current thread has not acquired the lock // if the state field of AQS is not 0, returnfalse
if(p == head && tryAcquire(arg)) {// If the thread is trying to acquire (arg), the thread is trying to acquire (arg). If the thread is trying to acquire (arg), the thread is trying to acquire (arG)setHead(node);
p.next = null; // helpGC // Indicates that competing locks succeeded failed =false; // This method does not respond to thread interrupts, but does return an interrupt flag if the thread is contending for a lockreturninterrupted; } // If the lock fails to be acquired, go here. The logic used to suspend the current thread is found in the two methods used to determine If //if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if(failed) cancelAcquire(node); }}Copy the codePark means to stop. So the name of this method makes sense: suspend the thread and check the thread’s interrupt status. Note here that the locksupport.part (this) method wakes up automatically when the thread is interrupted
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
Copy the codeThis method passes in the current race node and its precursor
Private static Boolean shouldParkAfterFailedAcquire (Node Mr Pred, Node Node) {/ / precursor nodes wait state. Just keep in mind that we're talking about AQS for // non-shared locks, non-fair locks. So, just make sure that the current race //'s precursor is in SIGNAL state. Int ws = pred.waitStatus;if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but donCaller will need to * retry to make sure it cannot acquire before parking. */ / SIGNAL compareAndSetWaitStatus(pred, ws, node.signal); } return false; }Copy the codeFrom shouldParkAfterFailedAcquire can see out, in ensuring SIGNAL for precursor node status, you can rest assured to unsafe. The park (). SIGNAL means wake up the successor node when the current node is OVER.
So it’s not hard to deduce that our node is now park. Wait for the precursor to release the lock, or wake up with the interrupt, but because this method is interrupt-blind, even if the interrupt is just setting a marker bit, it is still in the loop.
It is assumed that the precursor node acquires the lock and then releases it. The current node is woken up in parkAndCheckInterrupt() and loop for(;) again. This time it is entered on the first if, the current node acquires the lock, resets the node pointed to by Head, etc., and returns the interrupt flag for the current thread.
Return to acquire
public final void acquire(int arg) {
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeThe selfInterrupt() method in if simply resets the interrupt bit of the current thread. This is because the method that retrieves thread interrupt status also resets the status field as well as returning it, so you need to reset the corresponding value after marking.
Let’s take a look at the interface methods for AQS to release locks
ReentrantLock.unlock
public void unlock() {
sync.release(1);
}
Copy the codeAfter go in
ReentrantLock.release
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if(h ! = null && h.waitStatus ! = 0) unparkSuccessor(h);return true;
}
return false;
}
Copy the codeTryRelease is implemented in NonFairLock and, if released successfully, wakes up the Head’s successor if the Head exists and the state is not SIGNAL.
NonFailLock.tryRelease
Let’s look at the implementation of NonFairLock’s tryRelease method
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;
}
Copy the codeIf so, determine whether the current release lock is fully released, because the lock can be acquired repeatedly. If the release lock is fully released, set state to 0, indicating that the lock of AQS has been released.