UDN-企业互联网技术人气社区

板块导航

浏览  : 999
回复  : 0

[讨论交流] 深入浅出java同步器

[复制链接]
芭芭拉的头像 楼主
发表于 2016-9-15 15:06:57 | 显示全部楼层 |阅读模式
  前言

  在java.util.concurrent.locks包中有很多Lock的实现类,常用的有ReentrantLock、ReadWriteLock(实现类ReentrantReadWriteLock),内部实现都依赖AbstractQueuedSynchronizer类,接下去让我们看看Doug Lea大神是如何使用一个普通类就完成了代码块的并发访问控制。为了方便,本文中使用AQS代替AbstractQueuedSynchronizer。

  定义
  1. public abstract class AbstractQueuedSynchronizer extends
  2.     AbstractOwnableSynchronizer implements java.io.Serializable {
  3.     //等待队列的头节点
  4.     private transient volatile Node head;
  5.     //等待队列的尾节点
  6.     private transient volatile Node tail;
  7.     //同步状态
  8.     private volatile int state;
  9.     protected final int getState() { return state;}
  10.     protected final void setState(int newState) { state = newState;}
  11.     ...
  12. }
复制代码

  队列同步器AQS是用来构建锁或其他同步组件的基础框架,内部使用一个int成员变量表示同步状态,通过内置的FIFO队列来完成资源获取线程的排队工作,其中内部状态state,等待队列的头节点head和尾节点head,都是通过volatile修饰,保证了多线程之间的可见。

  在深入实现原理之前,我们先看看内部的FIFO队列是如何实现的。
  1. static final class Node {
  2.         static final Node SHARED = new Node();
  3.         static final Node EXCLUSIVE = null;
  4.         static final int CANCELLED =  1;
  5.         static final int SIGNAL    = -1;
  6.         static final int CONDITION = -2;
  7.         static final int PROPAGATE = -3;
  8.         volatile int waitStatus;
  9.         volatile Node prev;
  10.         volatile Node next;
  11.         volatile Thread thread;
  12.         Node nextWaiter;
  13.         ...
  14.     }
复制代码

  先来一张形象的图(该图其实是网上找的)
1.png

  黄色节点是默认head节点,其实是一个空节点,我觉得可以理解成代表当前持有锁的线程,每当有线程竞争失败,都是插入到队列的尾节点,tail节点始终指向队列中的最后一个元素。

  每个节点中, 除了存储了当前线程,前后节点的引用以外,还有一个waitStatus变量,用于描述节点当前的状态。多线程并发执行时,队列中会有多个节点存在,这个waitStatus其实代表对应线程的状态:有的线程可能获取锁因为某些原因放弃竞争;有的线程在等待满足条件,满足之后才能执行等等。一共有4中状态:

  • CANCELLED 取消状态
  • SIGNAL 等待触发状态
  • CONDITION 等待条件状态
  • PROPAGATE 状态需要向后传播

  等待队列是FIFO先进先出,只有前一个节点的状态为SIGNAL时,当前节点的线程才能被挂起。

  实现原理

  子类重写tryAcquire和tryRelease方法通过CAS指令修改状态变量state。
  1. public final void acquire(int arg) {   
  2. if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))   
  3.     selfInterrupt();
  4. }
复制代码

  线程获取锁过程

  下列步骤中线程A和B进行竞争。

  • 线程A执行CAS执行成功,state值被修改并返回true,线程A继续执行。
  • 线程A执行CAS指令失败,说明线程B也在执行CAS指令且成功,这种情况下线程A会执行步骤3。

  生成新Node节点node,并通过CAS指令插入到等待队列的队尾(同一时刻可能会有多个Node节点插入到等待队列中),如果tail节点为空,则将head节点指向一个空节点(代表线程B),具体实现如下:
  1. private Node addWaiter(Node mode) {
  2. Node node = new Node(Thread.currentThread(), mode);
  3. // Try the fast path of enq; backup to full enq on failure
  4. Node pred = tail;
  5. if (pred != null) {
  6.      node.prev = pred;
  7.      if (compareAndSetTail(pred, node)) {
  8.          pred.next = node;
  9.          return node;
  10.      }
  11. }
  12. enq(node);
  13. return node;
  14. }
  15. private Node enq(final Node node) {
  16. for (;;) {
  17.      Node t = tail;
  18.      if (t == null) { // Must initialize
  19.          if (compareAndSetHead(new Node()))
  20.              tail = head;
  21.      } else {
  22.          node.prev = t;
  23.          if (compareAndSetTail(t, node)) {
  24.              t.next = node;
  25.              return t;
  26.          }
  27.      }
  28. }
  29. }
复制代码

  node插入到队尾后,该线程不会立马挂起,会进行自旋操作。因为在node的插入过程,线程B(即之前没有阻塞的线程)可能已经执行完成,所以要判断该node的前一个节点pred是否为head节点(代表线程B),如果pred == head,表明当前节点是队列中第一个“有效的”节点,因此再次尝试tryAcquire获取锁,

  1、如果成功获取到锁,表明线程B已经执行完成,线程A不需要挂起。
  2、如果获取失败,表示线程B还未完成,至少还未修改state值。进行步骤5。
  1. final boolean acquireQueued(final Node node, int arg) {
  2. boolean failed = true;
  3. try {
  4.      boolean interrupted = false;
  5.      for (;;) {
  6.          final Node p = node.predecessor();
  7.          if (p == head && tryAcquire(arg)) {
  8.              setHead(node);
  9.              p.next = null; // help GC
  10.              failed = false;
  11.              return interrupted;
  12.          }
  13.          if (shouldParkAfterFailedAcquire(p, node) &&
  14.              parkAndCheckInterrupt())
  15.              interrupted = true;
  16.      }
  17. } finally {
  18.      if (failed)
  19.          cancelAcquire(node);
  20. }
  21. }
复制代码

  前面我们已经说过只有前一个节点pred的线程状态为SIGNAL时,当前节点的线程才能被挂起。

  1、如果pred的waitStatus == 0,则通过CAS指令修改waitStatus为Node.SIGNAL。
  2、如果pred的waitStatus > 0,表明pred的线程状态CANCELLED,需从队列中删除。
  3、如果pred的waitStatus为Node.SIGNAL,则通过LockSupport.park()方法把线程A挂起,并等待被唤醒,被唤醒后进入步骤6。

  具体实现如下:
  1. private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  2. int ws = pred.waitStatus;
  3. if (ws == Node.SIGNAL)
  4.      /*
  5.       * This node has already set status asking a release
  6.       * to signal it, so it can safely park.
  7.       */
  8.      return true;
  9. if (ws > 0) {
  10.      /*
  11.       * Predecessor was cancelled. Skip over predecessors and
  12.       * indicate retry.
  13.       */
  14.      do {
  15.          node.prev = pred = pred.prev;
  16.      } while (pred.waitStatus > 0);
  17.      pred.next = node;
  18. } else {
  19.      /*
  20.       * waitStatus must be 0 or PROPAGATE.  Indicate that we
  21.       * need a signal, but don't park yet.  Caller will need to
  22.       * retry to make sure it cannot acquire before parking.
  23.       */
  24.      compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  25. }
  26. return false;
  27. }
复制代码

  线程每次被唤醒时,都要进行中断检测,如果发现当前线程被中断,那么抛出InterruptedException并退出循环。从无限循环的代码可以看出,并不是被唤醒的线程一定能获得锁,必须调用tryAccquire重新竞争,因为锁是非公平的,有可能被新加入的线程获得,从而导致刚被唤醒的线程再次被阻塞,这个细节充分体现了“非公平”的精髓。

  线程释放锁过程:

  • 如果头结点head的waitStatus值为-1,则用CAS指令重置为0;
  • 找到waitStatus值小于0的节点s,通过LockSupport.unpark(s.thread)唤醒线程。

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

  10. /*
  11.   * Thread to unpark is held in successor, which is normally
  12.   * just the next node.  But if cancelled or apparently null,
  13.   * traverse backwards from tail to find the actual
  14.   * non-cancelled successor.
  15.   */
  16. Node s = node.next;
  17. if (s == null || s.waitStatus > 0) {
  18.      s = null;
  19.      for (Node t = tail; t != null && t != node; t = t.prev)
  20.          if (t.waitStatus <= 0)
  21.              s = t;
  22. }
  23. if (s != null)
  24.      LockSupport.unpark(s.thread);
  25. }
复制代码

  总结

  Doug Lea大神的思路跳跃的太快,把CAS指令玩的出神入化,以至于有些逻辑反反复复debug很多次才明白。

相关帖子

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关于我们
联系我们
  • 电话:010-86393388
  • 邮件:udn@yonyou.com
  • 地址:北京市海淀区北清路68号
移动客户端下载
关注我们
  • 微信公众号:yonyouudn
  • 扫描右侧二维码关注我们
  • 专注企业互联网的技术社区
版权所有:用友网络科技股份有限公司82041 京ICP备05007539号-11 京公网网备安1101080209224 Powered by Discuz!
快速回复 返回列表 返回顶部