AQS（2）-同步状态的获取和释放

AQS（2）-同步状态的获取和释放

目录

• AQS（2）-同步状态的获取和释放
• 一、独占式
  • 1.0 写在开头
  • 1.1 独占式同步状态获取
    • 1.1.1 acquire
    • 1.1.2 tryAcquire方法
    • 1.1.3 acquireQueued方法
    • 1.1.4 shouldParkAfterFailedAcquire
    • 1.1.5 parkAndCheckInterrupt
    • 1.1.6 NonFair锁中的实现
    • 1.1.7 Fair锁的tryAcquire
  • 1.2 独占式超时获取
    • 1.2.1 tryAcquireNanos
    • 1.2.2 doAcquireNanos(agr , nanosTimeOut)
    • 1.2.3 小结
  • 1.3 独占式响应中断
    • 1.3.1 acquireInterruptibly
    • 1.3.2 doAcquireInterruptibly
  • 1.4 独占式同步状态释放
    • 1.4.1 release方法
    • 1.4.2 unparkSuccessor方法
  • 1.5 总结

一、独占式

1.0 写在开头

对于每种方式都会先有个try和实际的do的过程，独占式和共享式最大的区别是进行唤醒后继线程的时间点

1.1 独占式同步状态获取

1.1.1 acquire

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试进行获取
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//添加到队列中
        selfInterrupt();//中断，等待其他节点唤醒
}

• 二行：尝试进行获取
• 三行：将节点通过addWaiter添加到尾部，在acquireQueued中尝试对当前线程进行park
• 四行：park会暂时屏蔽中断，如果线程在完成或者被取消后检测到期间被中断，这时候需要进行中断处理

1.1.2 tryAcquire方法

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

tryAcquire并未被final修饰，表示需要被子类进行继承，这里的方法是区分公平锁和非公平锁的重要地方。这里通过查看ReentrantLock中的两种锁的实现来查看他们的区别，因为和本文关系不大，放在本节的最后

1.1.3 acquireQueued方法

addWaiter在AQS(1)中已经讲解了，就不在赘述

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);//自旋失败，也代表获取锁失败，取消本节点
    }
}

• 循环进行操作
  • 获取前驱节点的值
    • 当前节点为head节点代表当前节点应该被唤醒，进行尝试获取锁并返回
    • 根据所的状态进行判断是否应该进行阻塞shouldParkAfterFailedAcquire （后文讲解）
    • 阻塞完成后需要进行中断处理就修改中断状态量
• 进行CAS获取锁失败，删除当前节点并修改状态值

1.1.4 shouldParkAfterFailedAcquire

Checks and updates status for a node that failed to acquire. Returns true if thread should block. This is the main signal control in all acquire loops. Requires that pred == node.prev.

当获取锁失败时用于检测或更新节点的状态值，这个函数是重要的信号控制函数

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        //前驱节点已经知晓本节点需要进入park
        return true;
    if (ws > 0) {
      //前驱节点已经被删除，也就是CANCELLED，需要修改指向，保证前驱节点指向一个未被删除的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /* 将节点变化为SINGAL             */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

这里引用一下AQS(1)中的一个关于SIGNAL状态量的定义

SIGNAL：该节点的后继结点已经通过LockSupport.park()方法阻塞，当前节点在被释放或者被删除后需要唤醒它的后继节点，为避免线程之间的竞争，获取资源acquire的所有方法都应该设置SIGNAL标志，然后重新进行原子性的获取操作，如果获取失败，就阻塞。

也就是说需要前驱节点知道当前节点需要被唤醒时（能保证它的后驱节点，也就是当前节点node可以在其被删除或者完成后被unpark）

• 前驱节点状态ws为SIGNAL时代表当前节点可以进行park（你前面的人已经知道到你的时候会叫醒你）
• 前驱节点状态ws>0也就是CANCELLED，表示前驱节点已经被废除，不能正常通知你，需要找到一个可靠的前驱节点(ws<=0)
• 前驱节点状态正常ws<=0，需要通知前驱节点**即将前驱节点状态ws设置为SIGNAL**
• 以上状况只有SIGNAL才需要被阻塞即return true

//todo hf 非常重要的一点是这些状态之间如何进行转换和各个状态之间应该处于什么状况下，这两个问题暂时没有解决

1.1.5 parkAndCheckInterrupt

算得上这篇文章里面最简洁的一个函数

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);//阻塞当前线程
    return Thread.interrupted();//中断
}

• 阻塞当前线程
• 阻塞完毕后进行中断处理，返回中断标志并恢复中断标志

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);//指定阻塞的对象，也就是将当前线程挂载到对象上，这样在解锁的时候也可以通过对象进行解锁
    UNSAFE.park(false, 0L);//进行阻塞
    setBlocker(t, null);//清楚阻塞的对象，进行到这步时表示阻塞已经完成
}

1.1.6 NonFair锁中的实现

protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {//如果剩余为0
        if (compareAndSetState(0, acquires)) {//修改状态
            setExclusiveOwnerThread(current);//将锁设置为当前线程持有
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {//当前线程已经持有锁
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

• 获取当前锁的重入值state（也就是当前的线程持有这个锁的层数，因为锁是可重入的）
  • 锁的重入值为0，也就是当前没有线程持有锁，将持有锁的线程设置为当前线程并变更锁的层数，尝试进行获取成功
  • 锁被当前持有，添加重入的层数并更新层数，尝试获取成功
  • 如果不是上面两种情况，那么尝试获取锁失败

1.1.7 Fair锁的tryAcquire

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

• 获取当前锁的重入值state（也就是当前的线程持有这个锁的层数，因为锁是可重入的）
  • 当前的层数为0(当前锁没有被线程锁定)，只有当它没有后继节点hasQueuedPredecessors() == false才会尝试获取
  • 锁被当前线程持有，与nonFair实现相同
  • 尝试获取失败

1.2 独占式超时获取

超时获取最核心问题是判断什么时候超时和进行定时的park

1.2.1 tryAcquireNanos

public final boolean tryAcquireNanos(int arg, long nanosTimeout)
    throws InterruptedException {
    if (Thread.interrupted())//中断
        throw new InterruptedException();
    return tryAcquire(arg) ||//尝试进行获取
        doAcquireNanos(arg, nanosTimeout);//进行正式的获取
}

• 判断是否已经中断，因如果线程已经中断，那么后面的操作都是无意义的
• 尝试进行获取，这是每次锁必须的，可以查看前面的解析
• 进行正式的超时获取，需要带上超时的时间

1.2.2 doAcquireNanos(agr , nanosTimeOut)

private boolean doAcquireNanos(int arg, long nanosTimeout)//nanosTimeOut是需要保持的时长
    throws InterruptedException {
    if (nanosTimeout <= 0L)//快速结束
        return false;
    final long deadline = System.nanoTime() + nanosTimeout;//计算出相对于机器的结束时间
    final Node node = addWaiter(Node.EXCLUSIVE);//添加后继节点
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {//设置头
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return true;
            }
            nanosTimeout = deadline - System.nanoTime();//目标时间和当前时间的差值
            if (nanosTimeout <= 0L)//超时
                return false;
            if (shouldParkAfterFailedAcquire(p, node) &&//当前已经过去锁失败
                nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout);//阻塞
            if (Thread.interrupted())//中断判断
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);//取消节点
    }
}

doAcquireNanos其实和doAcquire方法思路大致相同，主要解决的就是超时判断

• 快速的超时判断

• 将当前节点添加到队列中

  • 循环完成下列任务（部分函数中其实只做一部分的操作比如说将shouldParkAfterFailedAcquire中若waitStatus不为SIGNAL且小于0就会将其设置为SIGNAL但其实还没有进行parkAndCheckInterrupted，需要重新进行一次循环才行）

  • 与doAcquire相同

  • 计算时间差值

  • 如果超时，返回当前状态获取状态量失败

  • shouldParkAfterFailedAcquire方法在前面解析过，不再赘述。当前节点和它的前驱节点已经准备好进行阻塞并且最小的spinForTimeoutThreshold（这个自旋值为1000)，也就是说剩余时间少于1000毫微秒就不会进行阻塞，这里有个很好的解释就是从计算出nanosTimeout到进行nanosTimeout>spinForTimeoutThreshold这个比较的过程中实际上是需要花费一定时间的，至于具体是多长的时间，因为不能对源码进行修改并运行，这里做了一个简单的测试System.out.println(System.nanoTime()-System.nanoTime());输出值为-100，也就是两条指令的执行时间为100毫微秒。当然重要的还有LockSupport.parkNanos方法。这里因为是个核心api，并未有太多的代码 java

    public static void parkNanos(Object blocker, long nanos) {
        if (nanos > 0) {
            Thread t = Thread.currentThread();
            setBlocker(t, blocker);
            UNSAFE.park(false, nanos);
            setBlocker(t, null);
        }
    }
    

  • 进行中断判断

• 如果自旋失败，进行节点取消操作

1.2.3 小结

对于超时中断方式可以看出其实主要的方式还是对超时的控制，其他方面上与普通的doAcquire()方式其实并没有什么太大的不同

1.3 独占式响应中断

上面讲解的两个方法不论是acquire还是doAcquireNanos都是不会响应中断的，只有任务完成后，才会进行中断的响应。部分实现的代码是需要进行中断的处理，如果用户关心中断的话。

1.3.1 acquireInterruptibly

public final void acquireInterruptibly(int arg)
    throws InterruptedException {
    if (Thread.interrupted())//中断
        throw new InterruptedException();
    if (!tryAcquire(arg))//尝试进行获取
        doAcquireInterruptibly(arg);//带中断的进行获取
}

这里的流程和tryAcquireNanos基本一致，主要区别在于最后一个方法doAcquireInterruptibly

1.3.2 doAcquireInterruptibly

private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();//响应中断
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

这里基本实现也和acquireQueued操作相似，找出不相同的地方进行讲解，parkAndCheckInterrupt进行park其实和acquireQueued中的操作也是一样的，主要是这个方法是否会抛出异常在acquireQueued中我们只设置了一个interrupted变量并进行返回，并没有实际抛出错误，在这里就实际的抛出了InterruptedException

1.4 独占式同步状态释放

Releases in exclusive mode. Implemented by unblocking one or more threads if {@link #tryRelease} returns true. This method can be used to implement method {@link Lock#unlock}.

独占锁的释放需要解决的问题是如何释放节点和进行状态的传播。

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

• 尝试进行唤醒（基本操作）
• 如果头结点正常(不为空，并且waitStatus不是INITIAL)就唤醒一个节点，处于INITIAL状态的节点表示其并没有后去节点需要准备唤醒，因为在addWaiter中提到如果当前节点被阻塞的话会将他的前驱节点设置为SIGNAL，//todo 这里有个疑问，为什么CANCELLED和PROPAGATE`状态的头结点也算是正常的。

1.4.2 unparkSuccessor方法

 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.
         */
     //todo 这里不太理解为什么要进行状态的还原
        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.
         */
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {//如果当前节点是被CANCELLED的，就进行唤醒
            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);//
    }

• 将当前头结点的状态值设置为INITIAL
• 后驱节点被删除CANCELLED就进行反向遍历（反向遍历的原因通过我的猜测应该是之前提到的addWaiter添加为节点的都是先进行pre.prev=tail然后才进行tail的连接，可能是因为从后面进行遍历会更加问题），知道找到状态正常的点（一定不会找到当前的头结点，因为这个节点已经被唤醒过）
• 进行unpark也就是唤醒

这里重要的是进行后续遍历来找到正常的节点。

1.5 总结

简单的介绍了独占状态的获取和状态的释放以及中断的响应，其中大体思想都蕴含在acquire``release``addWaiter``enq方法中，后面的基本上都是添加一些特性上说需要的必要东西相对于骨架来说变化也并不是太大。

虽然方法的信息也确实是完成，对于head或者tail的指针变化方式和节点的状态也更加熟悉，但是对于状态的意义和如何进行变换还是比较模糊，希望下一篇解析的文章中能够弄清楚这些问题。