其他分享
首页 > 其他分享> > AQS的底层原理探究

AQS的底层原理探究

作者:互联网

java.utils.concurrent 简称 JUC,是jdk跟并发有关的包,AQS在其下的 locks 包下,全称为 AbstractQueuedSynchronizer,是一个抽象类。之所以有名是因为它是很多并发类的基类,最常见的 ReentrantLock 就是基于 AQS。也是因为AQS的存在,我们可以很方便的写出一个自己的并发工具类,只需要自定义如何获取及释放 state 属性即可。

Node节点

这是 AQS 中的内部类,用来形成一条双向队列封装阻塞的线程,其内部的属性名一看便知其意图。

prev   上一个节点
next   下一个节点
thread   当前线程

AQS类维护着 head 和 tail,就是等待队列的头和尾,抓住头尾节点,这条链就算是确定了。除了 node,另外还有一个非常重要的一个属性:state,所以线程靠抢占这个state来取得锁。

acquire 获取锁

这个方法写的很简单,难的都在它调用的方法上。

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

 tryAcquire 自定义获取锁

可以发现这是一个没有代码的方法,不用想就知道等着子类来自定义实现,跟后面释放锁 tryRelease是一样的。

protected boolean tryAcquire(int arg) {
  throw new UnsupportedOperationException();
}

来看一下 ReentrantLock 是如何重写 tryAcquire 抢占锁的。

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();  // 获取state
    if (c == 0) {  // 0 表示锁没有被任何线程占用
        if (compareAndSetState(0, acquires)) {  // 使用cas设置state
            setExclusiveOwnerThread(current);  // 设置成功,则设置锁的占用者为当前线程
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {   // 如果state不为0,先不急,看看占锁的线程是不是当前同一个线程
        int nextc = c + acquires; //如果是,则可以重入。state表示重入的次数
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

可以看到就是通过 cas 方式设置 state 属性来抢占锁;如果state不为0,则先判断当前线程和占用锁的线程是否为同一个:是则重入,否则执行后续addWaiter方法放入等待队列进行阻塞。

是否怀疑为什么只使用cas设置state,设置线程所有者 setExclusiveOwnerThread 为什么就可以直接设置无序考虑线程安全问题?个人认为:

  1. 虽然cas是乐观锁,但是底层调用的是cpu原子指令,尽量减少cas的使用提高性能
  2. 当cas设置完state后,其他线程只会被阻塞到等待队列,这不会造成线程安全问题
  3. 在AQS中有好几处这类的逻辑,看似线程不安全,其实逻辑上没有问题

addWaiter 添加到等待队列

当tryAcquire尝试设置state失败,也就是占用锁失败后,接下来就会把当前线程包装的node放入到等待队列中去。

private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;  // 获取队列的尾节点
    if (pred != null) {  // 队列已有节点,尝试将当前节点插入队列的尾部
        node.prev = pred; // 将当前节点的prev设置为尾节点
        if (compareAndSetTail(pred, node)) { // cas将tail指向当前节点
            pred.next = node;  // 将原来的尾结点的next设置为当前节点
            return node;
        }
    }
    enq(node); // 队列为空或设置尾节点失败走这里
    return node;
}

addWaiter要做的就是把封装了当前线程对象的node节点放在队列的尾部。这个设置需要改动 AQS 的tail属性,所以需要 cas 操作。下面再次看到了上面提到过的做法:没有“将原来的尾结点的next设置为当前node”设置成cas,后面会补上这个小坑。

enq方法

当等待队列为空时,会进入enq方法,会在头节点设置一个没有意义的空节点。想象一下,处在头节点的就是表示线程正在执行的,头节点本来就不需要排队,空节点就很合理。

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // 队列为空
            if (compareAndSetHead(new Node())) //cas设置头节点为空节点
                tail = head;
        } else { // 队列不为空
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}

除了队列为空, 其实 addWaiter 方法中cas设置尾结点 compareAndSetTail 方法失败也会进去enq方法。为什么设置尾结点会失败呢?说明当前线程没有抢到锁,正要将其node往队列的尾部塞的时候,cpu时间片到期,线程上下文切换来到了另外一个线程,它率先一步塞入尾部,那么这个线程就会cas失败。

所以enq才会逻辑判断列表是否会为空,不为空就是上面所说的情形,就会重复 addWaiter 方法中cas添加尾结点的操作。使用for循环就是考虑到上述情形一直发生。

 

acquireQueued 阻塞

添加完队列后,不用想就知道下一步就是阻塞线程啦。在这里会调用当前node的 predecessor 方法获取其前继节点。如果前继节点为是头节点,那么当前节点就是排队等待队列中的第一个,先让它调用 tryAcquire 再次尝试抢占锁。

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) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

shouldParkAfterFailedAcquire 设置前置节点

这里就要说一下节点的等待状态 waitStatus 问题。

所以说等待状态小于等于0才算是正常的。大于0就应该将其剔除出队列。

shouldParkAfterFailedAcquire就是在向上遍历,确保自己的前继节点的等待状态是有效的,不要因为前继节点的状态问题导致自己无法被唤醒。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
        return true;
    if (ws > 0) {
        /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt 阻塞当前线程

做完装备工作,终于线程可以安心睡去了。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

可以看到阻塞非常简单,调用的就是 LockSupport的 park方法,返回的是线程是否被标记打断。如果其他线程想要打断当前线程,那么当前线程的该标记就是true,那么就是走到acquire方法里调用 selfInterrupt 方法,进行自我打断。

static void selfInterrupt() {
    Thread.currentThread().interrupt();
}

interrupted、isInterrupted、interrupt 区别

为什么不使用 stop 方法打断

stop方法因为被jdk认定为不安全的方法,因为stop会立即中止线程的执行,导致线程的执行逻辑不完整。例如转账,小张刚转钱给小明,小张账户的钱少了,小明的账户上的钱还没加,就给stop了,要了命。合理的中断就应该是Interrupt,让线程自己去判断对于中断该如何去响应它。

再看 acquireQueued

这个方法没有想象中的简单。

如果当前节点的前继节点不是头节点,就调用 parkAndCheckInterrupt 方法阻塞其线程。所以所有的线程都不被阻塞在这里,等待释放锁被唤醒。唤醒后又进入这个死循环(就没有出去过),直到其前继节点是头节点并抢锁成功,才跳出这里的循环。

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) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

这里的 Interrupted 是可以取到 true 的,就是在线程将要阻塞的时候,其他线程想要中断它,那么这里返回的就是true,在acquire方法里就不用调用 selfInterrupt 了。

那么什么时候 failed 会取到 true 来执行 cancelAcquire 取消获取锁呢?

找来找去也就在阻塞的时候因为jvm的原因导致阻塞异常,也就是说这是代码健壮性考虑,能到这里的概率很小很小。

cancelAcquire 取消

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;

    node.thread = null;

    // Skip cancelled predecessors
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        //pred = pred.prev;
        //node.prev = pred;
        node.prev = pred = pred.prev;
    //如果node的前驱也是取消节点,则pred.next就不是node
    Node predNext = pred.next;

    node.waitStatus = Node.CANCELLED;

    // If we are the tail, remove ourselves.
    //如果node在尾部,tail前移
    if (node == tail && compareAndSetTail(node, pred)) {
        //node设置为null
        compareAndSetNext(pred, predNext, null);
    } else {
        //node不在尾部
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        //前继节点是个正常阻塞节点
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                //node后继成为pred的后继
                compareAndSetNext(pred, predNext, next);
        } else {
            // 如果node的前驱是个head,则唤醒node后继,
            //node前继节点不是一个正常的节点,唤醒后继节点
            unparkSuccessor(node);
        }
        node.next = node; // help GC
    }
}

unparkSuccessor 唤醒线程

这里跟阻塞一样,release方法调用tryRelease,由子类自定义释放 state,其他交给AQS。

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参数就是头节点。

private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    //唤醒后继节点的线程,若为空,从tail往后遍历找一个距离head最近的正常的节点
    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);
}

这里有个比较奇怪的逻辑,当头节点的下一个节点是为null或者状态大于0,也就是说这个节点无效时,就从等待列表尾部开始遍历,找到状态正常 <=0 的节点,唤醒它里面的线程。

之所以逆向遍历,是因为要补上面说的坑:在 addWaiter 中cas设置尾节点中,当前节点的prev已经设置为原来的尾结点了,cas设置tail的指向为当前节点成功后,t.next = node; 还没来得及设置,开始上下文开始切换并释放,这就造成了尴尬的局面:这个原先的尾结点没有next。如果从头遍历的尾,就会出现无法遍历完所有链表的情况。而从尾到头就没有这个问题。

同样的,在enq方法里,也有一个片段使用了 addWaiter 的设置尾结点的方法,所以跟enq方法也有关。

 

标签:node,Node,AQS,pred,next,探究,线程,节点,底层
来源: https://www.cnblogs.com/mebi/p/16350374.html