AQS原理及源码分析

作者: 坤坤坤坤杨 | 来源:发表于2022-12-01 19:50 被阅读0次

AQS是一个用来构建锁和同步器的框架，使用AQS能简单且高效地构造出应用广泛的大量的同步器，比如ReentrantLock，Semaphore，其他的诸如ReentrantReadWriteLock，SynchronousQueue，FutureTask等等皆是基于AQS的。

1. AQS核心思想

AQS核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态。如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制AQS是用CLH队列锁实现的，即将暂时获取不到锁的线程加入到队列中。

CLH(Craig,Landin,and Hagersten)队列是一个虚拟的双向队列(虚拟的双向队列即不存在队列实例，仅存在结点之间的关联关系)。AQS是将每条请求共享资源的线程封装成一个CLH锁队列的一个结点(Node)来实现锁的分配。其中Sync queue，即同步队列，是双向链表，包括head结点和tail结点，head结点主要用作后续的调度。而Condition queue不是必须的，其是一个单向链表，只有当使用Condition时，才会存在此单向链表。并且可能会有多个Condition queue。

AQS使用一个int成员变量来表示同步状态，通过内置的FIFO队列来完成获取资源线程的排队工作。AQS使用CAS对该同步状态进行原子操作实现对其值的修改。

private volatile int state; // 共享变量，使用volatile修饰保证线程可见性

状态信息通过procted类型的getState，setState，compareAndSetState进行操作

//  返回同步状态的当前值
protected final int getState() {  
        return state;
}
 // 设置同步状态的值
protected final void setState(int newState) { 
        state = newState;
}
//原子地(CAS操作)将同步状态值设置为给定值update如果当前同步状态的值等于expect(期望值)
protected final boolean compareAndSetState(int expect, int update) {
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

2. AQS对资源共享的方式

AQS定义两种资源共享方式

Exclusive(独占)：只有一个线程能执行，如ReentrantLock。又可分为公平锁和非公平锁：
- 公平锁：按照线程在队列中的排队顺序，先到者先拿到锁
- 非公平锁：当线程要获取锁时，无视队列顺序直接去抢锁，谁抢到就是谁的
Share(共享)：多个线程可同时执行，如Semaphore/CountDownLatch。Semaphore、CountDownLatCh、 CyclicBarrier、ReadWriteLock 。

ReentrantReadWriteLock 可以看成是组合式，因为ReentrantReadWriteLock也就是读写锁允许多个线程同时对某一资源进行读。

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

3. AQS使用的设计模式

同步器的设计是基于模板方法模式的，使用者继承AbstractQueuedSynchronizer并重写指定的方法。(这些重写方法很简单，无非是对于共享资源state的获取和释放) 将AQS组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。以下是AQS提供的可重写的方法：

protected boolean isHeldExclusively()；//该线程是否正在独占资源。只有用到condition才需要去实现它。
protected boolean tryAcquire(int)；//独占方式。尝试获取资源，成功则返回true，失败则返回false。
protected boolean tryRelease(int)；//独占方式。尝试释放资源，成功则返回true，失败则返回false。
protected int tryAcquireShared(int)；//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
protected boolean tryReleaseShared(int)；//共享方式。尝试释放资源，成功则返回true，失败则返回false。

4. Node类

该类为AbstractQueuedSynchronizer的一个静态内部类，每个被阻塞的线程都会被封装成一个Node结点，放入队列。

static final class Node {
    // 模式，分为共享与独占
    // 共享模式
    static final Node SHARED = new Node();
    // 独占模式
    static final Node EXCLUSIVE = null;        
    // 结点状态常量
    static final int CANCELLED =  1; // 表示当前的线程被取消，线程被中断的情况
    static final int SIGNAL    = -1; //表示当前节点的后继节点包含的线程需要运行，也就是需要去unpark后面的节点
    static final int CONDITION = -2; //表示当前节点在等待condition，也就是在condition队列中
    static final int PROPAGATE = -3; //表示当前场景下后续的acquireShared能够得以执行

    // 结点状态
    volatile int waitStatus;   //  值为0，表示当前节点在sync队列中，等待着获取锁，后面已经没有其他节点了，就不用去unPark（）
    // 前驱结点
    volatile Node prev;    
    // 后继结点
    volatile Node next;        
    // 结点所对应的线程
    volatile Thread thread;        
    // 下一个等待者
    Node nextWaiter;
    
    // 结点是否在共享模式下等待
    final boolean isShared() {
        return nextWaiter == SHARED;
    }
    
    // 获取前驱结点，若前驱结点为空，抛出异常
    final Node predecessor() throws NullPointerException {
        // 保存前驱结点
        Node p = prev; 
        if (p == null) // 前驱结点为空，抛出异常
            throw new NullPointerException();
        else // 前驱结点不为空，返回
            return p;
    }
    
    // 无参构造方法
    Node() {    // 由addWaiter使用用于建立初始标头或SHARED标记
    }
    
    // 构造方法
        Node(Thread thread, Node mode) {    // 由addWaiter使用
        this.nextWaiter = mode;
        this.thread = thread;
    }
    
    // 构造方法
    Node(Thread thread, int waitStatus) { // 由Condition使用
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

5. AbstractQueuedSynchronizer类属性

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {    
    // 版本号
    private static final long serialVersionUID = 7373984972572414691L;    
    // 头节点
    private transient volatile Node head;    
    // 尾结点
    private transient volatile Node tail;    
    // 状态
    private volatile int state;    
    // 自旋时间
    static final long spinForTimeoutThreshold = 1000L;
    
    // Unsafe类实例
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    // state内存偏移地址
    private static final long stateOffset;
    // head内存偏移地址
    private static final long headOffset;
    // state内存偏移地址
    private static final long tailOffset;
    // tail内存偏移地址
    private static final long waitStatusOffset;
    // next内存偏移地址
    private static final long nextOffset;
    // 静态初始化块
    static {
        try {
            //unsafe.objectFieldOffset();用于获取某个字段相对Java对象的“起始地址”的偏移量
            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); }
    }
}

6. 核心方法

6.1 acquire方法（加锁）

该方法以独占模式获取(资源)，忽略中断，即线程在acquire过程中，中断此线程是无效的。源码如下:

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

调用流程如下：

首先调用tryAcquire方法，调用此方法的线程会试图在独占模式下加锁，如果加锁成功什么都不做，如果加锁失败，那么!tryAcquire(arg)=true，说明此时的锁被其他线程所占有。在AbstractQueuedSynchronizer源码中默认会抛出一个异常，即需要子类去重写此方法完成自己的逻辑。
若tryAcquire失败，则调用addWaiter方法，addWaiter方法完成的功能是将调用此方法的线程封装成为一个结点并放入同步队列。
调用acquireQueued方法，此方法完成的功能是同步队列中的结点不断尝试获取锁，若成功，则返回true，否则，返回false。
如果if条件满足就会被中断。

addWaiter方法

// 添加等待者
private Node addWaiter(Node mode) {
    // 新生成一个结点，默认为独占模式
    Node node = new Node(Thread.currentThread(), mode);
    // 获取尾节点
    Node pred = tail;
    if (pred != null) { // 尾结点不为空，即已经被初始化，队列中是有元素的
        // 将新结点的prev连接到尾结点，也就是将新的节点加入到队列的后面
        node.prev = pred; 
        if (compareAndSetTail(pred, node)) { //为了解决有多个线程同时进入这个判断逻辑生成了多个节点，保证只能有一个节点成为队列的尾部。
            // 
            pred.next = node;
            return node; // 返回新生成的结点
        }
    }
    enq(node); 
// 如果尾结点为空(队列中没有元素)，或者是compareAndSetTail操作失败，则入队列
// 对于公平锁而言，没有竞争到队列的尾部，那么就一直去竞争直到竞争到。
    return node;
}

enq(node)

假设此时有两个线程：线程1、线程2进来。两个线程都在上面代码中判断if (pred != null)都为空则进入到enq方法。

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // 两个线程进来发现尾节点为空，那么就生成一个空的node节点，再次循环线程1和线程2进来判断不为空走else逻辑
                if (compareAndSetHead(new Node()))
                    tail = head; //头节点与尾结点都指向同一个新生结点(空节点)
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) { //两个线程再进来竞争尾节点，将属性指向空的node节点。
                    t.next = node;
                    return t; //假设线程2竞争到尾节点
                }
            }
        }
    }

acquireQueued方法

此时线程2进到这个方法

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();//获取当前节点的prev前一个节点
                if (p == head && tryAcquire(arg)) {//如果前一个节点等于head节点，那么说明它就是第一个排队的线程
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
              //parkAndCheckInterrupt ：阻塞
              // shouldParkAfterFailedAcquire： 修改上一个node节点的waitStatus为-1
                if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

tryAcquire方法

以ReentrantLock公平锁为例子，下面是ReentrantLock内部lock（）方法逻辑：

 final void lock() {
            acquire(1);
        }

//尝试获取锁
 protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();//可重入条件
            if (c == 0) { //表示第一次获取到锁
              //compareAndSetState(0, acquires)
              //利用 cas 将state状态改为 1
                if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current); // 设置锁被当前线程独占，也就是真正的加锁操作
                    return true;
                }
            }
            //重入逻辑
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }
    }

$\color{red}{判断是否有正在排队的线程（公平锁）}$

 public final boolean hasQueuedPredecessors() {
        Node t = tail;
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

AQS原理及源码分析

1. AQS核心思想

2. AQS对资源共享的方式

3. AQS使用的设计模式

4. Node类

5. AbstractQueuedSynchronizer类属性

6. 核心方法

6.1 acquire方法（加锁）

addWaiter方法

enq(node)

acquireQueued方法

tryAcquire方法

$\color{red}{判断是否有正在排队的线程（公平锁）}$

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读