Java中的锁（3）同步器AQS （AbstractQueuedSynchronizer）

雨点打透心脏的1/2处 2022-06-07 06:16 129阅读 0赞

看了两天AQS CAS的内容，终于弄清楚了。下文多数内容来自《Java并发编程的艺术》

在java.util.concurrent.locks包中有很多Lock的实现类，常用的有ReentrantLock、ReadWriteLock（实现类ReentrantReadWriteLock），内部实现都依赖AbstractQueuedSynchronizer类。

AQS它使用了一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作，并发包的作者（Doug Lea）期望它能够成为实现大部分同步需求的基础。

--------------------

## **如何使用AQS来实现一个锁** ##

同步器是实现锁的关键，利用同步器将锁的语义实现，然后在锁的实现中聚合同步器。

我们可以这样理解：锁的API是面向使用者的，它定义了与锁交互的公共行为，而每个锁需要完成特定的操作也是透过这些行为来完成的（比如：可以允许两个线程进行加锁，排除两个以上的线程），但是实现是依托给同步器来完成；同步器面向的是线程访问和资源控制，它定义了线程对资源是否能够获取以及线程的排队等操作。

重写同步器指定的方法时，需要使用同步器提供的如下3个方法来访问或修改同步状态。

> getState()：获取当前同步状态。  
> setState(int newState)：设置当前同步状态。  
> compareAndSetState(int expect,int update)：使用CAS设置当前状态，该方法能够保证状态设置的原子性。

以ReentrantLock为例，state初始化为0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到A线程unlock()到state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A线程自己是可以重复获取此锁的（state会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证state是能回到零态的。

同步器可重写的方法  
![这里写图片描述][SouthEast]

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

## 例子 ##

下面通过一个排它锁的例子来深入理解一下同步器的工作原理，而只有掌握同步器的工作原理才能够更加深入了解其他的并发组件。

排他锁的实现，一次只能一个线程获取到锁。

class Mutex implements Lock, java.io.Serializable {
    
       // 静态内部类，自定义同步器
       private static class Sync extends AbstractQueuedSynchronizer { 
           // 是否处于占用状态，状态为1就返回true，处于占用状态，为0就返回false，非占用状态
           protected boolean isHeldExclusively() {
                   return getState() == 1;
           }
    
           // 当状态为0的时候获取锁，通过CAS指令修改状态变量state。
           public boolean tryAcquire(int acquires) {
               if (compareAndSetState(0, 1)) {
                   setExclusiveOwnerThread(Thread.currentThread());
                   return true;
               }
               return false;
           }
    
           // 释放锁，将状态设置为0
           protected boolean tryRelease(int releases) {
               if (getState() == 0) throw new IllegalMonitorStateException();
               setExclusiveOwnerThread(null);
               setState(0);
               return true;
           }
    
           // 返回一个Condition，每个condition都包含了一个condition队列
           Condition newCondition() { return new ConditionObject(); }
       }
    
       // 仅需要将操作代理到Sync上即可
       private final Sync sync = new Sync();
       public void lock()                { sync.acquire(1); }
       public boolean tryLock()          { return sync.tryAcquire(1); }
       public void unlock()              { sync.release(1); }
       public Condition newCondition()   { return sync.newCondition(); }
       public boolean isLocked()         { return sync.isHeldExclusively(); }
       public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
       public void lockInterruptibly() throws InterruptedException {
         sync.acquireInterruptibly(1);
       }
       public boolean tryLock(long timeout, TimeUnit unit)
           throws InterruptedException {
         return sync.tryAcquireNanos(1, unit.toNanos(timeout));
       }
     }

可以看到Mutex将Lock接口均代理给了同步器的实现。

使用者将Mutex构造出来之后，调用lock获取锁，调用unlock进行解锁。Mutex中定义了一个静态内部类，该内部类继承了同步器并实现了独占式获取和释放同步状态。在tryAcquire(int acquires)方法中，如果经过CAS设置成功（同步状态设置为1），则代表获取了同步状态，而在tryRelease(int releases)方法中只是将同步状态重置为0。

在Mutex的实现中，以获取锁的lock()方法为例，只需要在方法实现中调用同步器的模板方法**acquire(int args)**即可，当前线程调用该方法获取同步状态失败后会被加入到同步队列中等待，这样就大大降低了实现一个可靠自定义同步组件的门槛。

在上面的例子中，我们只要使用同步器提供的getState(),setState(),compareAndSetState()这三个方法来重写tryAcquire(int acquires)和tryRelease(int releases)，compareAndSetState(int expect,int update)这三个方法，就可以直接使用同步器提供的acquire()，release()等方法了。再把这些方法代理给Lock()方法，就可以使用了！

> 所以说，同步器的设计是基于模版方法模式的，使用者需要**继承同步器并重写指定的方法**，随后将同步器组合在自定义同步组件的实现中，并调用同步器提供的模板方法，而这些模板方法将会调用使用者重写的方法。

同步器提供的模版方法  
![这里写图片描述][SouthEast 1]

--------------------

## **AQS的实现原理** ##

同步器依赖内部的同步队列（一个FIFO双向队列）来完成同步状态的管理，当前线程获取同步状态失败时，同步器会将当前线程以及等待状态等信息构造成为一个节点（Node）并将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，会把首节点中的线程唤醒，使其再次尝试获取同步状态。

**1.获取同步失败的线程会被加入同步队列的尾部**  
同步器拥有首节点（head）和尾节点（tail），没有成功获取同步状态的线程将会成为节点加入该队列的尾部。  
![这里写图片描述][SouthEast 2]  
同步器包含了两个节点类型的引用，一个指向头节点，而另一个指向尾节点。试想一下，当一个线程成功地获取了同步状态（或者锁），其他线程将无法获取到同步状态，转而被构造成为节点并加入到同步队列中，这个过程可能是并发的，而这个加入队列的过程必须要保证线程安全，因此同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Node expect,Node update)，它需要传递当前线程“认为”的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。

**2.成功获取同步状态的线程回从队列头部释放**  
![这里写图片描述][SouthEast 3]  
同步队列遵循FIFO，首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会唤醒后继节点，而后继节点将会在获取同步状态成功时将自己设置为首节点，设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功获取到同步状态，因此设置头节点的方法并不需要使用CAS来保证，它只需要将首节  
点设置成为原首节点的后继节点并断开原首节点的next引用即可。

--------------------

## **独占式同步状态的获取与释放** ##

我们可以知道，同步器提供的加锁模式分为独占式和共享式。AQS的内部类Node定义了两个常量SHARED和EXCLUSIVE，他们分别标识 AQS队列中等待线程的锁获取模式。

> 独占式：独占锁，它在同一时刻只能允许一个线程占有锁，reentrantLock就是独占锁  
> 共享式：共享锁，它在统一时刻允许多个线程占有锁，reentrantReadWriteLock就是共享锁

这里也顺便说一下公平锁与非公平锁

> 公平锁：公平锁遵循先来后到的原则，对于先对锁进行获取的请求一定先被满足  
> 非公平锁：反之，不遵循先来后到的原则。

一般来说，非公平锁的效率比较高，但是公平锁能减少”饥饿“的发生概率。

## acquire()方法 ##

由上面的例子我们知道，通过调用同步器的acquire(int arg)方法可以获取同步状态。

public void lock()                { sync.acquire(1); }

public final void acquire(int arg) {
        if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

上述代码主要完成了同步状态获取、节点构造、加入同步队列以及在同步队列中自旋等待的相关工作。

**其主要逻辑：**  
首先调用自定义同步器实现的tryAcquire(int arg)方法，该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。

**其函数逻辑：**  
1.tryAcquire()尝试直接去获取资源，如果成功则直接返回；  
2.addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；  
3.acquireQueued()使线程在等待队列中获取资源，一直获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。  
4.如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。

## addWaiter()方法 ##

再来分析构造借点以及加入同步队列的addWaiter（）方法。

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
    
         //尝试快速方式直接放到队尾。
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            //使用CAS方法保证线程被正确添加
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
    
        //上一步失败则通过enq入队。
        enq(node);
        return node;
    }
    
    private Node enq(final Node node) {
        //通过“死循环”来保证节点被正确添加
        for (;;) {
            Node t = tail;
                if (t == null) { // 队列为空，创建一个空的标志结点作为head结点，并将tail也指向它。
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else { //正常流程，放入队尾
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
        }
    }

1.通过使用compareAndSetTail(Node expect,Node update)方法来确保节点能够被线程安全添加。

2.在enq(final Node node)方法中，同步器通过“死循环”来保证节点的正确添加，在“死循环”中只有通过CAS将节点设置成为尾节点之后，当前线程才能从该方法返回，否则，当前线程不断地尝试设置。

## acquireQueued()方法 ##

**OK，通过tryAcquire()和addWaiter()，该线程获取资源失败，已经被放入等待队列尾部了。聪明的你立刻应该能想到该线程下一部该干什么了吧：进入等待状态休息，直到其他线程彻底释放资源后唤醒自己，自己再拿到资源，然后就可以去干自己想干的事了。没错，就是这样！是不是跟医院排队拿号有点相似~~acquireQueued()就是干这件事：在等待队列中排队拿号（中间没其它事干可以休息），直到拿到号后再返回。**

节点进入同步队列之后，就进入了一个自旋的过程，每个节点（或者说每个线程）都在自省地观察，当条件满足，获取到了同步状态，就可以从这个自旋过程中退出，否则依旧留在这个自旋过程中（并会阻塞节点的线程）

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)) {
                    setHead(node);
                    p.next = null; // help GC回收
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())
                    interrupted = true;
                }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

1.在acquireQueued(final Node node,int arg)方法中，当前线程在“死循环”中尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态。

2.头节点是成功获取到同步状态的节点，而头节点的线程释放了同步状态之后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节点是否是头节点。  
![这里写图片描述][SouthEast 4]

3.节点和节点之间在循环检查的过程中基本不相互通信，而是简单地判断自己的前驱是否为头节点，这样就使得节点的释放规则符合FIFO，并且也便于对过早通知的处理。

好了，这就是整个acquire()方法的调用流程，我们用一个流程图整理一下。  
![这里写图片描述][SouthEast 5]

## release()方法 ##

线程获取同步状态并执行了相应逻辑之后，就需要释放同步状态，使得后续节点能够继续获取同步状态。通过调用同步器的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;
    }

该方法执行时，会唤醒头节点的后继节点线程，unparkSuccessor(Node node)方法使用LockSupport来唤醒处于等待状态的线程。

**总结：**在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。

[SouthEast]: /images/20220607/6247ee1bf4e142cbae1993bf75f87671.png
[SouthEast 1]: /images/20220607/e0ac98367693425d844c75170877a593.png
[SouthEast 2]: /images/20220607/ff641e2932c34c9eaf4ded51fb6c6166.png
[SouthEast 3]: /images/20220607/7ec4dcfbcd9f447a8816bb65669c1155.png
[SouthEast 4]: /images/20220607/545fccca822545878e29c184c3c738d4.png
[SouthEast 5]: /images/20220607/eb153c916f69435dacefab6fc38a299b.png