This is my fifth day of the August Challenge
Phase to recommend
- Learn more about Java exceptions
- Java Basics: The Object class
- Do you understand these frequent meeting questions of the Object class?
- Spring common API: Spring class and related surface pilot
- Java concurrent programming: Threads
- Thread Pools
- Java Concurrency – details of common blocking queues
- In August in Java lock | more challenges
I. Introduction to AQS
AQS: AbstractQueuedSynchronizer, namely abstract queue synchronizer, is by maintaining a Shared resource status (Volatile Int State) and a first in first out (FIFO) threads waiting queue to implement a multithreaded synchronization framework of access to a Shared resource.
Building a synchronizer based on AQS:
- ReentrantLock
- Semaphore
- CountDownLatch
- ReentrantReadWriteLock
- SynchronusQueue
- FutureTask
Advantage:
- AQS solves many of the details involved in implementing a synchronizer, such as customizing standard synchronization status, FIFO synchronization queues.
- Building a synchronizer based on AQS offers many benefits. Not only does it greatly reduce the implementation effort, but it also doesn’t have to deal with the problem of competing in multiple locations.
2. Core knowledge of AQS
2.1 Core ideas of AQS
AQS (AbstractQueuedSynchronizer) core idea is: if the requested free sharing of resources, will thread is set to the current request resources effective worker threads, and set the Shared resource is locked. If the requested shared resource is occupied, then you need a mechanism for threads to block, wait, and allocate locks when they wake up. This mechanism AQS is implemented with CLH queue locking, which queues threads that are temporarily unable to acquire locks.
CLH(Craig,Landin,and Hagersten) queue is a virtual bidirectional queue (the virtual bidirectional queue does not exist queue instances, only the association between nodes). AQS encapsulates each thread requesting shared resources into a Node Node of a CLH lock queue to realize lock allocation.
AQS uses an int member variable to represent the synchronization state, and queues resource threads through the built-in FIFO queue. AQS uses CAS to perform atomic operations on the synchronization state to modify its value.
private volatile int state;// Share variables, using volatile to ensure thread visibility
Copy the codeState information is operated with getState, setState, compareAndSetState of procTED types
// Return the current value of the synchronization status
protected final int getState() {
return state;
}
// Set the value of synchronization status
protected final void setState(int newState) {
state = newState;
}
// atomically (CAS operation) set the synchronization state value to the given value update if the current synchronization state value equals expect(expected value)
protected final boolean compareAndSetState(int expect, int update) {
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
Copy the code2.2 Data structure of AQS
CLH queue figure
- Sync Queue: It is a two-way list. This includes the head node and the tail node. The head node is used for subsequent scheduling.
- Condition queue: Non-mandatory, unidirectional list. This list exists only when cindition exists in the program.
2.3 The template method pattern is used at the bottom of AQS
The design of the synchronizer is based on the template method pattern, that is, the consumer inherits the synchronizer and overwrites the specified methods, then combines the synchronizer in the implementation of a custom synchronization component and calls the template methods provided by the synchronizer, which in turn call the methods overridden by the consumer.
The methods that the synchronizer can override are shown in the following table.
| Method names | describe |
|---|---|
protected boolean tryAcquire(int arg) |
To obtain the synchronization status exclusively, you need to query the current status and determine whether the synchronization status meets expectations, and then set the synchronization status on the CAS server |
protected boolean tryRelease(int arg) |
An exclusive release of the synchronization state gives the waiting thread a chance to acquire the synchronization state |
protected int tryAcquireShared(int arg) |
If the value greater than or equal to 0 is returned, the synchronization status is obtained successfully. Otherwise, the synchronization status fails |
protected boolean tryReleaseShared(int arg) |
The synchronization state is released in shared mode |
protected boolean isHeldExclusively() |
Whether the current synchronizer is occupied by a thread in exclusive mode |
By default, throw an UnsupportedOperationException in every way. The implementation of these methods must be internally thread-safe and should generally be short rather than blocking. All other methods in the AQS class are final, so they cannot be used by other classes. Only these methods can be used by other classes.
The template methods provided by the synchronizer are divided into three types: exclusive obtaining and releasing the synchronization state, shared obtaining and releasing the synchronization state, and querying the status of waiting threads in the synchronization queue
2.4 AQS Resource sharing methods
AQS defines two resource sharing methods
-
Exclusive: Only one thread can execute, such as ReentrantLock. Can be divided into fair lock and non-fair lock:
- Fair lock: The first to arrive in the queue is the first to get the lock
- Unfair lock: When a thread wants to acquire a lock, it grabs the lock regardless of the queue order
-
Share: Multiple threads can latch simultaneously, such as Semaphore/CountDownLatch. Semaphore, CountDownLatCh, CyclicBarrier, and ReadWriteLock will be covered later.
ReentrantReadWriteLock can be considered as a combination, because the ReentrantReadWriteLock is a read/write lock that allows multiple threads to read/write a resource at the same time.
Different custom synchronizers compete for shared resources in different ways. Custom synchronizer in the implementation only need to achieve the shared resource state acquisition and release mode, as for the maintenance of the specific thread waiting queue (such as access to resources failed to join the queue/wake up the queue, etc.), AQS has helped us in the upper layer has been implemented.
2.5 AQS class structure
2.5.1 Inheritance Relationship of classes
AbstractQueuedSynchronizer inherited from AbstractOwnableSynchronizer abstract class, and implement the Serializable interface, can be serialized.
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable
Copy the codeThe AbstractOwnableSynchronizer abstract class source code is as follows:
public abstract class AbstractOwnableSynchronizer implements java.io.Serializable {
// Version serial number
private static final long serialVersionUID = 3737899427754241961L;
// The constructor
protected AbstractOwnableSynchronizer(){}// Thread in exclusive mode
private transient Thread exclusiveOwnerThread;
// Set the exclusive thread
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
// Get the exclusive thread
protected final Thread getExclusiveOwnerThread() {
returnexclusiveOwnerThread; }} AbstractOwnableSynchronizer abstract class, you can set the thread thread and get exclusive monopoly resources resources. The setExclusiveOwnerThread and getExclusiveOwnerThread methods are called by subclasses.Copy the code2.5.2 Inner class of class – Node class
static final class Node {
// Mode, divided into shared and exclusive
// Share mode
static final Node SHARED = new Node();
// Exclusive mode
static final Node EXCLUSIVE = null;
// Node state
CANCELLED: the value is 1, indicating that the current thread is CANCELLED
// SIGNAL, with a value of -1, indicates that the current node's successor contains threads that need to run, i.e., unpark
// CONDITION, with a value of -2, indicates that the current node is waiting for the CONDITION, that is, in the CONDITION queue
PROPAGATE, a value of -3 indicates that the subsequent acquireShared can be executed in the current scenario
// A value of 0 indicates that the current node is in the sync queue, waiting to acquire the lock
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
// Node state
volatile int waitStatus;
// Front node
volatile Node prev;
// The successor node
volatile Node next;
// The thread of the node
volatile Thread thread;
// Next to wait
Node nextWaiter;
// Whether the node is waiting in shared mode
final boolean isShared() {
return nextWaiter == SHARED;
}
// Get the precursor and throw an exception if it is null
final Node predecessor() throws NullPointerException {
// Save the predrive
Node p = prev;
if (p == null) // If the precursor is null, throw an exception
throw new NullPointerException();
else // If the precursor is not null, return
return p;
}
// The constructor has no parameters
Node() { // Used to establish initial head or SHARED marker
}
// The constructor
Node(Thread thread, Node mode) { // Used by addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
// The constructor
Node(Thread thread, int waitStatus) { // Used by Condition
this.waitStatus = waitStatus;
this.thread = thread; }}Copy the codeEach blocked thread is queued as a Node. Each node contains a reference to the Thread type, and each node has a state, which is described below.
- A value of 0 indicates that the current node is in the Sync queue, waiting for the lock.
CANCELLEDIf the value is 1, the current thread is cancelled.SIGNAL, the value is -1, indicating that the thread contained by the successor node of the current node needs to run and the unpark operation needs to be performed.CONDITION, is -2, indicating that the current node is waiting on the condition, that is, in the condition queue.PROPAGATEIf the value is -3, the subsequent acquireShared can be executed in the current scenario.
2.5.3 Attributes of the class
Attribute contains a head node head, tail, tail nodes spinForTimeoutThreshold state state, the spin time, offset and AbstractQueuedSynchronizer abstract attributes in ram, by the offset, You can get and set the value of this property, along with a static initializer block for loading the memory offset address.
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
implements java.io.Serializable {
/ / version number
private static final long serialVersionUID = 7373984972572414691L;
/ / head node
private transient volatile Node head;
/ / end nodes
private transient volatile Node tail;
/ / state
private volatile int state;
// Spin time
static final long spinForTimeoutThreshold = 1000L;
// Unsafe class instance
private static final Unsafe unsafe = Unsafe.getUnsafe();
// state Indicates the memory offset address
private static final long stateOffset;
// head Memory offset address
private static final long headOffset;
// state Indicates the memory offset address
private static final long tailOffset;
// tail Memory offset address
private static final long waitStatusOffset;
// next Memory offset address
private static final long nextOffset;
// Static initializer block
static {
try {
stateOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
waitStatusOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("waitStatus"));
nextOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("next"));
} catch (Exception ex) { throw new Error(ex); }}}Copy the codeIii. Exclusive mode and shared mode
3.1 AQS Exclusive mode
In exclusive mode, only one thread is allowed to obtain the synchronization state. When this thread has not released the synchronization state, other threads cannot obtain the synchronization state and can only join the synchronization queue and wait.
Exclusive mode has two functions
- 1. Obtain the function of synchronizing resources. When multiple threads compete for a synchronized resource, only one thread can obtain the synchronized resource, and the other threads that have not obtained the synchronized resource must wait at the current location.
- 2. Release synchronization resources. When the thread that acquired the synchronized resource runs out of it, it releases the synchronized resource and wakes up one or more threads that are waiting to synchronize the resource.
3.1.1 Acquire Resources in Exclusive Mode (Acquire Method)
In exclusive mode, acquire the right to use resources mainly through acquire() method, acquire() method code is as follows:
/** * this method does not respond to interrupts */
public final void acquire(int arg) {
TryAcquire () is called to try to acquire the synchronized resource
// If tryAcquire () returns true, then acquore () is finished
//tryAcquire () method is generally implemented in AQS self-mine
// if tryAcquire () returns false
// Enqueue the thread currently calling acquire () by calling the addWaiter() method
//3. Call the acquireQueued() method to fetch the synchronized resource in the wait queue
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeIts execution process can be divided into the following steps:
1. Execute the tryAcquire() method
Call the tryAcquire() method to try to obtain access to the synchronized resource. The tryAcquire() method returns true on success, and the entire if condition returns false, so acquire() is finished. If tryAcquire() returns false, there are competing threads in the current environment, so the current thread failed to obtain access to the synchronized resource. When tryAcquire() returns false, step 2 is performed.
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
Copy the code2. Perform the addWaiter() method
/** * Add a thread that has not acquired a shared resource to the end of the queue ** the method parameter node. EXCLUSIVE means that the thread is currently in EXCLUSIVE mode */
private Node addWaiter (Node mode){
// Create a Node with the current thread
Node node = new Node(Thread.currentThread(),mode);
// The execution logic of the following code is:
// Get the tail node of the original queue first
// Add the current node to the end of the queue, and return the new node if the queue is successful
// If you fail to join the queue, you will join the node with spin until you successfully join the queue and return to the node
Node pred = tail;
// If the end of the queue is not empty
if(pred ! =null) {// Point the prev reference of the node to the end of the queue
node.prev = pred;
// If the CAS server succeeds in joining the queue
// The node node becomes the new tail-node
if (compareAndSetTail(pred,node)){
// The next reference of the pred node at the end of the original queue points to node
pred.next = node;
/ / returns the node
returnnode; }}// If the end of the queue is empty
// Or the CAS fails to enter the queue, that is, the current environment has multiple threads competing to join the queue
// Spin by enq () method
enq (node);
// Return the node
return node;
}
Copy the codeThe addWaiter() method uses CAS (Compare And Swap) to enter the thread at the end of the queue. CAS has three operands, the memory value V, the old expected value A, and the new value B to be modified. Change the memory value V to the new value B if and only if the old expected value A is equal to the memory value V. Enter the queue using CAS in the addWaiter() method. If and only if the end node of the queue has not been modified, the current node can join the queue successfully. Otherwise, the enq() method is executed
/** * Spin the node to the end of the queue */
private Node enq(final Node node){
/ / spin
for(;;) {// End node
Node t = tail;
// If the end of the queue is empty, the queue is empty
if (t == null) {// Create a virtual header node
// Set the queue header by CAS
if (compareAndSetHead(new Node()))
// There is only one node in the queue
// The head node is equal to the tail node
tail = head;
} else (
// If the queue tail node is not null
// Set the node's prev reference to t
node.prev = t;
// Use CAS to set the new end node as node
if (compareAndSetTail(t,node)){
// The next reference to the end node of the original team points to node
// The node is the new tail node
t.next = node;
// End of spin, return the original tail node
returnt; }}}}Copy the code3. Execute the acquireQueued() method
When a thread enters the synchronization queue, it waits for the release of synchronization resources in the synchronization queue. When a thread gains access to a synchronization resource, it is dequeued from the synchronization queue.
final boolean acquireQueued(final Node node, int arg) {
// Indicates whether the lock has been acquired. The default value true means not yet
boolean failed = true;
try {
// A flag indicating whether the wait was interrupted
boolean interrupted = false;
for (;;) {
// Obtain the former node
final Node p = node.predecessor();
// If the current node is already a head node, try to acquire the lock successfully and return
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
/ / shouldParkAfterFailedAcquire according to the state of the previous node of the current node, and the judgment, making different action to the current node
//parkAndCheckInterrupt puts the thread into the wait state and checks whether the current thread can be interrupted
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true; }}finally {
// Set the current node to cancel; The cancel status is set to 1
if(failed) cancelAcquire(node); }}Copy the code- 3.1
acquireQueued()The method callpredecessor()Method to get the precursor of a node.
/** * Get the current node's precursor, throw Nul1PointerException */
final Node predecessor()throws NullPointerException {
// A reference to a precursor of the current node
Node p = prev;
// If the precursor is null, a NullPointerException is thrown
if (p == null)
throw new NullPointerException();
else
// Return the precursor
return p;
}
Copy the code- 3.2
acquireQueued()callpredecessor()Method to get the precursor of the current node, if the precursor of the current node is the head nodetryAcquire()Method attempts to obtain a resource. iftryAcquire()The getresource () method returns true, passing the resource on successsetHead()Method to set the current node to the head node.
private void setHead(Node node) {
head = rnode;
node.thread = null;
node.prev = null;
}
Copy the code- 3.3
acquireQueued()callshouldParkAfterFailedAcquire()methods
// Check and update the node state when the acquisition fails
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// Get the state of the precursor
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) // The status is SIGNAL, and the value is -1
/* * This node has already set status asking a release * to signal it, so it can safely park. */
// The park operation can be performed
return true;
if (ws > 0) { // CANCELLED (1)
/* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0); // Find the node whose state is CANCELLED until the last node before pred
// Assign the next field of the pred node
pred.next = node;
} else { // If PROPAGATE -3 or 0 means stateless,(if PROPAGATE -2 means the node is in the CONDITION queue)
/* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */
// Compare and set the state of the precursor to SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
// The park operation cannot be performed
return false;
}
Copy the code- 3.4
shouldParkAfterFailedAcquire()Method returns false, causing the thread to enteracquireQueued()Method spin; ifshouldParkAfterFailedAcquire()Method returns true, it executesparkAndCheckInterrupt()Methods.
// Park the thread and return whether the thread was interrupted
private final boolean parkAndCheckInterrupt() {
// Disable the current thread before the license is available, and blocker is set
LockSupport.park(this);
return Thread.interrupted(); // Whether the current thread has been interrupted and clears the interrupt flag bit
}
Copy the code4. Execute the selfInterrupt() method
If a thread interrupts during that time, the state of the thread interrupt is maintained, but the interrupt does not respond.
/** * sets thread interrupts and does not respond to interrupts */
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
Copy the codeThe execution process of obtaining resources in exclusive mode:
3.1.2 Releasing Resources in Exclusive Mode (Release)
In exclusive mode, releasing the right to use a resource is done by releasing the release() method, which has the following code:
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 codeIts execution process can be divided into the following steps:
1. Execute the tryRelease() method
Call the tryRelease() method to try to release access to the synchronized resource. If the release was successful, the tryRelease() method will return true and go to Step 2. If the tryRelease() method returns false, releasing the resource fails.
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
Copy the code2. Execute the unparksucceeded () method
The unparksucceeded () method wakes up successor nodes in the waiting queue so that they can compete again for synchronous resources.
/** * Wake up the next node in the synchronization queue */
private void unparkSuccessor(Node node) {
/* The node is CANCELLED (if less than zero) */
int ws = node.waitStatus;
if (ws < 0)
// Use CAS to change the node state to 0
compareAndSetWaitStatus(node, ws, 0);
/* * Node */
Node s = node.next;
// CANCELLED if the successor node is non-null and its state is greater than 0
// Indicates that the thread corresponding to the successor node cancellations the wait for the resource
if (s == null II s.waitStatus > 0) {// Set the successor node to null
s = null;
// Start from the end of the synchronous queue and go forward.
// Find the first node in the synchronization queue to wake up
// If t is not null and is not equal to the current node,
// Check the state of node t
for(Node t = tail; t ! =null&& t ! = node; t = t.prev)// If the state of the node is less than or equal to zero
if (t.waitStatus <= 0)
// Point s to t
s = t;
}
// If the node s is not null
if(s ! =null)
// Wake up the thread corresponding to node s
LockSupport.unpark(s.thread);
}
Copy the codeThe execution of the unparksucceeded () method goes as follows:
(1) Set the node status to 0.
(2) Find the next non-cancelled node s.
(3) If s is not null, call locksupport.unpark (S.htread) to wake up the thread corresponding to s.
(4) The order in which the unparksucceeded () method wakes up threads the order in which threads are added to the synchronous queue.
3.2 AQS Share Mode
3.2.1 Obtaining Resources in Shared Mode (acquireShared Method)
In shared mode, the acquireShared() method is used to obtain the right to use the resource. AcquireShared () ¶
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
Copy the codeIts execution process can be divided into the following steps:
1. Execute the tryAcquireShared() method
AbstractQueuedSynchronizer class does not implement tryAcquireShared () method, but by its subclasses. If the tryAcquireShared() method returns less than 0, that is, the thread failed to obtain the shared resource, the thread is put into the synchronization queue through the doAcquireShared() method.
- Negative values indicate failure to obtain;
- 0 indicates that the data is successfully obtained but no remaining resources are available.
- A positive number indicates success and that there are still resources left for other threads to retrieve.
// Attempts to obtain resources in shared mode. This method needs to be overridden by subclasses
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
Copy the code2. Execute the doAcquireShared() method
SHARED mode calls the addWaiter(node.shared) method to add the thread to the end of the queue, indicating that the Node is in SHARED mode. The thread that gets the resource calls the setHeadAndPropagate(node, r) method.
private void doAcquireShared(int arg) {
// Join the end of the queue
final Node node = addWaiter(Node.SHARED);
// The success flag
boolean failed = true;
try {
// A flag indicating whether the waiting process was interrupted
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();// Get the precursor node
if (p == head) {// If the node wakes up after head because head is the thread that got the resource, it is likely that head ran out of resources and woke itself up
int r = tryAcquireShared(arg);// Try to get the resource
if (r >= 0) {/ / success
setHeadAndPropagate(node, r);// Point the head to yourself and have the remaining resources to wake up later threads
p.next = null; // help GC
if (interrupted)// If there is an interruption in the process of waiting, it will be interrupted.
selfInterrupt();
failed = false;
return; }}// To determine the status, the queue looks for a suitable position and enters the waiting state, waiting to be unpark() or interrupt()
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true; }}finally {
if(failed) cancelAcquire(node); }}Copy the code3.2.2 Releasing Resources in Shared Mode (The releaseShared method)
ReleaseShared () ¶ releaseShared() ¶ releaseShared() is used to release resources in shared mode.
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
Copy the code1. Execute the tryReleaseShared() method
The tryReleaseShared() method succeeds, and the doReleaseShared() method is executed to wake up the succeeding node.
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
Copy the code2. Execute the doReleaseShared() method
Wake up the thread of the successor node
- When state is positive, the remaining shared resources are obtained.
- When state=0, obtain the shared resource.
private void doReleaseShared() {
/* * fails, if so rechecking. */
for (;;) {
Node h = head;
if(h ! =null&& h ! = tail) {/ / the head! =tail means that at least 2 nodes in the queue are blocked. Otherwise, there is no need to propagate wake.
int ws = h.waitStatus;// Head node status SIGNAL
if (ws == Node.SIGNAL) {// If the head status is
if(! compareAndSetWaitStatus(h, Node.SIGNAL,0))
continue; // loop to recheck cases
unparkSuccessor(h);// This is the same as when an exclusive lock is released. This is the same as when an exclusive lock is released
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // The value is set to -3 for the propagation of the wake. This means that the previous method has a waitStatus value less than 0
}
if (h == head)
break; }}Copy the code4 Reference Materials
Doug Lea: Concurrent Programming in Java in Action
Fang Tengfei: The Art of Java Concurrent Programming
Take a look at Java-AQS synchronizer source code interpretation < 1 > exclusive lock lock