AQS: full name AbstractQueuedSynchronizer (abstract queue synchronizer), it is an abstract class, main effect is as follows
- A framework for implementing locks is provided based on FIFO queues and state bits
- Exclusive and shared locks are supported
- ConditionObject is internally defined, which is an implementation class of the Condition interface that is essential to using Condition’s related functions
ReentrantLock implements its own locking mechanism based on AQS
ReentrantLock source code parsing
Let’s start by looking at the inheritance relationships between the ReentrantLock classes
Sync, NonfairSync, and FairSync are all internal classes of ReentrantLock
The Sync inner class inherits AQS and overwrites the tryRelease method, while the Sync subclasses NonfairSync (unfair) and FairSync (fair) overwrite the tryAcquire method, respectively
// You can specify fair or unfair locks when creating a ReentrantLock object
// The default constructor creates unfair locks
public ReentrantLock(a) {
sync = new NonfairSync();
}
// Enter true to create a fair lock, false to create an unfair lock
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
Copy the codestatic final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
// This object was cancelled due to timeout or interruption
static final int CANCELLED = 1;
// When a Node joins a queue, the state of the previous Node is set to SIGNAL, indicating that the Node needs to be woken up
static final int SIGNAL = -1;
// This object is currently in the condition wait queue
static final int CONDITION = -2;
// TODO: I don't know what this status bit does
static final int PROPAGATE = -3;
// The status of this object in the synchronization queue. The initial value is 0
volatile int waitStatus;
// The precursor node
volatile Node prev;
// Subsequent nodes
volatile Node next;
// The thread of the node
volatile Thread thread;
// points to the next node in CONDITION
Node nextWaiter;
}
Copy the codeFair lock lock
From the fair lock lock process to explain the source code
final void lock(a) {
acquire(1);
}
Copy the codepublic final void acquire(int arg) {
// There are two judgments here
// tryAcquire: Return true to acquire the lock successfully, false to acquire the lock failed. The method that succeeded in obtaining the lock is returned. You do not need to go down. Otherwise, you need to go to the queue operation
AcquireQueued (addWaiter(node.exclusive), arg) : The thread fails to obtain the lock and joins the queue
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeA hook method in AQS that subclasses must implement or throw an exception
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
Copy the code / / fair lock
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
// Override the template method in AQS
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
// Get the value of state
int c = getState();
// State is 0, indicating that the node is in the initial state
if (c == 0) {
Hasqueuedboomreturns false to indicate that there are no waiting threads in the wait queue
// compareAndSetState The CAS state is successfully changed from 0 to 1. The CAS state is successfully locked, ensuring that only one thread can obtain the lock at the same time
// Enter the team when the scramble fails
if(! hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
// Set an exclusive thread
setExclusiveOwnerThread(current);
return true; }}// ReentrantLock is a ReentrantLock, which means the same thread can be locked repeatedly
// Check whether it is the same thread
else if (current == getExclusiveOwnerThread()) {
// state + 1
int nextc = c + acquires;
// Repeat the lock to reach the upper limit
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
// false indicates that the thread did not acquire the lock
return false; }}}Copy the code// Return true to indicate that there are already waiting queues in the blocking queue
// Return false to indicate that there are no waiting queues in the blocking queue
public final boolean hasQueuedPredecessors(a) {
Node t = tail;
Node h = head;
Node s;
returnh ! = t && ((s = h.next) ==null|| s.thread ! = Thread.currentThread()); }Copy the codeh ! = t &&((s = h.next) == null || s.thread ! = Thread.currentThread());
Whether this judgment is true depends on three conditions:
- The queue is not initialized
- The queue is initialized, but there is only one node in the queue
- The queue has been initialized and there is more than one node in the queue
If the queue is not initialized, tail and head are null, h! If t is not true, return false
Tail and head both point to the same node. H! If t is not true, return false
Third case: there are multiple nodes in the queue. = t (true), h.next is not null (false), s.read! CurrentThread () returns false if the Thread currently acquiring the lock is the same as the Thread in the node, true if it is not
// Head node in AQS queue
private transient volatile Node head;
// Last node in AQS queue
private transient volatile Node tail;
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
// pred Set the value to tail
Node pred = tail;
// The FIFO queue is initialized and the FIFO queue is not initialized
// If pred is null and the queue is not initialized, use the enq(node) method
// If the queue is already initialized, it is equivalent to adding nodes behind the queue
if(pred ! =null) {
node.prev = pred;
// If the CAS fails, enq also has a CAS to the end of the queue, and it is for an infinite loop
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
// This method can be seen in the following figure
private Node enq(final Node node) {
/ / death cycle
for (;;) {
Node t = tail;
// tail is null when initialized, so the first loop goes if first
if (t == null) { // Must initialize
// Set a sentinel node (sentinel node is an empty node)
if (compareAndSetHead(new Node()))
// Both the head and tail nodes point to the sentinel node
tail = head;
// The second loop goes else
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
// Add node after queue
t.next = node;
returnt; }}}}Copy the code
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
// Indicates whether the thread is interrupted while waiting
boolean interrupted = false;
for (;;) {
// Get the precursor node of this node
final Node p = node.predecessor();
// Here are two judgments
// p == head: if the first node is the head node, the current thread is the first one waiting.
// tryAcquire(ARG) : If the current thread is the first thread waiting, try to acquire the lock again, which is equivalent to a spin operation
// If the spin lock succeeds, the current node becomes the head node
// If the current thread is not the first thread waiting, no spin is performed
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
// In theory this is the only exit from the for loop
return interrupted;
}
// The current thread is not the first thread to wait and park blocks
/ / shouldParkAfterFailedAcquire return true will perform parkAndCheckInterrupt this method of the nodes in the park
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// The thread has been interrupted
interrupted = true; }}finally {
if(failed) cancelAcquire(node); }}Copy the codeprivate static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// Get the status of the precursor node
int ws = pred.waitStatus;
// The status bit of the precursor node is already -1, no processing is done
if (ws == Node.SIGNAL)
return true;
// If the status bit of the precursor node is >0, it indicates that the precursor node may be interrupted or timed out. The cancelled node is discarded here
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// Set the state of the precursor node to -1
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
Copy the codeThe main purpose of this method is to ensure that the status of the precursor node of the current node is SIGNAL
P is the precursor node of node. Whether a node needs to be blocked is determined by the status of the precursor node of this node. There are three cases in this method
- The status of the precursor node is
singalIs displayed, indicating that the precursor node is still waiting and the current node must continue to be blocked. Returns true - If the precursor node is greater than 0, it indicates that the precursor node may be interrupted or timed out, and the node in cancelled state should be discarded
- The precursor node is in another state (0,PROPAGATE), which means that the current node needs to retry the attempt to acquire the lock. Returns false, enclosing a for loop that does one more spin lock
private final boolean parkAndCheckInterrupt(a) {
// Block the thread
LockSupport.park(this);
// Checks whether the thread has been interrupted, and returns true if it has
return Thread.interrupted();
}
Copy the codeLocking flow chart
Fair and Unfair Locks Unlock
Interpret the source code from the lock release process
public void unlock(a) {
sync.release(1);
}
Copy the codepublic final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if(h ! =null&& h.waitStatus ! =0)
// Wake up the next node in the wait queue
unparkSuccessor(h);
return true;
}
return false;
}
Copy the codeprotected final boolean tryRelease(int releases) {
// ReentrantLock is reentrant and will be state-1
int c = getState() - releases;
// Check whether the current thread is in an exclusive lock
if(Thread.currentThread() ! = getExclusiveOwnerThread())throw new IllegalMonitorStateException();
boolean free = false;
// a c value of 0 means that no thread holds the lock
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
// If free is true, no thread holds the lock
// If free is false, there are threads holding the lock
return free;
}
Copy the codeprivate void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// Get the successor node of the current node (the current node is the head node)
Node s = node.next;
// If the subsequent node is empty, or cancelled
// where s==null does not mean that S is tail. At this point, a new node may join the queue and the cas is completed but the next pointer is not updated
if (s == null || s.waitStatus > 0) {
s = null;
// Start at tail and walk forward to find the node whose status is not CANCELLED
In addWaiter and enq, joining the queue is not an atomic operation (3 steps: 1 set prev pointer,2 set tail pointer,3 set next // pointer).
// So if you go back and forth, you might miss the node because the next pointer might be null.
for(Node t = tail; t ! =null&& t ! = node; t = t.prev)if (t.waitStatus <= 0)
s = t;
}
// Wake up the blocking node
if(s ! =null)
LockSupport.unpark(s.thread);
}
Copy the codeAn unfair lock
abstract static class Sync extends AbstractQueuedSynchronizer {
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
// The difference with fair locking is that the queue does not determine whether there is a waiting thread, directly lock
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 codeconclusion
For ReentrantLock, the execution logic is as follows:
-
An attempt is made to acquire the lock of an object. If not, unfair locks are treated differently than fair locks
-
The fair lock goes straight to the end of the blocking queue,
-
In an unfair lock, CAS is performed first to obtain the lock. If the lock fails to be obtained, the system enters the end of the blocking queue in the same way as that in a fair lock
-
-
When it does, it determines whether it has been retrieved once before to decide whether to reenter
-
When the lock is released, the state member variable is decrement by 1. If state is not zero after decrement by 1, the release operation is completed. If the state value is 0 after the operation of minus 1, then the unpark method of LockSupport is called to wake up the first subsequent thread in the waiting queue behind the thread and wake it up so that it can acquire the lock of the object (the processing logic of fair lock and unfair lock is the same).
The Debug code
Start two threads, one thread to acquire the lock, you can see the other thread blocking process
public class ReentrantLockTest {
private final Lock reentrantLock = new ReentrantLock(true);
private void method(a) {
try {
reentrantLock.lock();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("method");
} finally{ reentrantLock.unlock(); }}public static void main(String[] args) {
ReentrantLockTest reentrantLockTest = new ReentrantLockTest();
Thread t1 = new Thread(reentrantLockTest::method);
Thread t2 = new Thread(reentrantLockTest::method);
t1.setName("t1");
t2.setName("t2"); t1.start(); t2.start(); }}Copy the code