1. Class definition
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
implements java.io.Serializable
Copy the codeYou can see this in the class definition
- AbstractQueuedSynchronizer is an abstract class, using the template method design pattern (implements most of the functionality, few to subclass implementation)
- AbstractQueuedSynchronizer inherited AbstractOwnableSynchronizer
- AbstractOwnableSynchronizer implements the Java. IO. The Serializable interface, said AbstractQueuedSynchronizer support serialization
2. Field attributes
// Serialize the version number
private static final long serialVersionUID = 7373984972572414691L;
// Wait for the queue head to delay initialization. SetHead is used only during initialization. Note: If the head exists, its wait state is guaranteed not to be cancelled
private transient volatile Node head;
// Wait for the end of the queue, delay initialization
private transient volatile Node tail;
0(default initial value) indicates no lock, 1 indicates a lock, and can be greater than 1 to indicate a reentrant lock. Some initial values are -1
//state uses the volatile modifier
private volatile int state;
Copy the codeThis can be seen from the field properties
- AbstractQueuedSynchronizer internal maintains a linked list queue
- AbstractQueuedSynchronizer storage state of the state of the lock
3. Construction method
// Default empty constructor, no operation
protected AbstractQueuedSynchronizer(a) {}Copy the codeYou can see from the constructor
- AbstractQueuedSynchronizer constructor use protected decorate, is an empty constructor, need a subclass implementation
Method 4.
GetState method
// Get the current synchronization status
protected final int getState(a) {
return state;
}
Copy the codeSetState method
// Set the current synchronization status
protected final void setState(int newState) {
state = newState;
}
Copy the codeCompareAndSetState method
// Use CAS to set the synchronization status
protected final boolean compareAndSetState(int expect, int update) {
// Using Unsafe's compareAndSwapInt method to update synchronized state, which is a native method
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
Copy the codeEnq method
// Add nodes to the end of the queue
private Node enq(final Node node) {
//for infinite loop, retry CAS
for (;;) {
// If the queue ends
Node t = tail;
if (t == null) { // Must initialize
// If the tail node is null, initialize it
//CAS sets a newly created node to be the head node. Other lines may be created first
if (compareAndSetHead(new Node()))
// Create the head node successfully, put the tail node to the head node
tail = head;
} else {
// If the tail node is not null
// add a reference to the parent node
node.prev = t;
//CAS sets the node as the tail node. It is possible that other nodes are also setting the tail node
if (compareAndSetTail(t, node)) {
// If added successfully
// Place the reference of the back drive of the precursor node to the added node
t.next = node;
// Return the precursor node added to the node (possibly the previous tail node, which may be modified by other threads)
returnt; }}}}Copy the codeCompareAndSetHead method
// Set the head node, called only in the enq method
private final boolean compareAndSetHead(Node update) {
// If the head node is null, the passed node is set as the head node
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
Copy the codeCompareAndSetTail method
// Set the tail node, called only in the enq method
private final boolean compareAndSetTail(Node expect, Node update) {
// Set the incoming node as the trailing node only if the trailing node is Expect
return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}
Copy the codeAddWaiter method
// Add nodes to the end of the queue
private Node addWaiter(Node mode) {
// Use the current thread and the incoming node to create a new node, followed by the newly created node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
// Get a copy of the tail node
Node pred = tail;
if(pred ! =null) {
// If the tail node is not null
// add node's precursor node reference to the tail node
node.prev = pred;
//CAS sets the node as the tail node. It is possible that other nodes are also setting the tail node
if (compareAndSetTail(pred, node)) {
// place the reference of the drive node to the added node
pred.next = node;
// Return the precursor node added to the node (possibly the previous tail node, which may be modified by other threads)
returnnode; }}// If the tail node is null, or is added successfully by another line first
// Add using the enq method
enq(node);
return node;
}
Copy the codeSetHead method
// Set the head node, called only by the acquire method
private void setHead(Node node) {
// The head node points to the passed node
head = node;
// Set thread to null for the incoming node
node.thread = null;
// Set the prev of the incoming node to null
node.prev = null;
}
Copy the codeUnparkSuccessor method
// If a node exits, wake up the thread next to the node that has not been canceled
// Note that the node argument is the head node
private void unparkSuccessor(Node node) {
// Get the waitStatus of the head node, waitStatus greater than 0 is cancelled
int ws = node.waitStatus;
if (ws < 0)
// If the waitStatus of the head node is less than 0, set the waitStatus of the head node to 0
compareAndSetWaitStatus(node, ws, 0);
// Get the back-end node of the header
Node s = node.next;
if (s == null || s.waitStatus > 0) {
// If the backdrive is nuL or the waitStatus of the backdrive is greater than 0 (cancelled)
s = null;
// Use the for loop to traverse from the last node forward
// Find the first node whose waitStatus is less than or equal to 0 (not cancelled) in the traversal
// Note that it is the first in order, not the first in reverse order, which will loop all the way to the end node
for(Node t = tail; t ! =null&& t ! = node; t = t.prev)if (t.waitStatus <= 0)
s = t;
}
if(s ! =null)
// If the node is not null, wake up the thread of the node
LockSupport.unpark(s.thread);
}
Copy the codeDoReleaseShared method
// Signal release in shared mode
private void doReleaseShared(a) {
/* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. */
//for an infinite loop
for (;;) {
// Get a copy of the head node
Node h = head;
if(h ! =null&& h ! = tail) {// If h is not null and h is not equal to the tail node (there are elements in the queue)
// Get the waitStatus of node H
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
// If h is in the -1 state
// Set waitStatus of h to 0
if(! compareAndSetWaitStatus(h, Node.SIGNAL,0))
// Repeat the next loop if it fails.
continue; // loop to recheck cases
// If the setting succeeds, wake up the back drive node of h, and the state of the head node may change
unparkSuccessor(h);
}
// If the status of node H is 0 and the CAS fails to set waitStatus of node -3, the next loop is performed
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
// The loop is interrupted if the h node is the head node, and continues if the head changes
break; }}Copy the codeSetHeadAndPropagate method
// Set the queue header and propagation mode
private void setHeadAndPropagate(Node node, int propagate) {
// Get a copy of the head node
Node h = head; // Record old head for check below
// Set the incoming node as the head node
setHead(node);
/* * Try to signal next queued node if: * Propagation was indicated by caller, * or was recorded (as h.waitStatus either before * or after setHead) by a previous operation * (note: this uses sign-check of waitStatus because * PROPAGATE status may transition to SIGNAL.) * and * The next node is waiting in shared mode, * or we don't know, because it appears null * * The conservatism in both of these checks may cause * unnecessary wake-ups, but only when there are multiple * racing acquires/releases, so most need signals now or soon * anyway. */
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// If the propagation mode passed in is greater than 0
// If the head node is null
// If the waitStatus of the head node is less than 0
// If node is null
// If node waitStatus is less than 0
// Get the back-end node
Node s = node.next;
if (s == null || s.isShared())
// If the rear-drive is null or the rear-drive is in shared mode
// Release signals in shared modedoReleaseShared(); }}Copy the codeCancelAcquire method
// Cancel the node's attempt to acquire the lock
private void cancelAcquire(Node node) {
if (node == null)
// If node is null, return
return;
// Set the node thread to null
node.thread = null;
// Get the previous node of node
Node pred = node.prev;
// While loop, if the waitStatus of the precursor is greater than 0 (cancelled), keep looking up
// Find the first uncanceled node
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
// Get the back-end node of the front-end node
Node predNext = pred.next;
// Set waitStatus of node to unset
node.waitStatus = Node.CANCELLED;
// If the current node is the tail node, remove itself and set the precursor node to the tail node
if (node == tail && compareAndSetTail(node, pred)) {
// Set the back-end node of the front node to NULL
compareAndSetNext(pred, predNext, null);
} else {
// Node is not a tail node
//If successor needs signal, try to set pred's next-link
//so it will get one. Otherwise wake it up to propagate.
int ws;
if(pred ! = head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <=0&& compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread ! =null) {
Node next = node.next;
if(next ! =null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
// Wake up the precursor
unparkSuccessor(node);
}
node.next = node; // help GC}}Copy the codeShouldParkAfterFailedAcquire method
// The current thread did not acquire the lock, whether to suspend the thread
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// Get the waitStatus of the precursor
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
// If waitStatus of the precursor node is -1, it indicates that the precursor node is healthy and can safely suspend the current thread
return true;
if (ws > 0) {
// If the waitStatus of the precursor is greater than 0, the precursor is unqueued
// Note: threads that enter the blocking queue are suspended, while the wake up operation is done by the precursor node
// Keep looking until you find a node whose waitStatus is greater than or equal to 0
// The thread of node is suspended and then woken up by the found precursor node
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// The waitStatus of the precursor node is not equal to 1 or -1, but can only be 0, -2, -3
// Set the waitStatus of the precursor node to -1
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
// Return false to indicate that no suspension is required
return false;
}
Copy the codeSelfInterrupt method
// Interrupt the current thread
static void selfInterrupt(a) {
Thread.currentThread().interrupt();
}
Copy the codeParkAndCheckInterrupt method
// Suspend the current thread and check for interruption
private final boolean parkAndCheckInterrupt(a) {
// Suspend the current thread
LockSupport.park(this);
// Check whether the current thread is interrupted
return Thread.interrupted();
}
Copy the codeAcquireQueued method
// Try to acquire the lock, fail to interrupt and wait for the precursor node to wake up
final boolean acquireQueued(final Node node, int arg) {
// Fails by default
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// Get the node precursor
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
// If p is the head node and the lock was acquired successfully
// Set node as the head node
setHead(node);
// Set the back-end node of the front node to null
p.next = null; // help GC
// Set to a success flag
failed = false;
// Return to the interrupted flag
return interrupted;
}
// Find a precursor node that can wake you up, then interrupt it and wait for it to wake up
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// If node needs to be suspended
// Interrupted sets the flag to true
interrupted = true; }}finally {
if (failed)
// If this fails, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeDoAcquireInterruptibly method
// Acquire the exclusive lock. If the lock fails, suspend the thread and throw an exception
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
// Add the exclusive mode node to the end of the queue
final Node node = addWaiter(Node.EXCLUSIVE);
// Set the failure flag to true
boolean failed = true;
try {
for (;;) {
// Get the precursor node of node
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
// If the node's precursor is a header and the lock is acquired successfully
// Set node to the head node
setHead(node);
// Set the backdrive node reference of the front node to NULL
p.next = null; // help GC
// Set the failure flag to false
failed = false;
// Return directly
return;
}
// If no lock is obtained, the thread is suspended
// If the thread fails to suspend, an exception is thrown
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw newInterruptedException(); }}finally {
if (failed)
// Failed, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeDoAcquireNanos method
// Get the exclusive lock and specify the timeout period
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
// Check parameters
if (nanosTimeout <= 0L)
return false;
// Calculate the expiration time
final long deadline = System.nanoTime() + nanosTimeout;
// Add the exclusive mode node to the end of the queue
final Node node = addWaiter(Node.EXCLUSIVE);
// Set the failure flag to true
boolean failed = true;
try {
//for an infinite loop
for (;;) {
// Get the precursor node of node
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
// If the node's precursor is a header and the lock is acquired successfully
// Set node to the head node
setHead(node);
// Set the backdrive node reference of the front node to NULL
p.next = null; // help GC
// Set the failure flag to false
failed = false;
// Lock successfully acquired, returns true
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
// Timeout, return false
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
// If no lock is obtained, the thread is suspended
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
// If the thread is interrupted, throw an exception
throw newInterruptedException(); }}finally {
if (failed)
// Failed, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeDoAcquireShared method
// Get the shared lock
private void doAcquireShared(int arg) {
// Add the shared mode node to the end of the queue
final Node node = addWaiter(Node.SHARED);
// Set the failure flag to true
boolean failed = true;
try {
// Set the interrupt flag to true
boolean interrupted = false;
//for an infinite loop
for (;;) {
// Get the precursor node of node
final Node p = node.predecessor();
if (p == head) {
// If the precursor node is a header
// Try to obtain the shared lock
int r = tryAcquireShared(arg);
if (r >= 0) {
// Get success, set the header to node, and continue propagation
setHeadAndPropagate(node, r);
// Set the backdrive node reference of the front node to NULL
p.next = null; // help GC
if (interrupted)
// If the interrupt flag is true, the current thread is interrupted
selfInterrupt();
// Set the failure flag to false
failed = false;
// Return directly
return; }}// If no lock is obtained, the thread is suspended
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
// Suspend thread successfully, set interrupt flag to true
interrupted = true; }}finally {
if (failed)
// Failed, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeDoAcquireSharedInterruptibly method
// Get the shared lock
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// Add the shared mode node to the end of the queue
final Node node = addWaiter(Node.SHARED);
// Set the failure flag to true
boolean failed = true;
try {
//for an infinite loop
for (;;) {
// Get the precursor node of node
final Node p = node.predecessor();
if (p == head) {
// If the precursor node is a header
// Try to obtain the shared lock
int r = tryAcquireShared(arg);
if (r >= 0) {
// Get success, set the header to node, and continue propagation
setHeadAndPropagate(node, r);
// Set the backdrive node reference of the front node to NULL
p.next = null; // help GC
// Set the failure flag to false
failed = false;
// Return directly
return; }}// If no lock is obtained, the thread is suspended
// If the thread fails to suspend, an exception is thrown
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw newInterruptedException(); }}finally {
if (failed)
// Failed, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeDoAcquireSharedNanos method
// Obtain the shared lock and specify the timeout period
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
// Check parameters
if (nanosTimeout <= 0L)
return false;
// Calculate the expiration time
final long deadline = System.nanoTime() + nanosTimeout;
// Add the shared mode node to the end of the queue
final Node node = addWaiter(Node.SHARED);
// Set the failure flag to true
boolean failed = true;
try {
//for an infinite loop
for (;;) {
// Get the precursor node of node
final Node p = node.predecessor();
if (p == head) {
// If the precursor node is a header
// Try to obtain the shared lock
int r = tryAcquireShared(arg);
if (r >= 0) {
// Get success, set the header to node, and continue propagation
setHeadAndPropagate(node, r);
// Set the backdrive node reference of the front node to NULL
p.next = null; // help GC
// Set the failure flag to false
failed = false;
// Return directly
return true;
}
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
// Timeout, return false
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
// If no lock is obtained, the thread is suspended
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
// If the thread is interrupted, throw an exception
throw newInterruptedException(); }}finally {
if (failed)
// Failed, cancel node's attempt to acquire the lockcancelAcquire(node); }}Copy the codeTryAcquire method
// Try to obtain an exclusive lock and force the subclass to implement it
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
Copy the codeTryRelease method
// Try to release the lock and force the subclass to implement it
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
Copy the codeTryAcquireShared method
// Try to obtain the shared lock and force the subclass to implement it
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
Copy the codeTryReleaseShared method
// Try to release the shared lock and force the subclass to implement it
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
Copy the codeIsHeldExclusively method
// Return the lock mode, true exclusive lock
protected boolean isHeldExclusively(a) {
throw new UnsupportedOperationException();
}
Copy the codeAcquire method
// Get the lock in exclusive mode
public final void acquire(int arg) {
// First try to acquire the lock, if it fails, add it to the end of the queue in exclusive mode, interrupt the current thread and wait to be woken up
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeAcquireInterruptibly method
If the lock fails to be acquired in exclusive mode, an exception is thrown
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if(! tryAcquire(arg)) doAcquireInterruptibly(arg); }Copy the codeTryAcquireNanos method
// Lock in exclusive mode and set timeout
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
Copy the codeRelease method
// Release the exclusive lock
public final boolean release(int arg) {
if (tryRelease(arg)) {
Succeeded in releasing the exclusive lock
// Copy a copy of the head node
Node h = head;
if(h ! =null&& h.waitStatus ! =0)
// If the head node is not null and the status of the head node is normal
// Wake up the next node
unparkSuccessor(h);
return true;
}
return false;
}
Copy the codeAcquireShared method
// Get lock in shared mode
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
Copy the codeAcquireSharedInterruptibly method
If the lock is not obtained in shared mode, an exception is thrown
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
Copy the codeTryAcquireSharedNanos method
// In shared mode, lock is acquired and timeout is set
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}
Copy the codeReleaseShared method
// Release the shared lock
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
Copy the codeHasQueuedThreads method
// Whether the queue contains threads
public final boolean hasQueuedThreads(a) {
returnhead ! = tail; }Copy the codeHasContended method
// Whether any thread is currently using the lock
public final boolean hasContended(a) {
returnhead ! =null;
}
Copy the codeGetFirstQueuedThread method
// Get the first thread in the queue (wait the longest thread)
public final Thread getFirstQueuedThread(a) {
// handle only fast path, else relay
// Return null if it is null
// Call fullGetFirstQueuedThread to return the first thread in the queue
return (head == tail) ? null : fullGetFirstQueuedThread();
}
Copy the codeFullGetFirstQueuedThread method
// Get the first thread in the queue
private Thread fullGetFirstQueuedThread(a) {
// In general, the node behind the head is the first element in the queue
// If head is not null, and the backdrive of head is not null, and the front drive of head is head
// Return directly to the back-end node of head
Node h, s;
Thread st;
if(((h = head) ! =null&& (s = h.next) ! =null&& s.prev == head && (st = s.thread) ! =null) || ((h = head) ! =null&& (s = h.next) ! =null&& s.prev == head && (st = s.thread) ! =null))
return st;
// May not have come and set the head of the rear drive node (multiple thread operation)
// In this case, go from tail to head
Node t = tail;
Thread firstThread = null;
while(t ! =null&& t ! = head) { Thread tt = t.thread;if(tt ! =null)
firstThread = tt;
t = t.prev;
}
return firstThread;
}
Copy the codeIsQueued method
// Check whether the given thread is queued
public final boolean isQueued(Thread thread) {
if (thread == null)
throw new NullPointerException();
// For loop through, from tail forward
for(Node p = tail; p ! =null; p = p.prev)
if (p.thread == thread)
return true;
return false;
}
Copy the codeGetQueueLength method
// Get the length of the queue
public final int getQueueLength(a) {
int n = 0;
for(Node p = tail; p ! =null; p = p.prev) {
if(p.thread ! =null)
++n;
}
return n;
}
Copy the codeGetQueuedThreads method
// Get all threads in the queue and return them as a collection
public final Collection<Thread> getQueuedThreads(a) {
ArrayList<Thread> list = new ArrayList<Thread>();
for(Node p = tail; p ! =null; p = p.prev) {
Thread t = p.thread;
if(t ! =null)
list.add(t);
}
return list;
}
Copy the codeGetExclusiveQueuedThreads method
// Get all the threads in the queue in exclusive mode as a collection
public final Collection<Thread> getExclusiveQueuedThreads(a) {
ArrayList<Thread> list = new ArrayList<Thread>();
for(Node p = tail; p ! =null; p = p.prev) {
if(! p.isShared()) { Thread t = p.thread;if(t ! =null) list.add(t); }}return list;
}
Copy the codeGetSharedQueuedThreads method
// Get all the threads in the queue in the shared mode as a collection
public final Collection<Thread> getSharedQueuedThreads(a) {
ArrayList<Thread> list = new ArrayList<Thread>();
for(Node p = tail; p ! =null; p = p.prev) {
if (p.isShared()) {
Thread t = p.thread;
if(t ! =null) list.add(t); }}return list;
}
Copy the code