Java AQS的实现原理详解

目录

• 使用
• lock
• Sync
• acquire
• NonfairSync.tryAcquire
• FairSync.tryAcquire
• acquireQueued
• acquireQueued
• unlock
• cancelAcquire

使用

我们这里借助ReentrantLock来搞清楚AQS的实现原理。

lock

这个方法就是开始获取锁运行的入口，在这个方法的实现中，交给了sync对象来获取锁。

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

private final Sync sync;
// Sync对象是一个ReentrantLock实现的内部抽象类，具体的实现又分为了公平版本与非公平两种
abstract static class Sync extends AbstractQueuedSynchronizer {}

// 在ReentrantLock的无参构造器中，默认使用的实现就是非公平锁的实现
public ReentrantLock() {
    sync = new NonfairSync();
}

// 也可以通过带参数的构造器来使用公平锁
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

Sync

由于公平锁FairSync和NonfairSync的差别主要在tryAcquire方法上，别的逻辑都是相同的，因此我们就直接看Sync和AQS中的实现。

acquire

方法实现如下，来自AQS的实现：

// 首先会调用 tryAcquire 和 acquireQueued 方法，如果2个方法都返回true的话，
// 那么才会调用自行中断的逻辑
if (!tryAcquire(arg) &&
    acquireQueued(addwaiter(Node.EXCLUSIVE), arg))
    selfInterrupt();

tryAcquire方法就会因为公平锁和非公平锁的差异，有2种不同的实现，首先来看看非公平锁的实现，也就是ReentrantLock的默认策略。

NonfairSync.tryAcquire

这个方法会直接调用并返回 Sync 实现的 nonfairTryAcquire(acquires)方法。

Sync类中的实现

// 这里的参数 acquires = 1
final boolean nonfairTryAcquire(int acquires) {
    // 获取当前调用者的线程对象
    final Thread current = Thread.currentThread();
    // 获取AQS中定义的state值，这个state值是AQS的核心之一
    int c = getState();
    // 在ReentrantLock的实现中，state就表示当前是否有线程持有锁，0代表没有线程持有锁，
    // 当前访问的线程就可以继续执行代码，如果大于0则表示当前持有锁的线程的数量。
    // 由于ReentrantLock属于可重入锁，因此，这个值会>=1
    if (c == 0) {
        // 能进来就表示当前没有线程持有锁，那么尝试用CAS获取锁
        if (compareAndSetState(0, acquires)) {
            // 获取锁成功，那么将当前线程设置到AQS中的当前线程中
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    // 如果当前持有锁的就是自己，那么就代表是锁的重入
    else if (current == getExclusiveOwnerThread()) {
        // 累计持有锁的次数
        int nextc = c + acquires;
        // 这里就说明了，state能够设置的最大值就是Int.MAX_VALUE，
        // 当处于MAX_VALUE的时候再加1，那么Int数字的最高位就会变成1，符号位为负
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        // 更新新的state值
        setState(nextc);
        return true;
    }
    return false;
}

AQS中的实现

private volatile int state;
// 当前持有锁的线程对象
private transient Thread exclusiveOwnerThread;

protected final void setExclusiveOwnerThread(Thread thread) {
    exclusiveOwnerThread = thread;
}

小结一下，非公平锁tryAcquire方法就是先看看有没有线程持有锁，没有的话自己就通过CAS的方式尝试获取一下锁，如果获取锁成功或者是自己重入，那么tryAcquire方法就会返回true，acquire方法中的条件判断就会直接返回false，lock方法结束，线程继续支持下面的代码。

FairSync.tryAcquire

下面来看看公平锁的实现，大体的逻辑跟非公平的是相同的。

FairSync中的实现

// 这里的参数 acquire = 1
protected final boolean tryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    // 判断当前是不是有线程持有锁
    if (c == 0) {
        // 当前没有线程持有锁就进来
        // 由于是公平锁，那么就要保证只有在当前等待队列为空或者队列中等待的线程
        // 都没有到运行的条件的时候，才尝试通过CAS来获取锁。否则就去乖乖排队
        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;
}

AQS中的实现

// 检查当前AQS等待队列中是否有正在等待的有效线程节点
public final boolean hasQueuedPredecessors() {
    Node h, s;
    // 首先让参数h指向当前队列的头部
    if ((h = head) != null) {
        // 队列不为空
        // 将临时变量赋值为当前第二个节点
        // 这里需要简单说明一下AQS的等待队列的构成，第一个节点是没有业务含义的，
        // 只是用作唤醒下一个待执行的线程节点
        if ((s = h.next) == null || s.waitStatus > 0) {
            // 能进来就表示当前队列只有一个头结点，或者第二个节点的状态是已取消
            // - > 参考说明1
javascript            // 如果第二个节点不为空，那么就释放这个引用
            s = null; // traverse in case of concurrent cancellation
            // 从后往前遍历，找到距离队列头最近的有效节点
            for (Node p = tail; p != h && p != null; p = p.prev) {
                if (p.waitStatus <= 0)
                    s = p;
            }
        }
        // 如果找到了正在队列中的排队的有效节点并且不是当前访问的线程，那么就返回true
        if (s != null && s.thread != Thread.currentThread())
            return true;
    }
    // 头结点指向NULL，那么说明队列是空的，直接返回false
    return false;
}

说明1：这里就引入了队列节点中的等待状态这个重要的概念，在ReentrantLock中，我们只需要关注CANCELLED和SIGNAL即可。只有CANCELLED是大于0的，新节点的默认值为0。因此只要等待状态大于0就代表该节点被取消了。

// 等待队列中节点的等待状态
volatile int waitStatus;
// 当前节点因为等待超时或者被中断了被取消
static final int CANCELLED =  1;
// 接下来有资格被唤醒获得锁的标记，只有获得了这个标记的节点才能被执行完的线程唤醒
static final int SIGNAL    = -1;

小结一下，公平锁对比非公平锁，在最开始有机会获取锁的时候，会先检查一下当前队列中是否已经有线程在排队等待执行了，如果等待队列中是空的或者没有有效的排队节点，才会获取锁。如果获取锁成功，或者锁重入成功，那么同样会结束AQS的逻辑，继续执行业务代码。

acquireQueued

上面分析完在tryAcquire方法中如果成功获得锁的情况，就会结束AQS的逻辑，接下来就来分析未能成功获得锁的逻辑，即：acquire方法中条件判断的第二个条件判断。

在AQS中实现

// 这个方法就会将当前线程添加到等待队列中，并且返回操作是否成功，arg就是传入的acquire值，为1
acquireQueued(addWaiter(Node.EXCLUShttp://www.devze.comIVE), arg)

首先通过addWaiter方法构建一个包含线程对象的节点并且添加到队列中

// 这里的参数为 Node.EXCLUSIVE，表示这是一个排它锁的实现，这里的值为NULL
private Node addWaiter(Node mode) {
    // 构建一个线程节点
    Node node = new Node(mode);

    // 这里就是AQS的核心理念了，通过不断的自旋，将线程节点插入到队列中
    for (;;) {
        // 获取原来的队列的队尾，因为AQS才去的尾插方式
        Node oldTail = tail;
        if (oldTail != null) {
            // 将新插入的节点指向原来的尾结点
            node.setPrevRelaxed(oldTail);
            // 通过CAS的方式将当前节点设置到线程共享队列的尾部去，这里要注意，
            // 凡是涉及到多线程操作的属性，都需要通过CAS保证操作的原子性
            if (compareAndSetTail(oldTail, node)) {
                oldTail.next = node;
                // 设置成功，就返回插入的节点对象；如若不成功，
                // 就表示有别的线程也修改了尾结点，那么就要等下一次循环重试
                return node;
            }
        } else {
            // 没有尾结点说明队列不存在，那么就进行初始化
            initializeSyncQueue();
     开发者_C教程   }
    }
}

// 初始化等待队列
private final void initializeSyncQueue() {
    Node h;
    // 还是通过CAS的方式，给队列初始化一个默认的Node节点，几个重要的属性的初始值如下
    // waitStatus = 0; thread = null
    if (HEAD.compareAndSet(this, null, (h = new Node())))
        tail = h;
}

小结一下，addWaiter方法通过尾插的方式将没有抢到锁的线程封装为Node节点，插入到AQS的等待队列中。如果队列还未初始化，那么就先初始化队列，队列自带一个无实际含义的头结点。

acquireQueued

在AQS中实现

// arg就是传入的acquire参数，为1；该方法返回值的含义为，线程在等待过程中是否被中断
final boolean acquireQueued(final Node node, int arg) {
    boolean interrupted = false;
    try {
        // 还是不断的自旋，等待机会进行操作
        for (;;) {
            // 获取新插入节点的上一个节点
            final Node p = node.predecessor();
          php  // 如果上一个节点已经是头结点，那么说明当前就已经轮到当前线程获取锁执行业务了
            // 那么就再次尝试抢一次锁
            if (p == head && tryAcquire(arg)) {
                // 能进来就说明获取锁成功了
                // 那么就将当前线程的节点设置为头结点
                // 在这个方法中，会去除节点中的信息，做一个纯粹的头结点
                setHead(node);
                // 将已经没有指向的原头结点的next指为空，等待回收
                p.next = null; // help GC
                // 这里返回的值为false，因为当前线程并未被阻塞就获得了锁
                return interrupted;
            }
            // 走到这里说明当前线程并没有获取到锁，那么就要考虑是否要将线程阻塞了
            if (shouldParkAfterFailedAcquire(p, node))
                interrupted |= parkAndCheckInterrupt();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        if (interrupted)
            selfInterrupt();
        throw t;
    }
}

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

// 线程获取锁失败之后，是否需要将线程阻塞，这里2个参数，
// pred是新插入节点的上一个节点，node是新插入的节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取上一个节点的等待状态
    int ws = pred.waitStatus;
    // 判断上一个节点的状态是不是SIGNAL，只有状态为SIGNAL的才是有效的可指向节点
    if (ws == Node.SIGNAL)
        // 如果上一个节点是SIGNAL状态，那就说明当前线程可以连接上该节点，然后被挂起了
        return true;
    // 上面我们提到过，只有节点被取消了，等待状态才会>0
    if (ws > 0) {
        // 一直往前找，直到找到等待状态<=0的，数据规范的话即找到最近的一个等待状态
        // 为SIGNAL的节点，跳过全部被取消的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        // 此时pred指向的即第一个合法的线程节点，指向当前新插入的节点
        pred.next = node;
    } else {
        // 这里的意思就是上一个节点就是有效节点，那么就将上一个节点的等待状态强制
        // 更新为SIGNAL，即-1。毫无意义的那个头结点也会被设置为SIGNAL状态
        pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
    }
    return false;
}

小结一下，这个方法的含义就是，由于当前线程没有获取到锁资源，因此就需要被阻塞。同时在这个方法中，也会整理等待队列，将那些已经被取消的节点从队列中移除。接MrcCnVu着就是调用条件判断中的执行方法，将线程挂起阻塞起来。

private final boolean parkAndCheckInterrupt() {
    // 调用了park方法，底层是调用UnSafe类的park方法来实现
    LockSupport.park(this);
    return Thread.interrupted();
}

至此，lock方法的全部情况都清楚了，如果线程能拿到锁，那就直接结束lock阶段，要是抢不到锁，那么就进入等待队列中php，在进入队列之前，如果发现还有机会获取锁，那么会再次尝试获取一次，如若还是获取不到，那么就以尾插的方式进入等待队列，通过调用LockSupport.park方法，将线程阻塞，等待被唤醒。

unlock

主动调用这个方法以后，就代表对于锁的占用结束。

在ReentrantLock中实现

public void unlock() {
    sync.release(1);
}

在AQS中实现

public final boolean release(int arg) {
    // 调用tryRelease方法真正实现解除锁
    // 只有state加的所有锁被解除了，那么才会唤醒下一个线程
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

在Sync中实现

// 这里的releases就是传入的参数1，即如果是重入锁，那么这里需要解锁多次
protected final boolean tryRelease(int releases) {
    // 计算state的值，这里的含义就是state-1
    int c = getState() - releases;
    // 如果当前线程不是持有锁的线程，那么就报错
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    // 锁是否完全释放完的标记
    boolean free = false;
    // state减到0说明锁已经完全释放完了
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    // 更新state的值
    setState(c);
    // 只有锁被完全释放完，才返回true
    return free;
}

在AQS中实现

// 唤醒下一个需要锁的线程，这里的node是头结点
private void unparkSuccessor(Node node) {
    // 获得头结点的等待状态
    int ws = node.waitStatus;
    // 如果头结点是SIGNAL，那么重置为0，因为这个节点已经没有意义了，会被移除
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);
    // 获得头结点后面的待唤醒的节点
    Node s = node.next;
    // 如果这个待唤醒的节点为空或者等待状态不正确，在这里就是不等于SIGNAL
    if (s == null || s.waitStatus > 0) {
        s = null;
        // 从尾部开始查询，找到合法的待唤醒节点
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    if (s != null)
        // 唤醒线程
        LockSupport.unpark(s.thread);
}

cancelAcquire

这个方法提供了取消线程等待获取锁的功能

在AQS中被实现

private void cancelAcquire(Node node) {
    // 健壮性判断，节点非空才可以被取消
    if (node == null)
        return;

    // 将节点中的线程指向去掉
    node.thread = null;

    // 获取到当前节点的上一个节点
    Node pred = node.prev;
    // 通过循环，跳过所有当前被取消节点之前的也已经被标记取消的节点
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;

    // 通过上面的循环以后pred的值就为等待队列中队尾的一个合法的未被取消的节点
    // 获取到该合法节点下一个指向的节点
    Node predNext = pred.next;

    // 标记当前被取消的节点的等待状态为被取消
    node.waitStatus = Node.CANCELLED;

    // 如果被取消的节点就是队尾的节点，那么就通过CAS将pred设置为尾结点
    // 就可以抛弃中间那些同样被标记为被取消的节点，如果有的是
    if (node == tail && compareAndSetTail(node, pred)) {
        // 将pred的下一个节点指向NULL，因为pred现在就是队尾
        pred.compareAndSetNext(predNext, null);
    } else {
        // 这说明取消的节点不是尾结点，而是中间的节点
        // 这个值会在下面的条件判断被赋值为上一个节点的等待状态
        int ws;
        // 如果找到的上一个合法节点不是头结点
        // 并且上一个节点的等待状态是SIGNAL，并且将不是SIGNAL状态的负数状态转换为SIGNAL
        // 并且线程不为空
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
            pred.thread != null) {
            // 能进来就说明被取消的节点处于中间，那么就要将这个node从队列中跳过
            Node next = node.next;
            // 如果被取消的节点是一个有效的节点，不为空并且状态也是对的
            if (next != null && next.waitStatus <= 0)
                // 那么就将上一个节点指向被取消节点的下一个节点
                pred.compareAndSetNext(predNext, next);
        } else {
            // 能进入这里说明上一个合法节点已经是头结点了，
            // 那么就说明被取消的这个节点已经是原本除了头结点以外的最靠前面的节点，
            // 那么被取消的这个节点其实就等价于头结点了，应该唤醒后面还在等待的线程节点
            // 唤醒下一个被挂起的线程，具体已经分析过了，这里就省略了
            unparkSuccessor(node);
        }

        node.next = node;
    }
}

以上就是Java AQS的实现原理详解的详细内容，更多关于Java AQS的资料请关注我们其它相关文章！