JUC包 — locks — AbstractQueuedSynchronizer(AQS源码分析)

以你之姓@ 2022-03-24 14:35 272阅读 0赞

继续拆解JUC包，其中子包除了atomic还有一个locks包。打开之后我们能看到几个熟悉的锁，比如：Lock，LockSupport，ReentrantLock，ReadWriteLock，ReentrantReadWriteLock等。但是里面还有一个非常重要的类，ReentrantLock，ReentrantReadWriteLock的底层都是由它实现。这个类就是——**AbstractQueuedSynchronizer**

# 参数 #

AbstractQueuedSynchronizer总共有三个参数。  
说参数前先说一个内部类**Node**  
还有一个内部类：Node

## Node ##

它其实是实现了一个内部的等待序列。实现顺序如下所示：

+------+  prev +-----+       +-----+
          head |      | <---- |     | <---- |     |  tail
               +------+       +-----+       +-----+

### int waitStatus ###

/**
             * 表示节点的状态。其中包含的状态有：
    		 * ① CANCELLED，值为1，当线程等待超时或者被中断，则取消等待，设等待状态为-1，进入取消状态则不再变化；
    		 * ② SIGNAL，值为-1，后继节点处于等待状态，当前节点（为-1）被取消或者中断时会通知后继节点，使后继节点的线程得以运行；
    		 * ③ CONDITION，值为-2，当前节点处于等待队列，节点线程等待在Condition上，当其他线程对condition执行signall方法时，等待队列转移到同步队列，加入到对同步状态的获取；
    		 * ④ PROPAGATE，值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态；
    		 * ⑤ 值为0，代表初始化状态，当前节点在sync队列中，等待着获取锁。
             */
            volatile int waitStatus;

### Node prev ###

/**
             * 前驱节点，当前节点前一个节点。
             */
            volatile Node prev;

### Node next ###

/**
             * 后置节点，当前节点后一个节点。
             */
            volatile Node next;

### Thread thread ###

/**
             * 当节点排队(入队列)的线程。在构造时初始化，使用后为空。
             */
            volatile Thread thread;

### Node nextWaiter ###

/**
             * 存放等待后继节点
             */
            Node nextWaiter;

### Node head ###

/**
         * Head of the wait queue, lazily initialized.  Except for
         * initialization, it is modified only via method setHead.  Note:
         * If head exists, its waitStatus is guaranteed not to be
         * CANCELLED.
         * 等待队列的头，已延迟初始化。除了初始化，它只能通过方法<sethead>进行修改。
         * 注意：如果head存在，则保证其waitstatus不会被取消。
         */
        private transient volatile Node head;

### Node tail ###

/**
         * Tail of the wait queue, lazily initialized.  Modified only via
         * method enq to add new wait node.
         * 等待队列的尾部，已延迟初始化。仅通过方法<enq>修改以添加新的等待节点。
         */
        private transient volatile Node tail;

### long state ###

/**
         * The synchronization state.
         * 同步状态
         */
        private volatile long state;

# 重要方法 #

我们看到AbstractQueuedSynchronizer中有非常多的方法，查看其中几个方法，不难发现其中都调用了这样几个方法。

1.  boolean **tryAcquire**(int arg)：以独占方式，尝试获取资源，成功则返回true，失败则返回false。
2.  boolean **tryRelease**(int arg)：以独占方式，尝试释放资源，成功则返回true，失败则返回false。
3.  int **tryAcquireShared**(int arg)：以共享方式，尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
4.  boolean **tryReleaseShared**(int arg)：以共享方式，尝试释放资源，成功则返回true，失败则返回false。

所以我们可以猜测到，为了使框架能得到广泛应用，AQS同步器定义两种资源共享方式：

1.  **Exclusive**：独占模式，同时只有一个线程能执行，如ReentrantLock
2.  **Share**：共享模式，多个线程可同时执行，如Semaphore/CountDownLatch。

一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现上面四种方法中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式，如**ReentrantReadWriteLock**。

自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。

AQS为了实现上述操作，需要下面三个基本组件的相互协作：

1.  同步状态的原子性管理；
2.  线程的阻塞与解除阻塞；
3.  队列的管理；

## 同步状态的原子性管理 ##

在一开始我们说参数的时候，state字段已经用volatile进行修饰，保证其可见性。但是不能保证其原子性，所以存在一个CAS方法，来保证原子性：

/**
         * CAS waitStatus field of a node.
         */
        private static final boolean compareAndSetWaitStatus(Node node,
                                                             int expect,
                                                             int update) {
            return unsafe.compareAndSwapInt(node, waitStatusOffset, expect, update);
        }

## 线程的阻塞与解除阻塞 ##

利用LockSupport.park() 和 LockSupport.unpark() 实现线程的阻塞和唤醒(底层调用Unsafe的native park和unpark实现)，同时支持超时时间。

## 队列管理 ##

在最开始介绍的内部类Node字段时，简单的解释了下AbstractOwnableSynchronizer中的队列。

# 源码 #

## 先看**acquire** ##

/**
         * 以独占模式获取，忽略中断。
         */
        public final void acquire(int arg) {
        	// 尝试获取资源，如果成功则直接返回；
            if (!tryAcquire(arg) &&
            	// addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
            	// acquireQueued()使线程在等待队列中获取到资源后返回。如果在整个等待过程中被中断过，则返回true，否则返回false。如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

逐个分析其中方法：

### **tryAcquire** ###

protected boolean tryAcquire(int arg) {
            throw new UnsupportedOperationException();
        }

我们看到这个方法直接返回一个异常。其实AQS这里只定义了一个接口，具体资源的获取交由自定义同步器去实现了（通过state的get/set/CAS）。至于能不能重入，能不能加塞，那就看具体的自定义同步器怎么去设计了！！！当然，自定义同步器在进行资源访问时要考虑线程安全的影响。

### **addWaiter** ###

/**
         * Creates and enqueues node for current thread and given mode.
         * 为当前线程和给定模式创建和排队节点。
         */
        private Node addWaiter(Node mode) {
        	// 以给定模式构造结点。将nextWaiter设置为EXCLUSIVE：
        	mode有两种：EXCLUSIVE（独占）和SHARED（共享）
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            // 尝试快速方式直接放到队尾。
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            //上一步失败则通过enq入队。
            enq(node);
            return node;
        }

### **enq** ###

private Node enq(final Node node) {
       	 	// CAS"自旋"，直到成功加入队尾
            for (;;) {
                Node t = tail;
                // 判断尾节点是否为空
                if (t == null) { 
                	// CAS算法设置头节点
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else { // 尾节点不为空
                	// 将节点的前驱更新为尾节点
                    node.prev = t;
                    // CAS算法设置尾节点
                    if (compareAndSetTail(t, node)) {
                    	// 将当前节点指向尾节点的后置
                        t.next = node;
                        return t;
                    }
                }
            }
        }

如果通过**tryAcquire()和addWaiter()**，该线程获取资源失败，已经被放入等待队列尾部了。进入等待状态休息，直到其他线程彻底释放资源后唤醒自己，自己再拿到资源，然后就可以去干自己想干的事了。

### **acquireQueued** ###

/**
         * Acquires in exclusive uninterruptible mode for thread already in
         * queue. Used by condition wait methods as well as acquire.
         * 为已经在队列中的线程以独占不间断模式获取。 由条件等待方法使用以及获取。
         */
        final boolean acquireQueued(final Node node, int arg) {
       		// 标记是否成功拿到资源
            boolean failed = true;
            try {
            	// 标记等待过程中是否被中断过
                boolean interrupted = false;
                // 自旋            
                for (;;) {
                	// 拿到前驱
                    final Node p = node.predecessor();
                    // 如果前驱是head，即该结点已成老二，那么便有资格去尝试获取资源（可能是老大释放完资源唤醒自己的，当然也可能被interrupt了）。
                    if (p == head && tryAcquire(arg)) {
               			// 拿到资源后，将head指向该结点。设置为头结点。
                        setHead(node);
                       	// setHead中node.prev已置为null，此处再将head.next置为null，就是为了方便GC回收以前的head结点。也就意味着之前拿完资源的结点出队了。
                        p.next = null; // help GC
                        failed = false;
                        // 返回等待过程中是否被中断过
                        return interrupted;
                    }
                    // 如果自己可以休息了，就进入waiting状态，直到被unpark()
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        // 如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为true
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

### shouldParkAfterFailedAcquire ###

/**
         * 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.
         *
         * 检查和更新无法获取的节点的状态。如果线程应该阻塞，则返回true。
         * 这是所有采集回路中的主信号控制。要求pred==node.prev。
         */
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            int ws = pred.waitStatus;
            // 前驱状态等于SIGNAL
            if (ws == Node.SIGNAL)
                /*
                 * 此节点已经设置了请求释放信号的状态，因此可以阻塞
                 */
                return true;
                // 如果说前驱被取消了。>0 只有 = 1 的情况
            if (ws > 0) {
                /*
                 * 前驱被取消，此时跳过前驱，重试
                 */
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                /*
                 * 更改前驱状态为SIGNAL
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

### **selfInterrupt** ###

/**
         * 中断当前线程
         */
        static void selfInterrupt() {
            Thread.currentThread().interrupt();
        }

### acquire总结 ###

流程：  
首先调用tryAcquire()方法，尝试获取获取同步状态，如果同步状态获取失败，则在当前线程构造独占式同步节点。通过addWaiter()方法将节点加入到同步队列的尾部，最后调用acquireQueued()方法，使得节点以自旋的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

下面说下

## release ##

/**
         * Releases in exclusive mode.  Implemented by unblocking one or
         * more threads if {@link #tryRelease} returns true.
         * This method can be used to implement method {@link Lock#unlock}.
         * 以独占模式释放。如果tryRelease返回true，通过唤醒一个或多个线程来实现（获取资源）。该方法可以实现方法锁定解锁。
         */
        public final boolean release(int arg) {
        	// 尝试释放
            if (tryRelease(arg)) {
            	// 获取头结点
                Node h = head;
                // 判断头结点状态
                if (h != null && h.waitStatus != 0)
                	// 唤醒节点
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

### tryRelease ###

/**
         * Attempts to set the state to reflect a release in exclusive mode.
         * 尝试将状态设置为独占模式
         */
        protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }

和tryAcquire()一样，这个方法是需要独占模式的自定义同步器去实现的。正常来说，tryRelease()都会成功的，因为这是独占模式，该线程来释放资源，那么它肯定已经拿到独占资源了，直接减掉相应量的资源即可(state-=arg)，也不需要考虑线程安全的问题。但要注意它的返回值，上面已经提到了，release()是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了！所以自义定同步器在实现时，如果已经彻底释放资源(state=0)，要返回true，否则返回false。

### unparkSuccessor ###

/**
         * Wakes up node's successor, if one exists.
         * 唤醒节点的后续节点（如果存在）。
         */
        private void unparkSuccessor(Node node) {
            // 如果状态为负（即可能需要信号），则尝试在预期信号时清除。如果失败或者状态被等待线程更改，这是正常的。
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
            // 唤醒的线程保存在后续节点中，后者通常只是下一个节点。
            // 但如果被取消或为空，则从尾部向后移动以找到未取消的节点将其唤醒。
            // 获取当前节点的后继节点
            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            if (s != null)
            	// 唤醒
                LockSupport.unpark(s.thread);
        }

### release总结 ###

先通过tryRelease尝试唤醒，如果唤醒失败通过unparkSuccessor唤醒，当前节点的后续节点，如果后续节点不符合，直到找到合适的为止。

## acquireShared ##

/**
         * 在共享模式下获取，同时忽略中断。
         */
        public final void acquireShared(int arg) {
        	// 尝试用tryAcquireShared获取
            if (tryAcquireShared(arg) < 0)
            	// 通过doAcquireShared获取
                doAcquireShared(arg);
        }

### tryAcquireShared ###

/**
         * Attempts to acquire in shared mode. This method should query if
         * the state of the object permits it to be acquired in the shared
         * mode, and if so to acquire it.
         * 尝试在共享模式下获取。这个方法应该查询对象的状态是否允许在共享模式下获取它，
         * 以及是否允许在共享模式下获取它。
         */
        protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }

这里tryAcquireShared()依然需要自定义同步器去实现。但是AQS已经把其返回值的语义定义好了：负值代表获取失败；0代表获取成功，但没有剩余资源；正数表示获取成功，还有剩余资源，其他线程还可以去获取。

### doAcquireShared ###

/**
         * Acquires in shared uninterruptible mode.
         * 以共享不间断模式获取。
         */
        private void doAcquireShared(int arg) {
        	// 用当前线程，以及共享模式创建node节点
            final Node node = addWaiter(Node.SHARED);
            // 是否成功标志
            boolean failed = true;
            try {
            	// 等待过程中是否被中断过的标志
                boolean interrupted = false;
                // 自旋
                for (;;) {
                	// 获取前驱节点
                    final Node p = node.predecessor();
                    // 判断是不是头节点
                    if (p == head) {
                    	// 尝试获取资源
                        int r = tryAcquireShared(arg);
                        // >=0 代表成功
                        if (r >= 0) {
                        	// 设置头节点和释放
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            // 判断中断
                            if (interrupted)
                            	// 中断当前线程
                                selfInterrupt();
                            failed = false;
                            return;
                        }
                    }
                    // 如果自己可以休息了，就进入waiting状态，直到被unpark()
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        // 如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为true
                        interrupted = true;
                }
            } finally {
                if (failed)
                	// 取消中断
                    cancelAcquire(node);
            }
        }

### setHeadAndPropagate ###

/**
         * Sets head of queue, and checks if successor may be waiting
         * in shared mode, if so propagating if either propagate > 0 or
         * PROPAGATE status was set.
         * 设置队列头，并检查后续任务是否在共享模式下等待，
         * 如果是释放状态，则在设置了释放>0或释放状态时设置
         */
        private void setHeadAndPropagate(Node node, int propagate) {
        	// 获取头节点
            Node h = head; // Record old head for check below
            // 设置头节点
            setHead(node);
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }

### doReleaseShared ###

/**
         * 共享模式的释放操作——信号继承并确保传播。
         * （注意：对于独占模式，如果需要信号，释放就等于调用head的唤醒后继。）
         */
        private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    // 如果头结点状态是signal，即需要唤醒后继节点
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    // 如果头结点状态是0且CAS成功状态重置为传播失败了，退出当前循环进入下次循环
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

### acquireShared总结 ###

tryAcquireShared()尝试获取资源，成功则直接返回；  
失败则通过doAcquireShared()进入等待队列park()，直到被unpark()/interrupt()并成功获取到资源才返回。整个等待过程也是忽略中断的。

## releaseShared ##

/**
         * Releases in shared mode.  Implemented by unblocking one or more
         * threads if {@link #tryReleaseShared} returns true.
         * 以共享模式发布。如果TryReleaseShared返回true，
         * 则通过取消阻止一个或多个线程来实现。
         */
        public final boolean releaseShared(int arg) {
        	// 尝试以共享方式释放
            if (tryReleaseShared(arg)) {
                doReleaseShared();
                return true;
            }
            return false;
        }