AbstractQueuedSynchronizer

Posted on 2024-03-13 In Java , Java进阶 , 同步&锁&volatile

Java提供了一个基于FIFO队列—AbstractQueuedSynchronizer，可以用于构建锁或者其他相关同步装置的基础框架。该同步器（AQS）利用了一个int来表示状态，期望它能够成为实现大部分同步需求的基础。子类通过继承AQS并需要实现它的方法来管理其状态，管理的方式就是通过类似acquire和 release的方式来操纵状态。然而多线程环境中对状态的操纵必须确保原子性，因此子类对于状态的把握，需要使用这个同步器提供的以下三个方法对状态进行操作：

1
2
3

protected final int getState() {
    return state;
}

1
2
3

protected final void setState(int newState) {
    state = newState;
}

protected final boolean compareAndSetState(int expect, int update) {
    // See below for intrinsics setup to support this
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

AQS的功能可以分为两类：独占功能和共享功能，它的所有子类中，要么实现并使用了它独占功能的API，要么使用了共享锁的功能，而不会同时使用两套 API，即便是它最有名的子类ReentrantReadWriteLock，也是通过两个内部类：读锁和写锁，分别实现的两套API来实现的，为什么这么做，后面我们再分析，到目前为止，我们只需要明白AQS在功能上有独占控制和共享控制两种功能即可。

同步器是实现锁的关键，利用同步器将锁的语义实现，然后在锁的实现中聚合同步器。可以这样理解：锁的API是面向使用者的，它定义了与锁交互的公共行为，而每个锁需要完成特定的操作也是透过这些行为来完成的，但是实现是依托给同步器来完成；同步器面向的是线程访问和资源控制，它定义了线程对资源是否能够获取以及线程的排队等操作。锁和同步器很好的隔离了二者所需要关注的领域，严格意义上讲，同步器可以适用于除了锁以外的其他同步设施上（包括锁）。

同步器的开始提到了其实现依赖于一个FIFO队列，那么队列中的元素Node就是保存着线程引用和线程状态的容器，每个线程对同步器的访问，都可以看做是队列中的一个节点。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;
    static final int CONDITION = -2;
    static final int PROPAGATE = -3;

    volatile int waitStatus;
    volatile Node prev;
    volatile Node next;
    volatile Thread thread;

    Node nextWaiter;

    。。。

waitStatus：表示节点的状态，存在有五种状态。

CANCELLED(1)表示线程已经被取消；

SIGNAL表示当前节点的后继节点包含的线程需要运行，也就是unpark；

CONDITION表示当前节点在等待condition，也就是在condition队列中；

PROPAGATE表示当前场景下后续的acquireShared能够得以执行；

当值为0时，表示当前节点在sync队列中，等待着获取锁。
prev：前驱节点，比如当前节点被取消，那就需要前驱节点和后继节点来完成连接。
next：后继节点。
thread：入队列时的当前线程。
nextWaiter：存储condition队列中的后继节点。

AbstractQueuedSynchronizer定义如下：

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

    private transient volatile Node head;

    private transient volatile Node tail;

    private volatile int state;
    
    。。。

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

class Mutex implements Lock, java.io.Serializable {
    // 内部类，自定义同步器
    private static class Sync extends AbstractQueuedSynchronizer {

        // 是否处于占用状态
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        // 当状态为0的时候获取锁
        public boolean tryAcquire(int acquires) {
            assert acquires == 1; // Otherwise unused
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        // 释放锁，将状态设置为0
        protected boolean tryRelease(int releases) {
            assert releases == 1; // Otherwise unused
            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为例子，详细分析以下同步器的实现逻辑。

源码分析

获取锁过程：

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

尝试获取锁，在tryAcquire方法中使用了同步器提供的对state操作的方法，利用compareAndSet保证只有一个线程能够对状态进行成功修改，如果获取不到，将当前线程构造成节点Node并加入sync队列。

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode); //构造Node
    //快速尝试在尾部添加
    Node pred = tail;
    if (pred != null) { //存在尾节点
        node.prev = pred; //新节点前驱节点指向尾节点
        if (compareAndSetTail(pred, node)) { //使自己成为tail
            pred.next = node; //原有的尾节点的后继节点指向自己
            return node;
        }
    }
    enq(node);//if块操作失败则使用该方式在尾部添加
    return node;
}

final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//获取前驱节点
            if (p == head && tryAcquire(arg)) { 
//前驱节点是head，并尝试获取到了锁
                setHead(node); //设置头结点head为当前节点
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
//检查前一个节点的状态，看当前获取锁失败的线程是否需要挂起(Node.SIGNAL时返回true；
////如果需要，借助JUC包下的LockSopport类的静态方法Park挂起当前线程。)。
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node); //取消请求，对应到队列操作，就是将当前节点从队列中移除。
    }
}

释放锁过程：

public final boolean release(int arg) {
    if (tryRelease(arg)) { //尝试释放状态
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h); //唤醒当前节点的后继节点所包含的线程
        return true;
    }
    return false;
}

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;
//注意是从队列尾部开始往前去找的最前面的一个waitStatus小于0的节点。
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null) 
        LockSupport.unpark(s.thread);// 唤醒线程
}

参看：
http://ifeve.com/introduce-abstractqueuedsynchronizer/
http://ifeve.com/jdk1-8-abstractqueuedsynchronizer/