本文基于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  }1213  //...14 }1516 static final class NonfairSync extends Sync {17  //...1819  NonfairSync(int permits) {20   super(permits);21  }2223  //...24 }2526 static final class FairSync extends Sync {27  //...2829  FairSync(int permits) {30   super(permits);31  }3233  //...34 }3536 public Semaphore(int permits) {37  sync = new NonfairSync(permits);38 }3940 public Semaphore(int permits, boolean fair) {41  sync = fair ? new FairSync(permits) : new NonfairSync(permits);42 }4344 //...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条件，将头节点状态改为了0105    而剩下的线程就不能跳进第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））118119    总结一下：因为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，但是此时获取到的新的head129    节点也就是它们自己，其状态肯定是<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 = 113 //释放锁资源，也就是做state+1的操作14 if (tryReleaseShared(arg)) {15  /*16  唤醒后续可以被唤醒的节点17  从这里就可以看出，在共享锁模式下，不仅释放锁的方法可以唤醒节点，加锁的方法也会触发唤醒后续节点的操作18   */19  doReleaseShared();20  return true;21 }22 return false;23}2425 /**26 * Semaphore:27 * 第14行代码处：28 */29 protected final boolean tryReleaseShared(int releases) {30 //releases = 131 for (; ; ) {32  int current = getState();33  int next = current + releases;34  //如果超出int最大值，则抛出Error。同时如果传进来的releases本身就小于0的话，也会抛出Error35  if (next < current)36   throw new Error("Maximum permit count exceeded");37  //CAS修改state+138  if (compareAndSetState(current, next))39   return true;40 }41 }

3 PROPAGATE状态

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

从上面可以看到，这个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 } 910 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 }

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

在setHeadAndPropagate方法中，早期版本对节点waitStatus状态的判断只是!=0，而现在改为了<0；
早期版本的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 }1314 private static class Thread2 extends Thread {15  @Override16  public void run() {17   sem.release();18  }19 }2021 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源码工具做一次总结，眼看年底将近，乘着最近有空，赶紧的给补上。

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

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

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

原文转载：http://www.shaoqun.com/a/488114.html

e票联：https://www.ikjzd.com/w/1452

盘古集团：https://www.ikjzd.com/w/1448

蜜芽宝贝：https://www.ikjzd.com/w/1320

本文基于JDK-8u261源码分析本篇文章为AQS系列文的第二篇，前文请看：[传送门]第一篇：AQS源码深入分析之独占模式-ReentrantLock锁特性详解1Semaphore概览共享模式就是有多个线程可以同时拿到锁资源，共享模式用Semaphore来举例，其与ReentrantLock的结构类似，也有公平和非公平两种模式：1publicclassSemaphoreimplementsSeri
捷汇：https://www.ikjzd.com/w/419
识货：https://www.ikjzd.com/w/1745
成为eBay大卖家前，你必须有这些运营技能和魄力：https://www.ikjzd.com/home/9261
东南亚社交营销增长策略：吸引与信任是营销的核心：https://www.ikjzd.com/home/132060
2020深圳春节哪里有好玩的？深圳春节去哪里玩？：http://tour.shaoqun.com/a/21667.html

Fashion World

2020-11-04

AQS源码深入分析之共享模式