Shulou(Shulou.com)06/01 Report--

This article mainly introduces the relevant knowledge of "what is the principle of java Synchronizer AQS Architecture AbstractQueuedSynchronizer". The editor shows you the operation process through an actual case. The method of operation is simple and fast, and it is practical. I hope this article "what is the principle of java Synchronizer AQS Architecture AbstractQueuedSynchronizer" can help you solve the problem.

Introduction language

The Chinese translation of AbstractQueuedSynchronizer is called synchronizer, or AQS for short, which is the basis of all kinds of locks, such as ReentrantLock, CountDownLatch and so on. The underlying implementation of these locks that we often use is AQS, so it is very important to learn AQS well for later understanding the implementation of locks.

The content of the lock chapter is arranged as follows:

There is a lot of 1:AQS source code. We will divide it into two sections to clarify the underlying principles first.

2: we usually do not use AQS, we only come into contact with locks such as ReentrantLock and CountDownLatch. Let's take two locks as examples to explain the source code, because as long as AQS understands all the locks, as long as you know the purpose of the lock, you can use AQS to implement it.

3: summarize the interview questions of the lock

4: summarize the use scenarios of locks at work, give a few practical examples, and see what matters needing attention when using locks.

5: finally, let's implement a lock ourselves and see what steps we have and what we need to pay attention to if we implement the lock ourselves.

Ps: this chapter requires a lot of basic knowledge of queues. For students who have not read the fourth chapter, it is recommended to read the queuing chapter first.

1. Overall structure

First, let's take a look at the overall architecture diagram of AQS, as follows:

This diagram summarizes the composition of the overall architecture of AQS, and the dynamic flow direction of some scenarios. Two points in the figure are explained to facilitate your viewing.

There are only two queues in AQS: synchronous queues + conditional queues, and the underlying data structures are both linked lists.

There are four colored lines in the picture that represent four different scenes, and the serial numbers 1, 2 and 3 represent the order in which they are viewed.

AQS itself is a lock framework that defines the code structure for acquiring and releasing locks, so if you want to create a new lock, you just need to inherit AQS and implement the corresponding methods.

Next, let's take a look at the details in this diagram.

1.1. Class comments

First, let's take a look at what we can get from the AQS class comments:

A framework is provided to customize the first-in-first-out synchronization queue, so that threads that cannot acquire the lock can queue in the synchronization queue.

The synchronizer has a status field, which can be used to determine whether the lock can be obtained. The key to the design is to rely on the safe atomic value to represent the state. (although this is what the annotation means, it actually ensures thread safety by declaring the state as volatile and modifying the state value in the lock.)

Subclasses can decide whether to acquire a lock by assigning a value to the state CAS. They can define which states can acquire the lock and which states indicate that the lock cannot be obtained (for example, if the state value is 0, the lock can not be obtained, and if the state value is 1, the lock can not be obtained).

The subclass can create a non-public inner class and inherit the AQS with the inner class, thus realizing the lock function.

AQS provides two lock modes: exclusive mode and shared mode. In exclusive mode: only one thread can acquire the lock, the shared mode allows multiple threads to acquire the lock, and the subclass ReadWriteLock implements two modes

The inner class ConditionObject can be used as a Condition, and we can get the conditional queue through new ConditionObject ().

AQS implements locks, queues, lock queues and other frameworks, but the code on how to acquire and release locks is not implemented, such as tryAcquire, tryRelease, tryAcquireShared, tryReleaseShared and isHeldExclusively. UnsupportedOperationException exceptions are thrown by default in AQS, which need to be implemented by subclasses.

AQS inherits AbstractOwnableSynchronizer to facilitate tracking of threads that acquire locks, and can help monitoring and diagnostic tools identify which threads hold locks.

AQS synchronizes queues and conditional queues. Nodes that cannot acquire locks are first-in-first-out when they join the queue, but may not be executed in first-in-first-out order when awakened.

There are many more comments on AQS, and the above 9 points are selected for a slightly more important summary of comments.

1.2. Class definition

The AQS class definition code is as follows:

Public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {

Two points can be seen:

AQS is an abstract class, which is used for inheritance of various lock subclasses. AQS defines many abstract methods of how to acquire and release locks. The purpose is to allow subclasses to implement.

Inheriting the function of AbstractOwnableSynchronizer,AbstractOwnableSynchronizer is to know which thread has acquired the lock, which is convenient for monitoring.

The code is as follows:

1.3. Basic attributes

The properties of AQS can be simply divided into four categories: synchronizer simple properties, synchronization queue properties, conditional queue properties, and public Node.

1.3.1, simple attributes

First, let's take a look at what simple properties are:

/ / the status of the synchronizer. The subclass will determine whether the lock can be obtained according to the status field. For example, CAS successfully assigns the lock to state. 1 calculates the lock. If the assignment fails, the lock is not obtained. The CAS successfully assigns the state 0 as the lock is released, and the assignment failure is the release failure / / reentrant lock. Each time the lock is obtained + 1, and each time the lock is released-1private volatile int state. / / spin timeout threshold, unit nanosecond / / this attribute static final long spinForTimeoutThreshold = 1000L is used only when the wait time is set.

The most important thing is the state attribute, which belongs to the int attribute. All locks that inherit AQS use this field to determine whether they can acquire the lock and release the lock.

1.3.2. Synchronization queue properties

First of all, let's introduce the following synchronization queue: when multiple threads come to request a lock, one and only one thread can acquire the lock (exclusive lock) at some time, then the remaining threads that cannot acquire the lock will queue in the synchronization queue and block themselves. When a thread actively releases the lock, a queued thread will be released from the synchronous queue header, allowing the thread to compete for the lock again.

So the main role of the synchronization queue is to block threads that cannot acquire the lock and release those threads at the appropriate time.

The underlying data structure of the synchronization queue is a two-way linked list. We can see the beginning and end of the linked list from the source code, as follows:

/ / the head of the synchronization queue. Tail private transient volatile Node tail of the private transient volatile Node head;// synchronization queue

The Node in the source code is an element in the synchronization queue, but the Node is common to the synchronization queue and the conditional queue, so we won't talk about Node until we finish talking about the conditional queue.

1.3.3. Attributes of conditional queues

First of all, let's introduce conditional queues: conditional queues have the same function as synchronous queues, which manage threads that cannot acquire locks, and the underlying data structure is also a linked list queue, but conditional queues do not directly deal with locks, but they are often used in conjunction with locks. It is a supplement to the locking function in certain scenarios.

The properties of the conditional queue are as follows:

/ / conditional queue, as can be seen from the attributes, it is a linked list structure public class ConditionObject implements Condition, java.io.Serializable {private static final long serialVersionUID = 1173984872572414699L; / / the first node private transient Node firstWaiter; / / the last node private transient Node lastWaiter; in the conditional queue}

ConditionObject is what we call conditional queues, and when we need to use it, we can simply new ConditionObject ().

ConditionObject implements the Condition interface, and the Condition interface is equivalent to various monitoring methods of Object, such as Object#wait (), Object#notify, and Object#notifyAll, which we can understand in this way and will be described in more detail later.

1.3.4 、 Node

Node is very important, that is, the node of the synchronous queue and the node of the conditional queue. When joining the queue, we wrap the thread with Node, and then put the Node into two queues. Let's take a look at the data structure of Node, as shown below:

Single attribute of static final class Node {/ * synchronous queue * / node is shared mode static final Node SHARED = new Node (); / / node is exclusive mode static final Node EXCLUSIVE = null; / / the former node of the current node / / node acquire becomes head / / head node cannot be cancelled volatile Node prev after success / / the next node of the current node volatile Node next; / * the attribute shared by the two queues * / / indicates the status of the current node. The behavior of the node is controlled by the state of the node / / ordinary synchronous node, which is 0. The condition node is CONDITION-2 volatile int waitStatus. / / the states of waitStatus are as follows / / canceled static final int CANCELLED = 1; / / meaning of SIGNAL status: when a node in the synchronization queue is spinning to acquire a lock, if the state of the previous node is SIGNAL, then it can block and rest, otherwise it keeps spinning trying to get the lock static final int SIGNAL =-1. / / indicates that the current node is in the conditional queue. When a node is transferred from the synchronization queue to the conditional queue, the state will be changed to CONDITION static final int CONDITION =-2; / / unconditionally propagated. In shared mode, the process in this state is in runnable state static final int PROPAGATE =-3; / / the thread volatile Thread thread of the current node / / in the synchronization queue, nextWaiter does not really point to the next node. We use next to represent the next node of the synchronization queue. NextWaiter only indicates whether the current Node is exclusive mode or shared mode / / but in the conditional queue, nextWaiter represents the next node element Node nextWaiter;}.

From the structure of Node, we need to focus on the waitStatus field, and many of the operations of Node revolve around the waitStatus field.

The pre and next attributes of Node are the pointing fields before and after the list in the synchronization queue, and nextWaiter is the pointing field of the next node in the conditional queue, but in the synchronization queue, nextWaiter is just an identifier indicating whether the current node is in shared or exclusive mode.

1.3.5, the difference between shared lock and exclusive lock

An exclusive lock means that only one thread can acquire the lock and only one thread can release the lock at a time.

Shared locks allow multiple threads to acquire the same lock, and you can set the number of threads that acquire the lock.

1.4 、 Condition

Just now, when we looked at the conditional queue ConditionObject, we found that it implemented the Condition interface. Now let's take a look at the Condition interface. This is what the class annotation says:

When lock replaces synchronized to add locks, Condition can be used to replace the corresponding monitoring methods in Object, such as Object#wait (), Object#notify, Object#notifyAll and so on.

Provides a way of thread collaboration: a thread is suspended until it is awakened by another thread

The Condition instance is bound to the lock, and can be generated by the Lock#newCondition method

Except for special instructions, any null value as the input parameter of the method will throw a null pointer

Condition provides explicit semantics and behavior, unlike the Object monitoring approach.

Class comments even give us an example:

Suppose we have a queue with bounded boundaries that supports put and take methods and needs to meet:

1: if you try to execute take on an empty queue, the thread will block until there are available elements in the queue

2: if you try to execute put on a full queue, the thread will block until there is free space in the queue.

Thread blocking in 1 and 2 will be blocked in the conditional queue.

If we rely on one conditional queue for take and put operations, we can only perform one operation at a time, so we can create two new conditional queues so that we can perform operations separately. Looking at this requirement, do you think it is very similar to the queue we learned in Chapter 3? In fact, the demo given in the comments is the queue we have studied, and the space is limited. If you are interested, you can take a look at the test class ConditionDemo.

In addition to class annotations, Condition defines some methods that lay the foundation for conditional queues, mainly:

Void await () throws InterruptedException

The main purpose of this method is to make the current thread wait until it is signalled or interrupted.

Threads in the conditional queue will be awakened when the following four situations occur

A thread uses the signal method, which wakes up the current thread in the conditional queue

Threads use the signalAll method

Another thread interrupted the current thread, and the current thread support was interrupted

Be falsely awakened (even if the above three conditions are not met, wait may be awakened occasionally. The definition of false awakening can be referred to: https://en.wikipedia.org/wiki/Spurious_wakeup).

One thing to note when awakened is that when a thread wakes up from a conditional queue, it must regain the lock in order to be really awakened, which we will emphasize when talking about the source code.

The await method also has a wait timeout, as follows:

/ / the long value returned represents the remaining given waiting time. If the returned time is less than or equal to 0, the waiting time has passed / / nanosecond is selected to avoid the truncation error long awaitNanos (long nanosTimeout) throws InterruptedException;// when calculating the remaining waiting time. Although the input parameter can be any unit of time, the underlying layer is still converted to nanosecond boolean await (long time, TimeUnit unit) throws InterruptedException.

In addition to the wait method, there are two ways to wake up the thread, as follows:

/ / Wake up a thread in the conditional queue, which must acquire the lock void signal () before being woken up; / / Wake up all threads in the conditional queue void signalAll ()

At this point, the basic properties of AQS have been introduced, and let's take a look at the important methods of AQS.

2. The state of the synchronizer

In the synchronizer, we have two states, one called state and the other called waitStatus, which are completely different concepts:

State is the state of the lock and the int type. When a subclass inherits AQS, it needs to determine whether it has been locked according to the state field. For example, if the current synchronizer state is 0, it means that the lock can be acquired. The current synchronizer state is 1, which means that the lock is already held by another thread, and the current thread cannot acquire the lock.

WaitStatus is the state of a node (Node), and there are many kinds, such as initialization (0), CANCELLED (1), SIGNAL (- 1), CONDITION (- 2), PROPAGATE (- 3). The meaning of each state can be found above.

We need to keep these two states in mind and don't confuse them.

3. Acquire the lock

The most intuitive way to acquire a lock is to use the Lock.lock () method to acquire the lock, with the ultimate goal of giving the thread access to the resource.

Lock is generally a subclass of AQS, and the lock method generally chooses to call the acquire or tryAcquire method of AQS depending on the situation.

The acquire method AQS has been implemented, and the tryAcquire method is waiting for the subclass to implement. The acquire method establishes a framework for acquiring the lock, first tries to use the tryAcquire method to acquire the lock, and then enters the synchronization queue to wait for the lock when it is not available. An exception is thrown directly in the tryAcquire method AQS, indicating that the subclass is required to implement it. The subclass can decide whether to acquire the lock according to the state status of the synchronizer. Next, let's take a look at the source code parsing of acquire in detail.

Acquire is also divided into two types, one is exclusive lock, the other is shared lock, let's take a look at one by one:

3.1.In acquire exclusive lock / / exclusive mode, the public final void acquire (int arg) {/ / tryAcquire method needs to be implemented by the implementation class. Generally speaking, cas assigns a value to the state to determine whether the lock if (! tryAcquire (arg) & & / addWaiter input parameter represents the exclusive mode acquireQueued (addWaiter (Node.EXCLUSIVE), arg) selfInterrupt ();}

The main steps of the above code are (see the red scene in the overall architecture diagram for the flow):

If the thread attempts to enter the synchronization queue, first call the addWaiter method to put the current thread at the end of the synchronization queue; then call the acquireQueued method, which has two functions: 1: block the current node, 2: when the node is awakened, it can obtain the lock; if 2, 3 fails, interrupt the thread.

3.1.1 、 addWaiter

There is very little code, and each method is the key, so let's take a look at the source code implementation of addWaiter:

/ / main purpose of the method: node appends to the end of the synchronous queue / / input parameter mode indicates the mode of Node (exclusive mode or shared mode) / / output parameter is the main idea of the new node//: / / New node.pre = end of queue / / end of queue. Next = New nodeprivate Node addWaiter (Node mode) {/ / initialization Node Node node = new Node (Thread.currentThread (), mode) / / the logic here is the same as enq. The logic of enq is only empty at the end of the queue. The idea of initialization logic / / is very common in java source code. Simply try to put it first, and then return immediately. If not, then while loop / / many times, this algorithm can help solve most of the problems, most of which may be successful at one time without the need to spin Node pred = tail. If (pred! = null) {node.prev = pred; if (compareAndSetTail (pred, node)) {pred.next = node; return node;}} / / spin guarantee that node is added to enq (node) at the end of the queue; return node } / / the thread joins the synchronization queue and appends it to the end of the queue / / it should be noted that the return value is the previous node private Node enq (final Node node) {for (;;) {/ / get the end node Node t = tail where the node was added. / / if the end of the queue is empty, the current synchronization queue is not initialized. Initialize / / tail = head = new Node (); if (t = = null) {if (compareAndSetHead (new Node () tail = head; / / if the end of the queue is not empty, append the current node to the end of the queue} else {node.prev = t / / node is appended to the end of the line if (compareAndSetTail (t, node)) {t.next = node; return t;}

If you have studied the queue before, you should feel no difficulty with this method, which is to append the new node to the end of the synchronization queue.

One of the things worth learning is that in the addWaiter method, we do not spin immediately after entering the method, but try to append it to the end of the queue first, and spin only if it fails, because most of the operations may succeed at one time. This idea can be used for reference when we write spin.

3.1.2 、 acquireQueued

The next step is to block the current thread, which is implemented by the acquireQueued method. Let's take a look at the source code implementation:

/ / do two main things: / / 1: change the state of your previous node to signal by constantly trying to spin, and then block yourself. / / 2: after the execution of the thread that acquired the lock is completed, the blocked node will be awakened when the lock is released, and the node will spin again after waking up. The attempt to acquire the lock / / return false indicates that the lock was successful. Returning true indicates failure final boolean acquireQueued (final Node node, int arg) {boolean failed = true; try {boolean interrupted = false; / / spin for ( ) {/ / choose a node final Node p = node.predecessor (); / / there are two cases where the lock is not acquired before p = = head: / / 1:node. When you enter the acquireQueued method, you find that his front node is the header node, so you try to get the lock once. / / 2:node has been blocking sleep before, and then is awakened. The node that wakes up node is the node in front of it, and can also go to if / / if your tryAcquire succeeds, immediately set yourself to head, and remove the previous node / / if tryAcquire fails. Try to enter the synchronization queue if (p = = head & & tryAcquire (arg)) {/ / acquire the lock and set it to head node setHead (node) / / p is recycled p.next = null; / / help GC failed = false; return interrupted } / / shouldParkAfterFailedAcquire sets the previous node state of node to SIGNAL / / as long as the previous node state is SIGNAL, then it can park / / parkAndCheckInterrupt blocks the current thread if (shouldParkAfterFailedAcquire (p, node) & & / / thread is blocked in this method When you wake up still in an infinite for loop, you can spin again and try to get the lock parkAndCheckInterrupt () interrupted = true. }} finally {/ / if it fails to obtain a lock for node, remove node from the queue if (failed) cancelAcquire (node);}}

The annotation of this method is still very clear, let's take a look at the core of this method: shouldParkAfterFailedAcquire, the main purpose of this method is to set the state of the previous node to SIGNAL, as long as the state of the previous node is SIGNAL, the current node can block (parkAndCheckInterrupt is the way to block the node)

The source code is as follows

/ / the current thread can rest assured that the thread state of the previous node is SIGNAL. / / the input parameter pred is the previous node, and node is the current node. / / key action: / / 1: confirm whether the previous node is valid. If not, go ahead and find the node whose status is not canceled. / / 2: set the previous node status to SIGNAL. The two-step operation of / / 1 and 2 may be successful at one time, and it may require an external loop to succeed many times (there is an infinite for loop outside), but in the end it must be a successful private static boolean shouldParkAfterFailedAcquire (Node pred, Node node) {int ws = pred.waitStatus. / / if the waitStatus status of the previous node is already SIGNAL, return it directly. There is no need to spin if (ws = = Node.SIGNAL) / * * This node has already set status asking a release * to signal it, so it can safely park. * / return true; / / if the current node status has been cancelled. If (ws > 0) {/ * * Predecessor was cancelled. Skip over predecessors and * indicate retry. * / / find the node whose previous state is not canceled, because the current node is hung on the valid node / / because the node status is cancelled, it is invalid and cannot be used as the front node of node, so you must find a valid node of node to do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0). Pred.next = node; / / otherwise directly set the node state to SIGNAL} else {/ * * 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. * / compareAndSetWaitStatus (pred, ws, Node.SIGNAL);} return false;}

The whole process of acquire is very long and there is a lot of code, but the comments are so clear that you can take a closer look at the code one by one.

To sum up, the acquire method is roughly divided into three steps:

Use the tryAcquire method to try to get the lock, get the lock and return it directly. If you can't get the lock, go 2.

Assemble the current thread into a node (Node) and append it to the end of the synchronization queue (addWaiter)

Spin so that the leading node state of the current node in the synchronization queue is signal, and then block itself.

The overall code structure is relatively clear, some points that need to be noted are indicated by comments, it is strongly recommended to read the source code.

3.2. acquireShared acquires the shared lock

The overall process of acquireShared is the same as that of acquire, and the code is also very similar. The repeated source code will not be posted, so we will post different code to facilitate comparison:

The first step is to try to acquire the lock differently. The exclusive lock uses the tryAcquire method, while the shared lock uses the tryAcquireShared method, as shown in the following figure:

The second step is different. When a node acquires an exclusive lock, it can just set itself as the head node of the synchronization queue (setHead method), but if it is a shared lock, it will also wake up its own subsequent nodes to obtain the lock (setHeadAndPropagate method). The differences are as follows (left exclusive lock, right shared lock):

Let's take a look at the source code of the setHeadAndPropagate method:

/ / mainly do two things / / 1: set the current node as the head node / / 2: see if the subsequent nodes are waiting and are also in shared mode, and if so, wake up these nodes private void setHeadAndPropagate (Node node, int propagate) {Node h = head; / / Record old head for check below / / the current node is set to the head node setHead (node) / * * Try to signal next queued node if: * Propagation was indicated (for instruction) 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 (conservative) 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. * / propagate > 0 indicates that a node has acquired a shared lock if (propagate > 0 | | h = = null | | h.waitStatus < 0 | (h = head) = = null | | h.waitStatus < 0) {Node s = node.next; / / sharing mode, and wakes up the rear node of the head node if (s = = null | | s.isShared ()) doReleaseShared () }} / / release private void doReleaseShared () {/ * * 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 (;;) {Node h = head; / / has not reached the end of the queue, and there are at least two nodes in the queue if (h! = null & & h! = tail) {int ws = h.waitStatus / / if the queue status is SIGNAL, all subsequent nodes need to wake up if (ws = = Node.SIGNAL) {/ / CAS to ensure that only one node can run the wake-up operation if (! compareAndSetWaitStatus (h, Node.SIGNAL, 0)) continue / / loop to recheck cases / / Wake up operation unparkSuccessor (h);} else if (ws = = 0 & &! compareAndSetWaitStatus (h, 0, Node.PROPAGATE)) continue / / loop on failed CAS} / / in the first case, the header node does not move and ends. / / the second case, because this method can be called in two places, once where the lock is obtained and where the lock is released. / / in addition, the shared lock feature is that multiple threads can acquire the lock, and the lock can also be released. This causes the header node to change, / / if the header node changes, it continues to loop until the header node does not change. End the cycle. If (h = = head) / / loop if head changed break;}}

This is what makes the shared lock unique. When a thread acquires the lock, it wakes up the other nodes behind it so that other nodes can also acquire the lock.

This is the end of the introduction to "what is the principle of java Synchronizer AQS Architecture AbstractQueuedSynchronizer". Thank you for reading. If you want to know more about the industry, you can follow the industry information channel. The editor will update different knowledge points for you every day.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.