Java并发系列[3]----AbstractQueuedSynchronizer源码分析之共享模式
通过上一篇的分析，我们知道了独占模式获取锁有三种方式，分别是不响应线程中断获取，响应线程中断获取，设置超时时间获取。在共享模式下获取锁的方式也是这三种，而且基本上都是大同小异，我们搞清楚了一种就能很快的理解其他的方式。虽然说AbstractQueuedSynchronizer源码有一千多行，但是重复的也比较多，所以读者不要刚开始的时候被吓到，只要耐着性子去看慢慢的自然能够渐渐领悟。就我个人经验来说，阅读AbstractQueuedSynchronizer源码有几个比较关键的地方需要弄明白，分别是独占模式和共享模式的区别，结点的等待状态，以及对条件队列的理解。理解了这些要点那么后续源码的阅读将会轻松很多。当然这些在我的《Java并发系列[1]—-AbstractQueuedSynchronizer源码分析之概要分析》这篇文章里都有详细的介绍，读者可以先去查阅。本篇对于共享模式的分析也是分为三种获取锁的方式和一种释放锁的方式。
1. 不响应线程中断的获取
1 //以不可中断模式获取锁(共享模式) 2 public final void acquireShared(int arg) { 3 //1.尝试去获取锁 4 if (tryAcquireShared(arg) < 0) { 5 //2.如果获取失败就进入这个方法 6 doAcquireShared(arg); 7 } 8 } 9 10 //尝试去获取锁(共享模式) 11 //负数：表示获取失败 12 //零值：表示当前结点获取成功, 但是后继结点不能再获取了 13 //正数：表示当前结点获取成功, 并且后继结点同样可以获取成功 14 protected int tryAcquireShared(int arg) { 15 throw new UnsupportedOperationException(); 16 }
调用acquireShared方法是不响应线程中断获取锁的方式。在该方法中，首先调用tryAcquireShared去尝试获取锁，tryAcquireShared方法返回一个获取锁的状态，这里AQS规定了返回状态若是负数代表当前结点获取锁失败，若是0代表当前结点获取锁成功，但后继结点不能再获取了，若是正数则代表当前结点获取锁成功，并且这个锁后续结点也同样可以获取成功。子类在实现tryAcquireShared方法获取锁的逻辑时，返回值需要遵守这个约定。如果调用tryAcquireShared的返回值小于0，就代表这次尝试获取锁失败了，接下来就调用doAcquireShared方法将当前线程添加进同步队列。我们看到doAcquireShared方法。
1 //在同步队列中获取(共享模式) 2 private void doAcquireShared(int arg) { 3 //添加到同步队列中 4 final Node node = addWaiter(Node.SHARED); 5 boolean failed = true; 6 try { 7 boolean interrupted = false; 8 for (;;) { 9 //获取当前结点的前继结点 10 final Node p = node.predecessor(); 11 //如果前继结点为head结点就再次尝试去获取锁 12 if (p == head) { 13 //再次尝试去获取锁并返回获取状态 14 //r < 0, 表示获取失败 15 //r = 0, 表示当前结点获取成功, 但是后继结点不能再获取了 16 //r > 0, 表示当前结点获取成功, 并且后继结点同样可以获取成功 17 int r = tryAcquireShared(arg); 18 if (r >= 0) { 19 //到这里说明当前结点已经获取锁成功了, 此时它会将锁的状态信息传播给后继结点 20 setHeadAndPropagate(node, r); 21 p.next = null; 22 //如果在线程阻塞期间收到中断请求, 就在这一步响应该请求 23 if (interrupted) { 24 selfInterrupt(); 25 } 26 failed = false; 27 return; 28 } 29 } 30 //每次获取锁失败后都会判断是否可以将线程挂起, 如果可以的话就会在parkAndCheckInterrupt方法里将线程挂起 31 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { 32 interrupted = true; 33 } 34 } 35 } finally { 36 if (failed) { 37 cancelAcquire(node); 38 } 39 } 40 }
进入doAcquireShared方法首先是调用addWaiter方法将当前线程包装成结点放到同步队列尾部。这个添加结点的过程我们在讲独占模式时讲过，这里就不再讲了。结点进入同步队列后，如果它发现在它前面的结点就是head结点，因为head结点的线程已经获取锁进入房间里面了，那么下一个获取锁的结点就轮到自己了，所以当前结点先不会将自己挂起，而是再一次去尝试获取锁，如果前面那人刚好释放锁离开了，那么当前结点就能成功获得锁，如果前面那人还没有释放锁，那么就会调用shouldParkAfterFailedAcquire方法，在这个方法里面会将head结点的状态改为SIGNAL，只有保证前面结点的状态为SIGNAL，当前结点才能放心的将自己挂起，所有线程都会在parkAndCheckInterrupt方法里面被挂起。如果当前结点恰巧成功的获取了锁，那么接下来就会调用setHeadAndPropagate方法将自己设置为head结点，并且唤醒后面同样是共享模式的结点。下面我们看下setHeadAndPropagate方法具体的操作。
1 //设置head结点并传播锁的状态(共享模式) 2 private void setHeadAndPropagate(Node node, int propagate) { 3 Node h = head; 4 //将给定结点设置为head结点 5 setHead(node); 6 //如果propagate大于0表明锁可以获取了 7 if (propagate > 0 || h == null || h.waitStatus < 0) { 8 //获取给定结点的后继结点 9 Node s = node.next; 10 //如果给定结点的后继结点为空, 或者它的状态是共享状态 11 if (s == null || s.isShared()) { 12 //唤醒后继结点 13 doReleaseShared(); 14 } 15 } 16 } 17 18 //释放锁的操作(共享模式) 19 private void doReleaseShared() { 20 for (;;) { 21 //获取同步队列的head结点 22 Node h = head; 23 if (h != null && h != tail) { 24 //获取head结点的等待状态 25 int ws = h.waitStatus; 26 //如果head结点的状态为SIGNAL, 表明后面有人在排队 27 if (ws == Node.SIGNAL) { 28 //先把head结点的等待状态更新为0 29 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) { 30 continue; 31 } 32 //再去唤醒后继结点 33 unparkSuccessor(h); 34 //如果head结点的状态为0, 表明此时后面没人在排队, 就只是将head状态修改为PROPAGATE 35 }else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) { 36 continue; 37 } 38 } 39 //只有保证期间head结点没被修改过才能跳出循环 40 if (h == head) { 41 break; 42 } 43 } 44 }
调用setHeadAndPropagate方法首先将自己设置成head结点，然后再根据传入的tryAcquireShared方法的返回值来决定是否要去唤醒后继结点。前面已经讲到当返回值大于0就表明当前结点成功获取了锁，并且后面的结点也可以成功获取锁。这时当前结点就需要去唤醒后面同样是共享模式的结点，注意，每次唤醒仅仅只是唤醒后一个结点，如果后一个结点不是共享模式的话，当前结点就直接进入房间而不会再去唤醒更后面的结点了。共享模式下唤醒后继结点的操作是在doReleaseShared方法进行的，共享模式和独占模式的唤醒操作基本也是相同的，都是去找到自己座位上的牌子(等待状态)，如果牌子上为SIGNAL表明后面有人需要让它帮忙唤醒，如果牌子上为0则表明队列此时并没有人在排队。在独占模式下是如果发现没人在排队就直接离开队列了，而在共享模式下如果发现队列后面没人在排队，当前结点在离开前仍然会留个小纸条(将等待状态设置为PROPAGATE)告诉后来的人这个锁的可获取状态。那么后面来的人在尝试获取锁的时候可以根据这个状态来判断是否直接获取锁。
2. 响应线程中断的获取
1 //以可中断模式获取锁(共享模式) 2 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 3 //首先判断线程是否中断, 如果是则抛出异常 4 if (Thread.interrupted()) { 5 throw new InterruptedException(); 6 } 7 //1.尝试去获取锁 8 if (tryAcquireShared(arg) < 0) { 9 //2. 如果获取失败则进人该方法 10 doAcquireSharedInterruptibly(arg); 11 } 12 } 13 14 //以可中断模式获取(共享模式) 15 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { 16 //将当前结点插入同步队列尾部 17 final Node node = addWaiter(Node.SHARED); 18 boolean failed = true; 19 try { 20 for (;;) { 21 //获取当前结点的前继结点 22 final Node p = node.predecessor(); 23 if (p == head) { 24 int r = tryAcquireShared(arg); 25 if (r >= 0) { 26 setHeadAndPropagate(node, r); 27 p.next = null; 28 failed = false; 29 return; 30 } 31 } 32 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { 33 //如果线程在阻塞过程中收到过中断请求, 那么就会立马在这里抛出异常 34 throw new InterruptedException(); 35 } 36 } 37 } finally { 38 if (failed) { 39 cancelAcquire(node); 40 } 41 } 42 }
响应线程中断获取锁的方式和不响应线程中断获取锁的方式在流程上基本是相同的，唯一的区别就是在哪里响应线程的中断请求。在不响应线程中断获取锁时，线程从parkAndCheckInterrupt方法中被唤醒，唤醒后就立马返回是否收到中断请求，即使是收到了中断请求也会继续自旋直到获取锁后才响应中断请求将自己给挂起。而响应线程中断获取锁会才线程被唤醒后立马响应中断请求，如果在阻塞过程中收到了线程中断就会立马抛出InterruptedException异常。
3. 设置超时时间的获取
1 //以限定超时时间获取锁(共享模式) 2 public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { 3 if (Thread.interrupted()) { 4 throw new InterruptedException(); 5 } 6 //1.调用tryAcquireShared尝试去获取锁 7 //2.如果获取失败就调用doAcquireSharedNanos 8 return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout); 9 } 10 11 //以限定超时时间获取锁(共享模式) 12 private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { 13 long lastTime = System.nanoTime(); 14 final Node node = addWaiter(Node.SHARED); 15 boolean failed = true; 16 try { 17 for (;;) { 18 //获取当前结点的前继结点 19 final Node p = node.predecessor(); 20 if (p == head) { 21 int r = tryAcquireShared(arg); 22 if (r >= 0) { 23 setHeadAndPropagate(node, r); 24 p.next = null; 25 failed = false; 26 return true; 27 } 28 } 29 //如果超时时间用完了就结束获取, 并返回失败信息 30 if (nanosTimeout <= 0) { 31 return false; 32 } 33 //1.检查是否满足将线程挂起要求(保证前继结点状态为SIGNAL) 34 //2.检查超时时间是否大于自旋时间 35 if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) { 36 //若满足上面两个条件就将当前线程挂起一段时间 37 LockSupport.parkNanos(this, nanosTimeout); 38 } 39 long now = System.nanoTime(); 40 //超时时间每次减去获取锁的时间 41 nanosTimeout -= now - lastTime; 42 lastTime = now; 43 //如果在阻塞时收到中断请求就立马抛出异常 44 if (Thread.interrupted()) { 45 throw new InterruptedException(); 46 } 47 } 48 } finally { 49 if (failed) { 50 cancelAcquire(node); 51 } 52 } 53 }
如果看懂了上面两种获取方式，再来看设置超时时间的获取方式就会很轻松，基本流程都是一样的，主要是理解超时的机制是怎样的。如果第一次获取锁失败会调用doAcquireSharedNanos方法并传入超时时间，进入方法后会根据情况再次去获取锁，如果再次获取失败就要考虑将线程挂起了。这时会判断超时时间是否大于自旋时间，如果是的话就会将线程挂起一段时间，否则就继续尝试获取，每次获取锁之后都会将超时时间减去获取锁的时间，一直这样循环直到超时时间用尽，如果还没有获取到锁的话就会结束获取并返回获取失败标识。在整个期间线程是响应线程中断的。
4. 共享模式下结点的出队操作
1 //释放锁的操作(共享模式) 2 public final boolean releaseShared(int arg) { 3 //1.尝试去释放锁 4 if (tryReleaseShared(arg)) { 5 //2.如果释放成功就唤醒其他线程 6 doReleaseShared(); 7 return true; 8 } 9 return false; 10 } 11 12 //尝试去释放锁(共享模式) 13 protected boolean tryReleaseShared(int arg) { 14 throw new UnsupportedOperationException(); 15 } 16 17 //释放锁的操作(共享模式) 18 private void doReleaseShared() { 19 for (;;) { 20 //获取同步队列的head结点 21 Node h = head; 22 if (h != null && h != tail) { 23 //获取head结点的等待状态 24 int ws = h.waitStatus; 25 //如果head结点的状态为SIGNAL, 表明后面有人在排队 26 if (ws == Node.SIGNAL) { 27 //先把head结点的等待状态更新为0 28 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) { 29 continue; 30 } 31 //再去唤醒后继结点 32 unparkSuccessor(h); 33 //如果head结点的状态为0, 表明此时后面没人在排队, 就只是将head状态修改为PROPAGATE 34 }else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) { 35 continue; 36 } 37 } 38 //只有保证期间head结点没被修改过才能跳出循环 39 if (h == head) { 40 break; 41 } 42 } 43 }
线程在房间办完事之后就会调用releaseShared方法释放锁，首先调用tryReleaseShared方法尝试释放锁，该方法的判断逻辑由子类实现。如果释放成功就调用doReleaseShared方法去唤醒后继结点。走出房间后它会找到原先的座位(head结点)，看看座位上是否有人留了小纸条(状态为SIGNAL)，如果有就去唤醒后继结点。如果没有(状态为0)就代表队列没人在排队，那么在离开之前它还要做最后一件事情，就是在自己座位上留下小纸条(状态设置为PROPAGATE)，告诉后面的人锁的获取状态，整个释放锁的过程和独占模式唯一的区别就是在这最后一步操作。
注：以上全部分析基于JDK1.7，不同版本间会有差异，读者需要注意