zoukankan      html  css  js  c++  java
  • AQS源码深入分析之共享模式-你知道为什么AQS中要有PROPAGATE这个状态吗?

    本文基于JDK-8u261源码分析


    本篇文章为AQS系列文的第二篇,前文请看:[传送门]

    第一篇:AQS源码深入分析之独占模式-ReentrantLock锁特性详解


    1 Semaphore概览

    共享模式就是有多个线程可以同时拿到锁资源,共享模式用Semaphore来举例,其与ReentrantLock的结构类似,也有公平和非公平两种模式:

     1  public class Semaphore implements Serializable {
     2    //...
     3
     4    private final Sync sync;
     5
     6    abstract static class Sync extends AbstractQueuedSynchronizer {
     7        //...
     8
     9        Sync(int permits) {
    10            setState(permits);
    11        }
    12
    13        //...
    14    }
    15
    16    static final class NonfairSync extends Sync {
    17        //...
    18
    19        NonfairSync(int permits) {
    20            super(permits);
    21        }
    22
    23        //...
    24    }
    25
    26    static final class FairSync extends Sync {
    27        //...
    28
    29        FairSync(int permits) {
    30            super(permits);
    31        }
    32
    33        //...
    34    }
    35
    36    public Semaphore(int permits) {
    37        sync = new NonfairSync(permits);
    38    }
    39
    40    public Semaphore(int permits, boolean fair) {
    41        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    42    }
    43
    44    //...
    45  }
    

    调用构造方法时需要传入一个控制同时并发次数的参数permits,该值会赋值给AQS的state(注意:这里是可以赋值成小于等于0的参数的,如果acquire的参数没有设置好的话,所有线程可能都会一直处于阻塞状态而无法被唤醒)。


    2 非公平锁

    2.1 acquire方法

    Semaphore的非公平锁方式下的acquire方法:

      1  /**
      2   * Semaphore:
      3   */
      4  public void acquire() throws InterruptedException {
      5    sync.acquireSharedInterruptibly(1);
      6  }
      7
      8  /**
      9   * AbstractQueuedSynchronizer:
     10   */
     11  public final void acquireSharedInterruptibly(int arg)
     12        throws InterruptedException {
     13    //arg = 1
     14    //如果当前线程已经中断了,直接抛出异常。因为被中断了就没有意义再去获取锁资源了
     15    if (Thread.interrupted())
     16        throw new InterruptedException();
     17    //尝试去获取共享资源
     18    if (tryAcquireShared(arg) < 0)
     19        //获取资源失败的话,进CLH队列进行排队等待
     20        doAcquireSharedInterruptibly(arg);
     21}
     22
     23  /**
     24   * Semaphore:
     25   * 第18行代码处:
     26   */
     27  protected int tryAcquireShared(int acquires) {
     28    return nonfairTryAcquireShared(acquires);
     29}
     30
     31  final int nonfairTryAcquireShared(int acquires) {
     32    //acquires = 1
     33    for (; ; ) {
     34        int available = getState();
     35        int remaining = available - acquires;
     36        /*
     37        如果剩余资源小于0或者CAS设置state-1成功了的话,退出死循环
     38        注意,这里不需要判断溢出了,因为这里是在做state-1
     39         */
     40        if (remaining < 0 ||
     41                compareAndSetState(available, remaining))
     42            return remaining;
     43    }
     44  }
    

    2.2 doAcquireSharedInterruptibly方法

    doAcquireSharedInterruptibly方法和独占模式的acquireQueued方法类似,但区别是共享模式在一个节点获取锁后,会通知后续的节点也来一起尝试获取:

      1  /**
      2   * AbstractQueuedSynchronizer:
      3   * 和独占模式下的acquireQueued方法的代码类似,只不过这里是共享模式下的响应中断模式
      4   */
      5  private void doAcquireSharedInterruptibly(int arg)
      6        throws InterruptedException {
      7    //CLH队列尾加入一个新的共享节点
      8    final Node node = addWaiter(Node.SHARED);
      9    boolean failed = true;
     10    try {
     11        for (; ; ) {
     12            //获取当前节点的前一个节点
     13            final Node p = node.predecessor();
     14            if (p == head) {
     15                /*
     16                和独占模式一样,只有前一个节点是头节点,也就是当前节点
     17                是实际上的第一个等待着的节点的时候才尝试获取资源(FIFO)
     18                 */
     19                int r = tryAcquireShared(arg);
     20                if (r >= 0) {
     21                    /*
     22                    r大于等于0说明此时还有锁资源(等于0说明锁资源被当前线程拿走后就没了),
     23                    设置头节点,并且通知后面的节点也获取锁资源。独占锁和共享锁的差异点就在于此,
     24                    共享锁在前一个节点获取资源后,会通知后续的节点也一起来获取
     25                     */
     26                    setHeadAndPropagate(node, r);
     27                    p.next = null;
     28                    failed = false;
     29                    return;
     30                }
     31            }
     32            /*
     33            和独占模式一样,将CLH队列中当前节点之前的一些CANCELLED状态的节点剔除;前一个节点状态如果
     34            为SIGNAL时,就会阻塞当前线程。不同的是,这里会抛出异常,而不是独占模式的会设定中断位为true
     35            即响应中断模式,如果线程被中断了会抛出InterruptedException
     36             */
     37            if (shouldParkAfterFailedAcquire(p, node) &&
     38                    parkAndCheckInterrupt())
     39                throw new InterruptedException();
     40        }
     41    } finally {
     42        if (failed)
     43            //如果线程被中断后唤醒,就会取消当前线程获取锁资源的请求
     44            cancelAcquire(node);
     45    }
     46}
     47
     48  /**
     49   * 第26行代码处:
     50   */
     51  private void setHeadAndPropagate(Node node, int propagate) {
     52    //记录旧head节点
     53    Node h = head;
     54    //执行完setHead方法后,node节点成为新的head节点
     55    setHead(node);
     56    /*
     57    <1>propagate>0表示还有剩余锁资源;
     58    <2>旧head节点的状态<0(旧head节点是null这个条件是为了调用waitStatus时防止空指针异常);
     59    <3>新head节点的状态<0(新head节点是null这个条件是为了调用waitStatus时防止空指针异常)
     60    这些条件满足其一就会尝试调用doReleaseShared方法来唤醒后面的节点
     61     */
     62    if (propagate > 0 || h == null || h.waitStatus < 0 ||
     63            (h = head) == null || h.waitStatus < 0) {
     64        Node s = node.next;
     65        /*
     66        具体是否会调用doReleaseShared方法还需要判断node是最后一个节点或者node的下一个节点是
     67        共享节点的时候才去唤醒(判断s是否为null一方面也是为了后面判断s是否是共享节点时不会抛
     68        出空指针异常;但更重要的原因是因为如果node是CLH队列中的最后一个节点的话,这个时候虽然
     69        拿到的s是null,但如果此时有其他的线程在CLH队列中新添加了一个节点后,此处并不能及时感
     70        知到这个变化。于是此时也会走进doReleaseShared方法中去处理这种情况(当然,如果没有发生
     71        多线程插入节点的时候,多调用一次doReleaseShared方法也是无妨的,在该方法里面会过滤掉这
     72        种情况)。同时这里会特殊判断共享节点是因为CLH队列中可能会存在独占节点和共享节点共存的
     73        场景出现,也就是ReentrantReadWriteLock读写锁的场景。这里会一直传播唤醒共享节点直到遇
     74        到一个独占节点为止,后面的节点不管是独占或共享状态都不会再被唤醒了)
     75         */
     76        if (s == null || s.isShared())
     77            doReleaseShared();
     78    }
     79}
     80
     81  /**
     82   * 唤醒后续节点(加锁和释放锁都会调用本方法)
     83   */
     84  private void doReleaseShared() {
     85    for (; ; ) {
     86        Node h = head;
     87        //h != null && h != tail说明此时CLH队列中至少有两个节点(包括空节点),即至少含有一个真正在等待着的节点
     88        if (h != null && h != tail) {
     89            int ws = h.waitStatus;
     90            if (ws == Node.SIGNAL) {
     91                /*
     92                因为下面要唤醒下一个节点,所以将头节点的状态SIGNAL改为0(因为SIGNAL表示的是下一个节点是阻塞状态)
     93                如果CAS没成功,就继续尝试
     94                 */
     95                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
     96                    continue;
     97                //唤醒下一个可以被唤醒的节点
     98                unparkSuccessor(h);
     99            } else if (ws == 0 &&
    100                    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    101                /*
    102                需要注意的是,在共享锁模式下,不论是acquire方法还是release方法,都会调用到doReleaseShared的,
    103                而且每个方法也可能有多个线程在调用。也就是说doReleaseShared方法会有多个线程在调用。假如此时有
    104                多个线程进入到第89行代码处,而其中一个线程先执行了第90行代码处的if条件,将头节点状态改为了0
    105                而剩下的线程就不能跳进第90行代码处的if条件中,而只能走到第99行代码处,ws == 0条件满足,
    106                于是剩下的线程就去CAS竞争修改头节点状态为PROPAGATE(表示需要将唤醒动作向后继续传播)。修改成功的
    107                那个线程就跳到了第135行代码处,进行下个判断逻辑,而再剩下的那些线程就让它们继续循环就行了
    108                (剩下的那些线程会发现head节点此时已经变成了PROPAGATE状态,于是会在下一次循环的第86行代码处
    109                和第135行代码处两次判断head指针是否指向了同一个节点(包括之前那个CAS修改成功的线程和执行唤醒
    110                动作的线程最后也会走到这里)。如果相同了,说明:
    111                <1>可能是当前唤醒传播停止了(每个被唤醒的线程都可能会走入到本方法中的unparkSuccessor处
    112                唤醒下一个节点,相当于把唤醒动作“传播”下去。同时每次唤醒后会变更head指针,如果head不发生变动了,
    113                就说明唤醒传播停止了(注意上面所说的读写锁场景,也有可能是遇到了一个独占节点才停止的));
    114                <2>可能是将要唤醒下一个节点但还没唤醒前的瞬间
    115                不管是属于哪种情况,这些线程都可以退出了(第二种情况下只要等下一个节点唤醒并抢到锁后,还是会走到
    116                本方法里面的,也就是会将唤醒动作继续传播下去。但那个时候就不需要这些线程来操心了,只需要保证唤醒
    117                能一直传播下去就OK))
    118
    119                总结一下:因为head节点的状态为0就说明此时是一个中间过渡状态,最简单的情况下只有这个线程以及它所
    120                唤醒的下个线程们在一直传递地唤醒着,是不会走入到99行代码处的if条件中来的。而如果有线程能走到这里,
    121                就说明此时在doReleaseShared方法也就是本方法中有多个线程在同时调用着。PROPAGATE状态的出现,
    122                我认为是为了创造出一种区别于SIGNAL状态的另外一种状态(因为SIGNAL状态的含义定义死了就是代表后一个
    123                节点是阻塞状态,所以这里不能用SIGNAL状态来代替)。这个时候将head节点由原来的0置为PROPAGATE状态,
    124                以此来保证之前的那些线程也可以读取到此时旧的head节点状态是PROPAGATE,是<0的,从而可以调用到
    125                doReleaseShared方法继续去唤醒下一个节点,也就是将唤醒动作传播下去(在之前某个版本的
    126                setHeadAndPropagate方法中,if条件中是没有最后那两个判断新head节点状态的条件的。如果是这样的话,
    127                我上面的这些分析就是没问题的,但是后来不知道为什么又添加了那两个条件,这个时候的解释就略显苍白了
    128                (因为即使没有PROPAGATE状态,这些获取锁的线程虽然拿到旧的head节点状态是0,但是此时获取到的新的head
    129                节点也就是它们自己,其状态肯定是<0的,所以一样会走doReleaseShared方法)。但是之前确实是这样的,
    130                也就是PROPAGATE状态添加的本意就是为了将唤醒传播下去,可能是后来为了修复某个bug,就又做了些改动
    131                吧,这里就不再深究了)
    132                 */
    133                continue;
    134        }
    135        if (h == head)
    136            break;
    137    }
    138  }
    

    2.3 release方法

    Semaphore的release方法:

     1  /**
     2   * Semaphore:
     3   */
     4  public void release() {
     5    sync.releaseShared(1);
     6}
     7
     8  /**
     9   * AbstractQueuedSynchronizer:
    10   */
    11  public final boolean releaseShared(int arg) {
    12    //arg = 1
    13    //释放锁资源,也就是做state+1的操作
    14    if (tryReleaseShared(arg)) {
    15        /*
    16        唤醒后续可以被唤醒的节点
    17        从这里就可以看出,在共享锁模式下,不仅释放锁的方法可以唤醒节点,加锁的方法也会触发唤醒后续节点的操作
    18         */
    19        doReleaseShared();
    20        return true;
    21    }
    22    return false;
    23}
    24
    25  /**
    26   * Semaphore:
    27   * 第14行代码处:
    28   */
    29  protected final boolean tryReleaseShared(int releases) {
    30    //releases = 1
    31    for (; ; ) {
    32        int current = getState();
    33        int next = current + releases;
    34        //如果超出int最大值,则抛出Error。同时如果传进来的releases本身就小于0的话,也会抛出Error
    35        if (next < current)
    36            throw new Error("Maximum permit count exceeded");
    37        //CAS修改state+1
    38        if (compareAndSetState(current, next))
    39            return true;
    40    }
    41  }
    

    3 PROPAGATE状态

    值得一提的是:纵观整个AQS的源码,只有在doReleaseShared方法中具体用到了PROPAGATE这个状态,在其他地方都是没有显式用到的,那么可能就会对这个状态存在的意义有些许质疑了。其实在早期版本的AQS源码中是没有PROPAGATE这个状态的,之所以要引入它是为了解决一个bug(JDK-6801020):

    img

    从上面可以看到,这个bug是在Java 7中修复的(在Java 6中的一些版本中也已经添加了PROPAGATE状态),同时在bug清单的下面也贴出了可能出现bug的测试代码。那么下面就来看一下离现在非常久远的Java 5u22中的该处代码是如何实现的:

     1  private void setHeadAndPropagate(Node node, int propagate) {
     2    setHead(node);
     3    if (propagate > 0 && node.waitStatus != 0) {
     4        Node s = node.next;
     5        if (s == null || s.isShared())
     6            unparkSuccessor(node);
     7    }
     8  }
     9
    10   public final boolean releaseShared(int arg) {
    11    if (tryReleaseShared(arg)) {
    12        Node h = head;
    13        if (h != null && h.waitStatus != 0)
    14            unparkSuccessor(h);
    15        return true;
    16    }
    17    return false;
    18  }
    

    可以看到,早期版本的实现相比于现在的实现来说简单了很多,总结起来最主要的区别有以下几个:

    1. 在setHeadAndPropagate方法中,早期版本对节点waitStatus状态的判断只是!=0,而现在改为了<0;
    2. 早期版本的releaseShared方法中的执行逻辑和独占锁下的release方法是一样的,而现在将具体的唤醒逻辑写在了doReleaseShared方法里面,和setHeadAndPropagate方法共同调用。

    而可能出现bug的测试代码如下:

     1  import java.util.concurrent.Semaphore;
     2
     3  public class TestSemaphore {
     4
     5    private static Semaphore sem = new Semaphore(0);
     6
     7    private static class Thread1 extends Thread {
     8        @Override
     9        public void run() {
    10            sem.acquireUninterruptibly();
    11        }
    12    }
    13
    14    private static class Thread2 extends Thread {
    15        @Override
    16        public void run() {
    17            sem.release();
    18        }
    19    }
    20
    21    public static void main(String[] args) throws InterruptedException {
    22        for (int i = 0; i < 10000000; i++) {
    23            Thread t1 = new Thread1();
    24            Thread t2 = new Thread1();
    25            Thread t3 = new Thread2();
    26            Thread t4 = new Thread2();
    27            t1.start();
    28            t2.start();
    29            t3.start();
    30            t4.start();
    31            t1.join();
    32            t2.join();
    33            t3.join();
    34            t4.join();
    35            System.out.println(i);
    36        }
    37    }
    38  }
    

    其实上面所做的操作无非就是创建了四个线程:t1和t2用于获取信号量,而t3和t4用于释放信号量,其中的10000000次for循环是为了放大出现bug的几率,join操作是为了阻塞主线程。现在就可以说出出现bug的现象了:也就是这里可能会出现线程被hang住的情况发生(遗憾的是,我并没有模拟出来这个bug)。

    可以想象这样一种场景:假如说当前CLH队列中有一个空节点和两个被阻塞的节点(t1和t2想要获取信号量但获取不到被阻塞在CLH队列中(state初始为0)):head->t1->t2(tail)。

    • 时刻1:t3调用release->releaseShared->tryReleaseShared,将state+1变为1,同时发现此时的head节点不为null并且waitStatus为-1,于是继续调用unparkSuccessor方法,在该方法中会将head的waitStatus改为0;
    • 时刻2:t1被上面t3调用的unparkSuccessor方法所唤醒,调用了tryAcquireShared,将state-1又变为了0。注意,此时还没有调用接下来的setHeadAndPropagate方法;
    • 时刻3:t4调用release->releaseShared->tryReleaseShared,将state+1变为1,同时发现此时的head节点虽然不为null,但是waitStatus为0,所以就不会执行unparkSuccessor方法;
    • 时刻4:t1执行setHeadAndPropagate->setHead,将头节点置为自己。但在此时propagate也就是剩余的state已经为0了(propagate是在时刻2时通过传参的方式传进来的,那个时候-1后剩余的state是0),所以也不会执行unparkSuccessor方法。

    至此可以发现一轮循环走完后,CLH队列中的t2线程永远不会被唤醒,主线程也就永远处在阻塞中,这里也就出现了bug。那么来看一下现在的AQS代码在引入了PROPAGATE状态后,在面对同样的场景下是如何解决这个bug的:

    • 时刻1:t3调用release->releaseShared->tryReleaseShared,将state+1变为1,继续调用doReleaseShared方法,将head的waitStatus改为0,同时调用unparkSuccessor方法;
    • 时刻2:t1被上面t3调用的unparkSuccessor方法所唤醒,调用了tryAcquireShared,将state-1又变为了0。注意,此时还没有调用接下来的setHeadAndPropagate方法;
    • 时刻3:t4调用release->releaseShared->tryReleaseShared,将state+1变为1,同时继续调用doReleaseShared方法,此时会将head的waitStatus改为PROPAGATE
    • 时刻4:t1执行setHeadAndPropagate->setHead,将新的head节点置为自己。虽然此时propagate依旧是0,但是“h.waitStatus < 0”这个条件是满足的(h现在是PROPAGATE状态),同时下一个节点也就是t2也是共享节点,所以会执行doReleaseShared方法,将新的head节点(t1)的waitStatus改为0,同时调用unparkSuccessor方法,此时也就会唤醒t2了。

    至此就可以看出,在引入了PROPAGATE状态后,可以有效避免在高并发场景下可能出现的、线程没有被成功唤醒的情况出现。


    4 公平锁

    4.1 tryAcquireShared方法

    同ReentrantLock一样,Semaphore的公平锁和非公平锁实现上的区别也非常少,只有tryAcquireShared方法是不同的。所以下面就来看一下这个方法的实现:

     1  /**
     2   * Semaphore:
     3   */
     4  protected int tryAcquireShared(int acquires) {
     5    for (; ; ) {
     6        /*
     7        可以看到公平锁模式下的tryAcquireShared方法和非公平锁模式下的nonfairTryAcquireShared方法的区别
     8        一样是多调用了一次hasQueuedPredecessors方法,以此来判断CLH队列中是否有线程的等待获取锁的时间
     9        比当前线程的还要长。如果有的话就会直接返回-1,也就是获取资源失败,然后会进CLH队列进行排队等待
    10        (体现“公平”的含义);没有的话就会去进行state-1,然后返回剩余的锁资源
    11         */
    12        if (hasQueuedPredecessors())
    13            return -1;
    14        int available = getState();
    15        int remaining = available - acquires;
    16        if (remaining < 0 ||
    17                compareAndSetState(available, remaining))
    18            return remaining;
    19    }
    20  }
    


    在这行干的越久真是越觉得:万丈高楼平地起,这绝B是句真理!在应用业务里待太久很多底层的东西往往容易忽略掉,今年的年初计划是把常用的JDK源码工具做一次总结,眼看年底将近,乘着最近有空,赶紧的给补上。

    1. ArrayList你真懂?说说foreach与iterator时remove的区别(已完结)
    2. 你是否想过互联网公司一面为什么总爱问集合?聊聊经典数据结构HashMap(已完结)
    3. AQS源码深入分析之独占模式-ReentrantLock锁特性详解(当前文章)(已完结)
    4. AQS源码深入分析之共享模式-为什么AQS中要有PROPAGATE这个状态?(当前文章)
    5. AQS源码深入分析之条件队列-Java中的阻塞队列是如何实现的?(创作中)
    6. AQS源码深入分析之应用工具CountDownLatch(创作中)
    7. AQS源码深入分析之应用工具CyclicBarrier(创作中)
    8. ConcurrentHashMap源码分析-ConcurrentHashMap在Java 8中的实现还有bug?而且还不止一处!这个坑还比较大,后面会重点总结一下!(已完结)
    9. ThreadPoolExecutor源码分析-问烂了的Java线程池执行流程,其实如果问的细,很多人还是一脸懵逼?(已完结)
    10. ScheduledThreadPoolExecutor源码分析-重点屡屡定时线程池是如何实现延迟执行和周期执行!
    11. ThreadLocal源码分析-重点总结,内存泄漏,软引用弱引用虚引用,面试经常喜欢问,我也喜欢问别个
    12. 红黑树TreeMap、LinkedHashMap(不确定要不要写,有时间写,看项目情况)
    13. 有序并且线程的Map容器ConcurrentSkipListMap(跳表)深入理解
    14. LinkedList(不确定要不要写,有时间写,看项目情况)
    15. 1T数据快速排序!十种经典排序算法总结(已完结)

    每一次总结都是对知识点掌握程度的审视,技术不易,每日精进一点,与大家共勉。

    另外笔者公众号:奇客时间,有更多精彩的文章,有兴趣的同学,可以关注

  • 相关阅读:
    Mybatis深入浅出之工作原理
    Mybatis深入浅出之缓存机制
    Error :Unable to access jarfile *.jar
    Mysql与JDBC版本兼容性问题
    找工作的正确方法
    关于制作云主机基准镜像
    笔记分享
    Android4.0.1找不到R.java
    android 反编译出错 can not merge I and Z
    [论文理解] Improving the imporved training of Wasserstesin GANS: A consistency term and its dual effect
  • 原文地址:https://www.cnblogs.com/tomakemyself/p/13928620.html
Copyright © 2011-2022 走看看