Java并发系列之AbstractQueuedSynchronizer源码分析(共享模式)

通过上一篇的分析,我们知道了独占模式获取锁有三种方式,分别是不响应线程中断获取,响应线程中断获取,设置超时时间获取。在共享模式下获取锁的方式也是这三种,而且基本上都是大同小异,我们搞清楚了一种就能很快的理解其他的方式。虽然说AbstractQueuedSynchronizer源码有一千多行,但是重复的也比较多,所以读者不要刚开始的时候被吓到,只要耐着性子去看慢慢的自然能够渐渐领悟。就我个人经验来说,阅读AbstractQueuedSynchronizer源码有几个比较关键的地方需要弄明白,分别是独占模式和共享模式的区别,结点的等待状态,以及对条件队列的理解。理解了这些要点那么后续源码的阅读将会轻松很多。当然这些在我的《Java并发系列[1]----AbstractQueuedSynchronizer源码分析之概要分析》这篇文章里都有详细的介绍,读者可以先去查阅。本篇对于共享模式的分析也是分为三种获取锁的方式和一种释放锁的方式。

1. 不响应线程中断的获取

//以不可中断模式获取锁(共享模式)
public final void acquireShared(int arg) {
  //1.尝试去获取锁
  if (tryAcquireShared(arg) < 0) {
    //2.如果获取失败就进入这个方法
    doAcquireShared(arg);
  }
}

//尝试去获取锁(共享模式)
//负数:表示获取失败
//零值:表示当前结点获取成功, 但是后继结点不能再获取了
//正数:表示当前结点获取成功, 并且后继结点同样可以获取成功
protected int tryAcquireShared(int arg) {
  throw new UnsupportedOperationException();
}

调用acquireShared方法是不响应线程中断获取锁的方式。在该方法中,首先调用tryAcquireShared去尝试获取锁,tryAcquireShared方法返回一个获取锁的状态,这里AQS规定了返回状态若是负数代表当前结点获取锁失败,若是0代表当前结点获取锁成功,但后继结点不能再获取了,若是正数则代表当前结点获取锁成功,并且这个锁后续结点也同样可以获取成功。子类在实现tryAcquireShared方法获取锁的逻辑时,返回值需要遵守这个约定。如果调用tryAcquireShared的返回值小于0,就代表这次尝试获取锁失败了,接下来就调用doAcquireShared方法将当前线程添加进同步队列。我们看到doAcquireShared方法。

//在同步队列中获取(共享模式)
private void doAcquireShared(int arg) {
  //添加到同步队列中
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      //获取当前结点的前继结点
      final Node p = node.predecessor();
      //如果前继结点为head结点就再次尝试去获取锁
      if (p == head) {
        //再次尝试去获取锁并返回获取状态
        //r < 0, 表示获取失败
        //r = 0, 表示当前结点获取成功, 但是后继结点不能再获取了
        //r > 0, 表示当前结点获取成功, 并且后继结点同样可以获取成功
        int r = tryAcquireShared(arg);
        if (r >= 0) {
          //到这里说明当前结点已经获取锁成功了, 此时它会将锁的状态信息传播给后继结点
          setHeadAndPropagate(node, r);
          p.next = null;
          //如果在线程阻塞期间收到中断请求, 就在这一步响应该请求
          if (interrupted) {
            selfInterrupt();
          }
          failed = false;
          return;
        }
      }
      //每次获取锁失败后都会判断是否可以将线程挂起, 如果可以的话就会在parkAndCheckInterrupt方法里将线程挂起
      if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
        interrupted = true;
      }
    }
  } finally {
    if (failed) {
      cancelAcquire(node);
    }
  }
}

进入doAcquireShared方法首先是调用addWaiter方法将当前线程包装成结点放到同步队列尾部。这个添加结点的过程我们在讲独占模式时讲过,这里就不再讲了。结点进入同步队列后,如果它发现在它前面的结点就是head结点,因为head结点的线程已经获取锁进入房间里面了,那么下一个获取锁的结点就轮到自己了,所以当前结点先不会将自己挂起,而是再一次去尝试获取锁,如果前面那人刚好释放锁离开了,那么当前结点就能成功获得锁,如果前面那人还没有释放锁,那么就会调用shouldParkAfterFailedAcquire方法,在这个方法里面会将head结点的状态改为SIGNAL,只有保证前面结点的状态为SIGNAL,当前结点才能放心的将自己挂起,所有线程都会在parkAndCheckInterrupt方法里面被挂起。如果当前结点恰巧成功的获取了锁,那么接下来就会调用setHeadAndPropagate方法将自己设置为head结点,并且唤醒后面同样是共享模式的结点。下面我们看下setHeadAndPropagate方法具体的操作。

//设置head结点并传播锁的状态(共享模式)
private void setHeadAndPropagate(Node node, int propagate) {
  Node h = head;
  //将给定结点设置为head结点
  setHead(node);
  //如果propagate大于0表明锁可以获取了
  if (propagate > 0 || h == null || h.waitStatus < 0) {
    //获取给定结点的后继结点
    Node s = node.next;
    //如果给定结点的后继结点为空, 或者它的状态是共享状态
    if (s == null || s.isShared()) {
      //唤醒后继结点
      doReleaseShared();
    }
  }
}

//释放锁的操作(共享模式)
private void doReleaseShared() {
  for (;;) {
    //获取同步队列的head结点
    Node h = head;
    if (h != null && h != tail) {
      //获取head结点的等待状态
      int ws = h.waitStatus;
      //如果head结点的状态为SIGNAL, 表明后面有人在排队
      if (ws == Node.SIGNAL) {
        //先把head结点的等待状态更新为0
        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
          continue;
        }
        //再去唤醒后继结点
        unparkSuccessor(h);
       //如果head结点的状态为0, 表明此时后面没人在排队, 就只是将head状态修改为PROPAGATE
      }else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
        continue;
      }
    }
    //只有保证期间head结点没被修改过才能跳出循环
    if (h == head) {
      break;
    }
  }
}

调用setHeadAndPropagate方法首先将自己设置成head结点,然后再根据传入的tryAcquireShared方法的返回值来决定是否要去唤醒后继结点。前面已经讲到当返回值大于0就表明当前结点成功获取了锁,并且后面的结点也可以成功获取锁。这时当前结点就需要去唤醒后面同样是共享模式的结点,注意,每次唤醒仅仅只是唤醒后一个结点,如果后一个结点不是共享模式的话,当前结点就直接进入房间而不会再去唤醒更后面的结点了。共享模式下唤醒后继结点的操作是在doReleaseShared方法进行的,共享模式和独占模式的唤醒操作基本也是相同的,都是去找到自己座位上的牌子(等待状态),如果牌子上为SIGNAL表明后面有人需要让它帮忙唤醒,如果牌子上为0则表明队列此时并没有人在排队。在独占模式下是如果发现没人在排队就直接离开队列了,而在共享模式下如果发现队列后面没人在排队,当前结点在离开前仍然会留个小纸条(将等待状态设置为PROPAGATE)告诉后来的人这个锁的可获取状态。那么后面来的人在尝试获取锁的时候可以根据这个状态来判断是否直接获取锁。

2. 响应线程中断的获取

//以可中断模式获取锁(共享模式)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  //首先判断线程是否中断, 如果是则抛出异常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //1.尝试去获取锁
  if (tryAcquireShared(arg) < 0) {
    //2. 如果获取失败则进人该方法
    doAcquireSharedInterruptibly(arg);
  }
}

//以可中断模式获取(共享模式)
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  //将当前结点插入同步队列尾部
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
    for (;;) {
      //获取当前结点的前继结点
      final Node p = node.predecessor();
      if (p == head) {
        int r = tryAcquireShared(arg);
        if (r >= 0) {
          setHeadAndPropagate(node, r);
          p.next = null;
          failed = false;
          return;
        }
      }
      if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
        //如果线程在阻塞过程中收到过中断请求, 那么就会立马在这里抛出异常
        throw new InterruptedException();
      }
    }
  } finally {
    if (failed) {
      cancelAcquire(node);
    }
  }
}

响应线程中断获取锁的方式和不响应线程中断获取锁的方式在流程上基本是相同的,唯一的区别就是在哪里响应线程的中断请求。在不响应线程中断获取锁时,线程从parkAndCheckInterrupt方法中被唤醒,唤醒后就立马返回是否收到中断请求,即使是收到了中断请求也会继续自旋直到获取锁后才响应中断请求将自己给挂起。而响应线程中断获取锁会才线程被唤醒后立马响应中断请求,如果在阻塞过程中收到了线程中断就会立马抛出InterruptedException异常。

3. 设置超时时间的获取

//以限定超时时间获取锁(共享模式)
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //1.调用tryAcquireShared尝试去获取锁
  //2.如果获取失败就调用doAcquireSharedNanos
  return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout);
}

//以限定超时时间获取锁(共享模式)
private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
  long lastTime = System.nanoTime();
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
    for (;;) {
      //获取当前结点的前继结点
      final Node p = node.predecessor();
      if (p == head) {
        int r = tryAcquireShared(arg);
        if (r >= 0) {
          setHeadAndPropagate(node, r);
          p.next = null;
          failed = false;
          return true;
        }
      }
      //如果超时时间用完了就结束获取, 并返回失败信息
      if (nanosTimeout <= 0) {
        return false;
      }
      //1.检查是否满足将线程挂起要求(保证前继结点状态为SIGNAL)
      //2.检查超时时间是否大于自旋时间
      if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) {
        //若满足上面两个条件就将当前线程挂起一段时间
        LockSupport.parkNanos(this, nanosTimeout);
      }
      long now = System.nanoTime();
      //超时时间每次减去获取锁的时间
      nanosTimeout -= now - lastTime;
      lastTime = now;
      //如果在阻塞时收到中断请求就立马抛出异常
      if (Thread.interrupted()) {
        throw new InterruptedException();
      }
    }
  } finally {
    if (failed) {
      cancelAcquire(node);
    }
  }
}

如果看懂了上面两种获取方式,再来看设置超时时间的获取方式就会很轻松,基本流程都是一样的,主要是理解超时的机制是怎样的。如果第一次获取锁失败会调用doAcquireSharedNanos方法并传入超时时间,进入方法后会根据情况再次去获取锁,如果再次获取失败就要考虑将线程挂起了。这时会判断超时时间是否大于自旋时间,如果是的话就会将线程挂起一段时间,否则就继续尝试获取,每次获取锁之后都会将超时时间减去获取锁的时间,一直这样循环直到超时时间用尽,如果还没有获取到锁的话就会结束获取并返回获取失败标识。在整个期间线程是响应线程中断的。

4. 共享模式下结点的出队操作

//释放锁的操作(共享模式)
public final boolean releaseShared(int arg) {
  //1.尝试去释放锁
  if (tryReleaseShared(arg)) {
    //2.如果释放成功就唤醒其他线程
    doReleaseShared();
    return true;
  }
  return false;
}

//尝试去释放锁(共享模式)
protected boolean tryReleaseShared(int arg) {
  throw new UnsupportedOperationException();
}

//释放锁的操作(共享模式)
private void doReleaseShared() {
  for (;;) {
    //获取同步队列的head结点
    Node h = head;
    if (h != null && h != tail) {
      //获取head结点的等待状态
      int ws = h.waitStatus;
      //如果head结点的状态为SIGNAL, 表明后面有人在排队
      if (ws == Node.SIGNAL) {
        //先把head结点的等待状态更新为0
        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
          continue;
        }
        //再去唤醒后继结点
        unparkSuccessor(h);
       //如果head结点的状态为0, 表明此时后面没人在排队, 就只是将head状态修改为PROPAGATE
      }else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
        continue;
      }
    }
    //只有保证期间head结点没被修改过才能跳出循环
    if (h == head) {
      break;
    }
  }
}

线程在房间办完事之后就会调用releaseShared方法释放锁,首先调用tryReleaseShared方法尝试释放锁,该方法的判断逻辑由子类实现。如果释放成功就调用doReleaseShared方法去唤醒后继结点。走出房间后它会找到原先的座位(head结点),看看座位上是否有人留了小纸条(状态为SIGNAL),如果有就去唤醒后继结点。如果没有(状态为0)就代表队列没人在排队,那么在离开之前它还要做最后一件事情,就是在自己座位上留下小纸条(状态设置为PROPAGATE),告诉后面的人锁的获取状态,整个释放锁的过程和独占模式唯一的区别就是在这最后一步操作。

注:以上全部分析基于JDK1.7,不同版本间会有差异,读者需要注意。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

您可能感兴趣的文章:

  • Java并发系列之Semaphore源码分析
  • Java并发系列之CountDownLatch源码分析
  • Java并发系列之CyclicBarrier源码分析
  • Java并发系列之ConcurrentHashMap源码分析
  • Java并发系列之AbstractQueuedSynchronizer源码分析(独占模式)
  • Java并发系列之AbstractQueuedSynchronizer源码分析(概要分析)
  • Java并发系列之ReentrantLock源码分析
  • Java并发系列之AbstractQueuedSynchronizer源码分析(条件队列)
  • Java并发系列之Semaphore源码分析
  • Java并发系列之ReentrantLock源码分析
(0)

相关推荐

  • Java并发系列之AbstractQueuedSynchronizer源码分析(独占模式)

    在上一篇<Java并发系列[1]----AbstractQueuedSynchronizer源码分析之概要分析>中我们介绍了AbstractQueuedSynchronizer基本的一些概念,主要讲了AQS的排队区是怎样实现的,什么是独占模式和共享模式以及如何理解结点的等待状态.理解并掌握这些内容是后续阅读AQS源码的关键,所以建议读者先看完我的上一篇文章再回过头来看这篇就比较容易理解.在本篇中会介绍在独占模式下结点是怎样进入同步队列排队的,以及离开同步队列之前会进行哪些操作.AQS为在独占模

  • Java并发系列之CountDownLatch源码分析

    CountDownLatch(闭锁)是一个很有用的工具类,利用它我们可以拦截一个或多个线程使其在某个条件成熟后再执行.它的内部提供了一个计数器,在构造闭锁时必须指定计数器的初始值,且计数器的初始值必须大于0.另外它还提供了一个countDown方法来操作计数器的值,每调用一次countDown方法计数器都会减1,直到计数器的值减为0时就代表条件已成熟,所有因调用await方法而阻塞的线程都会被唤醒.这就是CountDownLatch的内部机制,看起来很简单,无非就是阻塞一部分线程让其在达到某个条

  • Java并发之ReentrantLock类源码解析

    ReentrantLock内部由Sync类实例实现. Sync类定义于ReentrantLock内部. Sync继承于AbstractQueuedSynchronizer. AbstractQueuedSynchronizer继承于AbstractOwnableSynchronizer. AbstractOwnableSynchronizer类中只定义了一个exclusiveOwnerThread变量,表示当前拥有的线程. 除了Sync类,ReentrantLock内部还定义了两个实现类. No

  • Java并发系列之Semaphore源码分析

    Semaphore(信号量)是JUC包中比较常用到的一个类,它是AQS共享模式的一个应用,可以允许多个线程同时对共享资源进行操作,并且可以有效的控制并发数,利用它可以很好的实现流量控制.Semaphore提供了一个许可证的概念,可以把这个许可证看作公共汽车车票,只有成功获取车票的人才能够上车,并且车票是有一定数量的,不可能毫无限制的发下去,这样就会导致公交车超载.所以当车票发完的时候(公交车以满载),其他人就只能等下一趟车了.如果中途有人下车,那么他的位置将会空闲出来,因此如果这时其他人想要上车

  • Java并发系列之CyclicBarrier源码分析

    现实生活中我们经常会遇到这样的情景,在进行某个活动前需要等待人全部都齐了才开始.例如吃饭时要等全家人都上座了才动筷子,旅游时要等全部人都到齐了才出发,比赛时要等运动员都上场后才开始.在JUC包中为我们提供了一个同步工具类能够很好的模拟这类场景,它就是CyclicBarrier类.利用CyclicBarrier类可以实现一组线程相互等待,当所有线程都到达某个屏障点后再进行后续的操作.下图演示了这一过程. 在CyclicBarrier类的内部有一个计数器,每个线程在到达屏障点的时候都会调用await

  • Java并发系列之ReentrantLock源码分析

    在Java5.0之前,协调对共享对象的访问可以使用的机制只有synchronized和volatile.我们知道synchronized关键字实现了内置锁,而volatile关键字保证了多线程的内存可见性.在大多数情况下,这些机制都能很好地完成工作,但却无法实现一些更高级的功能,例如,无法中断一个正在等待获取锁的线程,无法实现限定时间的获取锁机制,无法实现非阻塞结构的加锁规则等.而这些更灵活的加锁机制通常都能够提供更好的活跃性或性能.因此,在Java5.0中增加了一种新的机制:Reentrant

  • Java并发系列之AbstractQueuedSynchronizer源码分析(条件队列)

    通过前面三篇的分析,我们深入了解了AbstractQueuedSynchronizer的内部结构和一些设计理念,知道了AbstractQueuedSynchronizer内部维护了一个同步状态和两个排队区,这两个排队区分别是同步队列和条件队列.我们还是拿公共厕所做比喻,同步队列是主要的排队区,如果公共厕所没开放,所有想要进入厕所的人都得在这里排队.而条件队列主要是为条件等待设置的,我们想象一下如果一个人通过排队终于成功获取锁进入了厕所,但在方便之前发现自己没带手纸,碰到这种情况虽然很无奈,但是它

  • Java并发系列之AbstractQueuedSynchronizer源码分析(概要分析)

    学习Java并发编程不得不去了解一下java.util.concurrent这个包,这个包下面有许多我们经常用到的并发工具类,例如:ReentrantLock, CountDownLatch, CyclicBarrier, Semaphore等.而这些类的底层实现都依赖于AbstractQueuedSynchronizer这个类,由此可见这个类的重要性.所以在Java并发系列文章中我首先对AbstractQueuedSynchronizer这个类进行分析,由于这个类比较重要,而且代码比较长,为了

  • Java并发系列之ConcurrentHashMap源码分析

    我们知道哈希表是一种非常高效的数据结构,设计优良的哈希函数可以使其上的增删改查操作达到O(1)级别.Java为我们提供了一个现成的哈希结构,那就是HashMap类,在前面的文章中我曾经介绍过HashMap类,知道它的所有方法都未进行同步,因此在多线程环境中是不安全的.为此,Java为我们提供了另外一个HashTable类,它对于多线程同步的处理非常简单粗暴,那就是在HashMap的基础上对其所有方法都使用synchronized关键字进行加锁.这种方法虽然简单,但导致了一个问题,那就是在同一时间

  • Java并发系列之AbstractQueuedSynchronizer源码分析(共享模式)

    通过上一篇的分析,我们知道了独占模式获取锁有三种方式,分别是不响应线程中断获取,响应线程中断获取,设置超时时间获取.在共享模式下获取锁的方式也是这三种,而且基本上都是大同小异,我们搞清楚了一种就能很快的理解其他的方式.虽然说AbstractQueuedSynchronizer源码有一千多行,但是重复的也比较多,所以读者不要刚开始的时候被吓到,只要耐着性子去看慢慢的自然能够渐渐领悟.就我个人经验来说,阅读AbstractQueuedSynchronizer源码有几个比较关键的地方需要弄明白,分别是

  • Java并发编程之Condition源码分析(推荐)

    Condition介绍 上篇文章讲了ReentrantLock的加锁和释放锁的使用,这篇文章是对ReentrantLock的补充.ReentrantLock#newCondition()可以创建Condition,在ReentrantLock加锁过程中可以利用Condition阻塞当前线程并临时释放锁,待另外线程获取到锁并在逻辑后通知阻塞线程"激活".Condition常用在基于异步通信的同步机制实现中,比如dubbo中的请求和获取应答结果的实现. 常用方法 Condition中主要的

随机推荐