什么是AQS？

短命女 2023-10-14 09:23 12阅读 0赞

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

**AQS**中维护了一个**volatile int state**（共享资源）和一个**CLH**队列。当state=1时代表当前对象锁已经被占用，其他线程来加锁时则会失败，失败的线程被放入一个FIFO的等待队列中，然后会被**UNSAFE.park()**操作挂起，等待已经获得锁的线程释放锁才能被唤醒。

我们拿具体场景来分析，假设同时有三个线程并发抢占锁，此时线程一抢占成功，线程二、三抢占失败，具体流程如下：

![format_png][]

此时AQS内部数据结构为：

![format_png 1][]

上图可以看到等待队列中的节点Node是一个双向链表，这里SIGNAL是Node中waitStatus属性。

以非公平锁看下具体实现：

java.util.concurrent.locks.ReentrantLock.NonfairSync：

static final class NonfairSync extends Sync {
            final void lock() {
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);
            }
        }

线程进来直接利用CAS尝试抢占锁，如果抢占成功state值会被修改为1，且设置对象独占锁线程为当前线程。

### 线程抢占实现[\#][Link 1] ###

线程二抢占失败，执行acquire(1)方法。

**java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()**：

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

tryAcquire是AbstractQueuedSynchronizer的方法，未提供对应实现，由子类实现：

**java.util.concurrent.locks.ReentrantLock .nonfairTryAcquire()**：

final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (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;
    }

**nonfairTryAcquire()**方法中首先会获取**state**的值，如果不为0则说明当前对象的锁已经被其他线程占有，接着判断占有锁的线程是否为当前线程，如果是则累加**state**值，这里其实就是可重入锁的具体实现。如果**state**为0，则执行**CAS**操作，尝试更新**state**值为1，如果更新成功则代表当前线程加锁成功。

当前线程二执行tryAcquire()后返回false，接着执行**addWaiter(Node.EXCLUSIVE)**逻辑，将自己加入到一个FIFO等待队列中，代码实现如下：

**java.util.concurrent.locks.AbstractQueuedSynchronizer.addWaiter()**：

private Node addWaiter(Node mode) {    
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

此时队列中**tail**指针为空，直接调用**enq(node)**方法将当前线程加入等待队列尾部：

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) {
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

第一次循环时**tail**为空，创建一个哨兵节点，head指向这个哨兵节点；第二次循环，将线程二对应的node节点挂载到head节点后面并返回当前线程创建的节点信息。继续往后执行**acquireQueued(addWaiter(Node.EXCLUSIVE), arg)**逻辑，此时传入的参数为线程二对应的node节点信息。

**java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued()**：

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)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndChecknIterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

**acquireQueued()**会先判断当前传入的Node对应的前置节点是否为head，如果是则尝试加锁。加锁成功则将当前节点设置为head节点，然后删除之前的head节点。

如果加锁失败或者Node的前置节点不是head节点，就会通过**shouldParkAfterFailedAcquire**方法将head节点的waitStatus变成SIGNAL=-1，最后执行**parkAndChecknIterrupt**方法，调用**LockSupport.park()**挂起当前线程。此时线程二需要等待其他线程释放锁来唤醒。

### 线程释放实现[\#][Link 2] ###

线程一执行完后释放锁，具体代码如下：

**java.util.concurrent.locks.AbstractQueuedSynchronizer.release()：**

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方法，如果执行成功，则继续判断head节点的waitStatus是否为0，这个值为SIGNAL=-1不为0，继续执行**unparkSuccessor()**方法唤醒head的后置节点。

**ReentrantLock.tryRelease()：**

protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

执行完**ReentrantLock.tryRelease()**后，state被设置为0，Lock对象的独占锁被设置为null。

接着执行**java.util.concurrent.locks.AbstractQueuedSynchronizer.unparkSuccessor()**方法，唤醒head的后置节点：

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);
    }

这里主要是将head节点的waitStatus设置为0，然后解除head节点next的指向，使head几点空置，等待被垃圾回收。

此时重新将head指针指向线程二对应的Node节点，且使用**LockSupport.unpark**方法来唤醒线程二。被唤醒的线程会接着尝试获取锁，用CAS指令修改state数据。执行完成后AQS中的数据结构如下：

![format_png 2][]

[format_png]: https://img-blog.csdnimg.cn/img_convert/4e03dc5f73b8a80a8190b639151df5de.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/6d18135ef12b5f7ad6ba0195e5d0c343.webp?x-oss-process=image/format,png
[Link 1]: https://links.jianshu.com/go?to=https%3A%2F%2Fwww.cnblogs.com%2Fcsp1993%2Fp%2F15222511.html%23%25E7%25BA%25BF%25E7%25A8%258B%25E6%258A%25A2%25E5%258D%25A0%25E5%25AE%259E%25E7%258E%25B0
[Link 2]: https://links.jianshu.com/go?to=https%3A%2F%2Fwww.cnblogs.com%2Fcsp1993%2Fp%2F15222511.html%23%25E7%25BA%25BF%25E7%25A8%258B%25E9%2587%258A%25E6%2594%25BE%25E5%25AE%259E%25E7%258E%25B0
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/af4b938c22dc72daf102219c7b1abc29.webp?x-oss-process=image/format,png