The 24th “365 Original Plan”.

Today! Mr. Lighthouse told us:

Based on the analysis of Java and contract awarding Java. Util. Concurrent source, without need to know about AbstractQueuedSynchronizer (hereinafter abbreviated AQS) this abstract class, because it is the foundation of Java and contracting out the tools, Is the basis for implementing classes such as ReentrantLock, CountDownLatch, Semaphore, FutureTask, etc.

Google AbstractQueuedSynchronizer, we can find a lot of AQS introduction, but many are not presented clearly, because most of the articles did not clarify some of the key details.

This article will from already fair lock source code, the analysis of AbstractQueuedSynchronizer how this class works, hoping to provide some simple help for you.

State the following:

1. This article is long, but simple, and is intended for beginners in concurrent programming, or developers who want to read Java and distribute the source code. It may take a few hours for beginners to fully understand, but it’s worth the time.

2. Source code environment JDK1.7 (1.8 no changes), see do not understand or confused parts, it is best to open the source code to see. Doug Lea’s code is really good.

3. This paper does not analyze the sharing mode, so as to reduce the burden on readers. The sharing mode will be analyzed in the third article. And the condition part is not analyzed, so it is easy to read.

4. This article uses ReentrantLock, the most commonly used ReentrantLock concept, which is intrinsically incorrect. Readers should be aware that AQS is not only used to realize ReentrantLock, but only hope that readers can associate the use of AQS with locks to reduce reading pressure.

5. ReentrantLock’s fair and unfair locks are only slightly different and will be covered in a second article.

6. In the comments section, some readers gave feedback that this article is not friendly to use the code directly and should be equipped with more point flow charts. This article does have this problem. But as someone who has been there before, I want to tell you that for AQS, form really doesn’t matter. What matters is getting the details right.

AQS structure

Let’s take a look at what attributes AQS have, figure out these basic know what routine AQS is, after all, you can guess!

Private TRANSIENT volatile Node head; private transient volatile Node head; Private transient volatile Node tail; private transient volatile Node tail; // This value can be greater than 1 because locks can be reentrant. Each time the lock is reentrant, the value is added with 1private volatile int state. // represents the thread currently holding the exclusive lock. The most important use example is because the lock can be reentrantlock. lock() can be nested multiple times, so this is used each time to determine whether the current thread already owns the lockif(currentThread == getExclusiveOwnerThread()) {state++}private transient Thread exclusiveOwnerThread; / / since the AbstractOwnableSynchronizer inheritanceCopy the code

Well, it looks like it should be easy, because there are only four attributes.

AbstractQueuedSynchronizer waiting queue beckoned as shown below, pay attention to, after the analysis process in the queue, which is blocking queue does not contain the head, does not contain a head, did not include the head.

Each thread in the queue is wrapped as a Node instance. The data structure is a linked list.

Static final class Node {// Static final Node SHARED = new Node(); Static final Node EXCLUSIVE = null; // ======== the following several int constants are given towait=========== /** is used for StatuswaitStatus value to indicate thread has cancelled */ / Static final int cancelled = 1; / * *waitStatus value to indicate successorStatic final int SIGNAL = -1; 'thread needs unparking */ / static final int SIGNAL = -1; /** indicate thread is waiting on condition */ The next article will cover static final int CONDITION = -2; /** * waitStatus value to indicate the next acquireShared should */ / The same unconditionally Static final int PROPAGATE = PROPAGATE; / / = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = = / / values for the above 1, 1, 2, 3, or 0 (talk about later) / / so understand, If the value is greater than 0, the thread has cancelled the wait. // ps: ReentrantLock is not available. volatile int waitStatus; // References to the precursor Node volatile Node prev; // References to subsequent nodes volatile Node next; // Volatile Thread Thread; }Copy the code

Public class OrderService {// Use static, so that each thread has the same lock, of course, spring MVC service default singleton, Private static ReentrantLock ReentrantLock = new ReentrantLock(true);    public void createOrderReentrantlock. lock(); reentrantLock.lock(); // Usually, the lock is followed by the try statement try {// only one thread can come in at a time (the thread that acquired the lock), // other threads block on the lock() method, waiting for the lock to come in, and then execute the code... // Execute code... // Execute code... } finally {// unlock reentrantlock. unlock(); }}}Copy the code

abstract static class Sync extends AbstractQueuedSynchronizer {}Copy the code

public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }Copy the code

static final class FairSync extends Sync { private static final long serialVersionUID = -3000897897090466540L; // Lock contention final voidlock() { acquire(1); } // if tryAcquire(arg) returns, tryAcquire(arg) returnstrueAnd it's over. Public final void acquire(int arg) {public final void acquire(int arg) {tryAcquire(1) == 1; // The semantics of a fair lock are: there is no one holding the lock in the first place, so there is no need to queue.if(! TryAcquire (ARG) && // tryAcquire(ARG) did not succeed. acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) { selfInterrupt(); } } /** * Fair version of tryAcquire. Don// / Run the following command to check the status of the device: // Run the following command to check the status of the device: // Run the following command to check the status of the device: No thread is waiting for the lock; 2. Reentrant lock. The thread already holds the lock. Protected Final Boolean tryAcquire(int acquires) {final Thread current = thread.currentThread (); int c = getState(); If (c == 0) {// If (c == 0) {// If (c == 0) {// If (c == 0) {// If (c == 0) { Hasqueuedtoraise () &&if no thread is waiting, try the withcas and succeed and obtain the lock. If it fails, it can only mean one thing: a thread stole the toraise at almost the same moment =_= // since there was no one else, SetExclusiveOwnerThread (current) {compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current); return true; }} // This branch will enter the else if branch, which indicates that it is reentrant. Else if (current == getExclusiveOwnerThread()) {int nexTC = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } // if (!) {// if (!) {// if (!); tryAcquire(arg) // && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // selfInterrupt(); return false; } // If tryAcquire(arg) returns false, the code will execute: // acquireQueued(addWaiter(node.exclusive), arg), addWaiter(Node.EXCLUSIVE) /** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for EXCLUSIVE, node. SHARED for SHARED * @return the new Node */ / // the mode argument is node.exclusive, Private Node addWaiter(Node mode) {Node Node = new Node(thread.currentThread (), mode); // Try the fast path of enq; // The following lines of code want to add the current node to the bottom of the list, i.e. the last node in the blocking queue. // tail! If (tail==head, tail==head) if (pred! = null) {// Set node.prev = pred; If (compareAndSetTail(pred, node)) {// If (compareAndSetTail(pred, node)) { Pred. Next = node; pred. Next = node; // The thread is queued and can return the return node; }} pred==null(queue empty) or CAS failed (thread competing to join queue) Enq (node) that would otherwise waste time; return node; } /** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node'S predecessor */ / Adopts the mode of rotation to join the queue // As previously said, there are only two possibilities to this method: waiting queue is empty, or there are threads competing to join the queue, // Spins on this side of the semantics are: Private Node ENq (final Node Node) {for(;;) { Node t = tail; As mentioned earlier, an empty queue also comes in hereif(t == null) {// head Must initialize // tail Must initialize // head Must initialize // tail Must initialize //if(compareAndSetHead(new Node())waitStatus==0, new Node() constructor ==0, new Node() constructor =0, new Node() constructor ==0 We just set tail=head, not tail=headreturnOh, noreturn, there is noreturn// So, after setting up, continueforLoop. Next time we'll go to the bottomelseBranch tail = head; }else{// the following lines are the same as the previous method addWaiter, except that this is in an infinite loop where the current thread is queued to the end of the queue. Node.prev = t;if (compareAndSetTail(t, node)) {                    t.next = node;                    returnt; }}}} // Now back to this code //if(! tryAcquire(arg) // && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // selfInterrupt(); If (addWaiter(Node.exclusive), arg) acquireQueued(addWaiter(Node.exclusive), arg)trueThis means that the code above will enter selfInterrupt(), so normally this should returnfalseFinal Boolean acquireQueued(final Node Node, int arg) {Boolean failed = final Boolean acquireQueued(final Node Node, int arg) {Boolean failed =true;        try {            boolean interrupted = false;            for(;;) { final Node p = node.predecessor(); // p == head; // p == head; // P == head; // So the current node can try to grab the lock // here we explain why we can try: // This is the first condition. Second, it is possible that the current head is the newly initialized node. As mentioned in the enq(node) method, the head is delayed and no thread is set when new node () is created. The current head does not belong to any thread, so as the head of the queue, you can try it, // tryAcquire has already analyzed, please look ahead, just use CAS to try stateif (p == head && tryAcquire(arg)) {                    setHead(node);                    p.next = null; // help GC                    failed = false;                    returninterrupted; } // Go to this point to illustrate the aboveifThe branch failed, either the current node is not the team leader, // or tryAcquire(ARG) did not beat the others, continue to readif (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    interrupted = true; }} finally {// When failedtrue????? // The tryAcquire() method throws exceptionsif (failed)                cancelAcquire(node);        }    }    /**     * Checks and updates status for a node that failed to acquire.     * Returns true if thread should block. This is the main signal     * control in all acquire loops.  Requires that pred == node.prev     *     * @param pred nodeThe predecessor holding status * @param node the node * @return {@code true} if thread should block */ / This method says: "The current thread did not grab the lock. Do we need to suspend the current thread?" // The first parameter is the precursor node, The second parameter is the representative of the current thread Node private static Boolean shouldParkAfterFailedAcquire (Node Mr Pred, Node Node) {int ws = Mr Pred. WaitStatus; // If waitStatus == -1, the status of the precursor node is normal and the current thread needs to suspend. If (ws == node. SIGNAL) /* * This Node has already set status asking a release * to SIGNAL it, so it can safely park. */ return true; // The precursor waitStatus is greater than 0, which means that the precursor is unqueued. // The thread that enters the blocking queue is suspended, and the wake up is done by the precursor node. // Set the prev of the current node to a node whose waitStatus<=0; // Simply to find a good parent, because you still have to rely on it to wake up. If (ws > 0) {* * Predecessor was cancelled. Skip over boomand * indicate retry. */ do {* * Predecessor was cancelled over boomand * indicate retry. node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don'T park yet. Caller will need to * retry to make sure it cannot acquire before parking. What does it mean if you enter this branch // the precursor nodewaitStatus is not equal to -1 or 1, that is, it can only be 0, -2, -3 // in our previous source, there is no settingwaitSo when each new node joins the queue,waitStatu is 0; // In normal cases, tail is the previous node, then its tailwaitStatus should be 0 // Use CAS to place the precursor nodewaitCompareAndSetWaitStatus (pred, ws, node. SIGNAL); } // This method returnsfalse"Then it will go againforStep by step, // then enter the method again, which will return from the first branchtrue        return false; } / / private static Boolean shouldParkAfterFailedAcquire (Node Mr Pred, Node Node) / / the end of this method according to the return value we simple analysis: / / if it returnstrue, indicating that of the precursor nodewaitStatus==-1, which is normal, then the current thread needs to be suspended, waiting to be woken up laterfalse"Does not currently need to be suspended. Why? Look back // jump back to the front is this method //if (shouldParkAfterFailedAcquire(p, node) &&    //                parkAndCheckInterrupt())    //                interrupted = true; / / 1. If shouldParkAfterFailedAcquire (p, node) to returntrue, // Then parkAndCheckInterrupt(): // This method is easy because the previous returntrue// Locksupport. park(this) suspends the thread and then waits to be woken up ======= private final BooleanparkAndCheckInterrupt() {        LockSupport.park(this);        returnThread.interrupted(); } / / 2. Then tell me if shouldParkAfterFailedAcquire (p, node) to returnfalse/ / see shouldParkAfterFailedAcquire (p, node), we can find that actually came in for the first time, generally will not returntrueThe reason is very simple, the precursor nodewaitStatus=-1 depends on the setting of subsequent nodes. In other words, I haven't even set the front drive to -1. How could it betrueBut you see, this method is stuck in the loop, so the second time it comes in it's going to be negative 1. / / explain why shouldParkAfterFailedAcquire (p, node) returnfalse// => in case node is a direct descendant of head after this method. Figure it out for the rest of you. }Copy the code

Public void * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *unlock() { sync.release(1); }public final Boolean release(int arg) {public final Boolean release(int arg)if (tryRelease(arg)) {        Node h = head;        if(h ! = null && h.waitStatus ! = 0) unparkSuccessor(h);return true;    }    return false; }// Go back to ReentrantLock and look at the tryRelease method protected Final Boolean tryRelease(int releases) {int c = getState() -releases;if(Thread.currentThread() ! = getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); // Whether to release the lock completely Boolean free =false; If c==0, no nested locks can be released. If c==0, no nested locks can be releasedif (c == 0) {        free = true;        setExclusiveOwnerThread(null);    }    setState(c);    returnfree; }/** * Wakes up nodeSucceeded, if one exists. * * @param node the node *// Know from the preceding call, Private void unparksucceeded (node node) {/* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; If (ws <0) compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled nodes. */ / S = node.next; s= node.next; s= node.next; if (s == null || s.waitStatus > 0) { s = null; For (Node t = tail; waitStatus==1; waitStatus==1; t ! = null && t ! = node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s ! = null) // Wake up thread locksupport. unpark(s.read); }Copy the code

private final boolean parkAndCheckInterrupt() { LockSupport.park(this); // The thread was suspended herereturnThread.interrupted(); AcquireQueued (final Node Node, int arg) acquireQueued(final Node Node, int argCopy the code