服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - 基于ReentrantLock的实现原理讲解

基于ReentrantLock的实现原理讲解

2022-01-21 11:47一撸向北 Java教程

这篇文章主要介绍了ReentrantLock的实现原理,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教

java.util.concurrent包中的工具实现核心都是AQS,了解ReentrantLock的实现原理,需要先分析AQS以及AQS与ReentrantLock的关系。

这篇文章中分析了ReentrantLock#lock与ReentrantLock#unlock的实现,对于Condition的实现分析,另外文章再讲,基本上大同小异。

 

ReentrantLock实现核心CAQS(AbstractQueuedSynchronizer)

AQS,队列同步器,在juc包中的工具类都是依赖于AQS来实现同步控制,看一张AQS的结构图。

基于ReentrantLock的实现原理讲解

同步控制中主要使用到的信息如上图所示。AQS可以被当做是一个同步监视器的实现,并且具有排队功能。当线程尝试获取AQS的锁时,如果AQS已经被别的线程获取锁,那么将会新建一个Node节点,并且加入到AQS的等待队列中,这个队列也由AQS本身自己维护。当锁被释放时,唤醒下一个节点尝试获取锁。

  • 变量exclusiveOwnerThread在互斥模式下,表示当前持有锁的线程。
  • 变量tail指向等待获取AQS的锁的节点队列的最后一个
  • 变量head指向队列中head节点,head节点信息为空,不表示任何正在等待的线程。
  • 变量state表示AQS同步器的状态,在不同情况下含义可能不太一样,例如以下几种
    • 在ReentrantLock中,表示AQS的锁是否已经被占用获取,0:没有,>=1:已被获取,当大于1时表示被同一线程多次重入锁。
    • 在CountDownLatch中,表示计数还剩的次数,当到达0时,唤醒等待线程。
    • 在Semaphore中,表示AQS还可以被获取锁的次数,获取一次就减1,当到达0时,尝试获取的线程将会阻塞。

Node结构

Node节点是AQS管理的等待队列的节点元素,除了head节点之外,其他一个节点就代表一个正在等待线程的队列。Node一般的重要参数有几个。

  • prev 前置节点
  • next后置节点
  • thread 代表的线程
  • waitStatus节点的等待状态
    • 1表示节点已经取消,也就是线程可能已经中断,不需要再等待获取锁了,在后续代码中会处理跳过waitStatus等于1的节点
    • -1表示当前节点的后置节点代表的线程需要被挂起
    • -2表示当前线程正在等待的是Condition锁

 

ReentrantLock实现分析

二者关联

ReentrantLock实现核心是基于AQS,看下面一张图,分析AQS与ReentrantLock的关系。

基于ReentrantLock的实现原理讲解

从图中可以看出,ReentrantLock里面有最终两个内部类,FairSync和NonfairSync通过继承AbstractQueuedSynchronizer的功能,来实现两种同步互斥方案:公平锁和非公平锁。

在ReentrantLock中最终lock和unlock操作,都由FairSync和NonfairSync实际完成。

NonfairSync分析

下面看一个最简单利用ReentrantLock实现互斥的例子。

      ReentrantLock lock = new ReentrantLock();
		//尝试获取锁
      lock.lock();
      //获得锁后执行逻辑......
      //......
		//解锁
      lock.unlock();

在ReentrantLock的默认无参构造方法中,创建的是非公平锁

  public ReentrantLock() {
      sync = new NonfairSync();
  }

下面分析lock.lock();这句代码是如何实现同步互斥的。

NonfairSync#lock

点开lock.lock();源码,可以看到最终实际调用的是NonfairSync#lock,这是分析的入口。

NonfairSync#lock源码如下。

final void lock() {
  if (compareAndSetState(0, 1))//【step1】
      setExclusiveOwnerThread(Thread.currentThread());//【step2】
  else
      acquire(1);//【step3】
}

【step1】上面有提到,NonfairSync继承自AbstractQueuedSynchronizer,NonfairSync就是一个AQS,因此在步骤【1】其实就是利用CAS(一个原子性的比较并设置操作)尝试设置AQS的state为1。如果设置成功,表示获取锁成功;如果设置失败,表示state之前已经是>=1,已经被别的线程占用了AQS的锁,所示无法设置state为1,稍后会把线程加入到等待队列。

**非公平锁与公平锁:**对于NonfairSync非公平锁来说,线程只要执行lock请求,就会立马尝试获取锁,不会管AQS当前管理的等待队列中有没有正在等待的线程,这种操作是不公平的,没有先来后到;而稍后介绍的FairSync公平锁,则会在lock请求进行时,先判断AQS管理的等待队列中是否已经有正在等待的线程,有的话就是不尝试获取锁,直接进入等待队列,保证了公平性。

这一步需要熟悉的是CAS操作,分析一下compareAndSetState源码,如下。这一步利用unsafe包的cas操作,unsafe包类似一种java留给开发者的后门,可以用来直接操作内存数据,并且保证这个操作的原子性。在下面的代码中,stateOffset表示state比变量的内存偏移地址,用来寻找state变量内存位置。整个cas操作就是找到内存中当前的state变量值,并且与expect期待值比较,如果跟期待值相同,那么表示这个值是可以修改的,此时就会对state值进行更新;如果与期待值不一样,那么将不能进行更新操作。unsafe保证了比较与设置值的过程是原子性的,在这一步不会出现线程安全问题。

protected final boolean compareAndSetState(int expect, int update) {
  // See below for intrinsics setup to support this
  return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

【step2】操作是在设置state成功之后,表示当前线程获取AQS锁成功,需要设置AQS中的变量exclusiveOwnerThread为当前持有锁的线程,做保存记录。

【step3】当尝试获取锁失败的时候,就需要进行步骤3,执行acquire,进行再次尝试或者线程进入等待队列。

AbstractQueuedSynchronizer#acquire

当第一次尝试获取锁失败之后,执行【step3】acquire方法,这个方法可以讲一个尝试获取锁失败的线程放入AQS管理的等待队列进行等待,并且将线程挂起。实现逻辑在AbstractQueuedSynchronizer实现,NonfairSync通过继承AbstractQueuedSynchronizer获得等待队列管理的功能。

下面看源码。

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
}

NonfairSync#tryAcquireC锁重入实现

首先,执行tryAcquire再次尝试一次获取lock,tryAcquire是由子类实现,也就是这里调用NonfairSync#tryAcquire方法,跟踪调用,最终执行代码如下。

final boolean nonfairTryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
      //如果此时state已经变为1,再次尝试一次获取锁
      if (compareAndSetState(0, acquires)) {
          setExclusiveOwnerThread(current);
          return true;
      }
  }
  else if (current == getExclusiveOwnerThread()) {
      //否则判断当前持有AQS的锁的线程是不是当前请求获取锁的线程,是的话就进行锁重入。
      int nextc = c + acquires;
      if (nextc < 0) // overflow
          throw new Error("Maximum lock count exceeded");
      setState(nextc);
      return true;
  }
  return false;
}

对于NonfairSync#tryAcquire的实现,首先是重新尝试一次获取锁,如果还是获取不到,就尝试判断当前的是不是属于重入锁的情形。

对于重入锁的情形,就需要对state进行累加1,意思就是重入一次就对state加1。这样子,在解锁的时候,每次unlock就对state减一,等到state的值为0的时候,才能唤醒下一个等待线程。

如果获取成功,返回true;否则返回false,继续执行下一步acquireQueued(addWaiter(Node.EXCLUSIVE), arg)),添加一个Node节点进入AQS管理的等待队列。

AbstractQueuedSynchronizer#addWaiterC添加Node到等待队列

这个方法属于队列管理,也是由AbstractQueuedSynchronizer默认实现,NonfairSync继承获得。

查看addWaiter源码实现。

private Node addWaiter(Node mode) {
  //新建一个Node节点,mode传入表示当前线程之间竞争是互斥模式
  Node node = new Node(Thread.currentThread(), mode);
  // Try the fast path of enq; backup to full enq on failure
  //尝试添加当前新建Node到链表队列末尾
  Node pred = tail;
  if (pred != null) {
      node.prev = pred;
      //利用cas设置尾指针指向的节点为当前线新建节点
      if (compareAndSetTail(pred, node)) {
          pred.next = node;
          return node;
      }
  }
  //当前是空的没有任何正在等待的线程Node的时候,执行enq(node),初始化等待队列
  enq(node);
  return node;
}
private Node enq(final Node node) {
  for (;;) {
      Node t = tail;
      if (t == null) { // Must initialize
          //新建一个空的头节点
          if (compareAndSetHead(new Node()))
              tail = head;
      } else {
          //跟之前一样,利用cas这只当前新建节点为尾节点
          node.prev = t;
          if (compareAndSetTail(t, node)) {
              t.next = node;
              return t;
          }
      }
  }
}

以上操作,完成将一个节点加入队列操作。加入完成之后,返回这个新加入的节点Node给acquireQueued方法继续执行。

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))C锁竞争优化

上面既然都完成了等待节点如队列的操作,为什么不直接挂起线程进入等待呢?因此这里的设计者做了一个优化操作。acquireQueued方法其实就是为了减少线程挂起、唤醒次数而作的优化操作。

下面看看acquireQueued的代码实现。

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
      boolean interrupted = false;
      for (;;) {
          //获取当前竞争锁的线程Node的前置节点
          final Node p = node.predecessor();
          //【step1】
          if (p == head && tryAcquire(arg)) {
              setHead(node);
              p.next = null; // help GC
              failed = false;
              return interrupted;
          }
          //【step2】
          if (shouldParkAfterFailedAcquire(p, node) &&
              parkAndCheckInterrupt())
              interrupted = true;
      }
  } finally {
      if (failed)
          cancelAcquire(node);
  }
}

【step1】假如前置节点是head,那么表示当前线程是等待队列中最大优先级的等待线程,可以继续最后的尝试获取锁,因为很有可能会获取到锁,从而避免线程挂起、唤醒,耗费资源,这里也算是最终一次尝试获取。

【step2】shouldParkAfterFailedAcquire(p, node)是检查当前是否有必要挂起,前面我们说过,只有当前置节点的waitStatus是-1的时候才会挂起当前节点代表的线程。当shouldParkAfterFailedAcquire(p, node)返回true的时候,就可以执行parkAndCheckInterrupt()来挂起线程。

shouldParkAfterFailedAcquire(p, node)源码如下。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  if (ws == Node.SIGNAL)
      //前置节点是-1,返回true表示线程可挂起
      return true;
  if (ws > 0) {
      //前置节点大于0表示前置节点已经取消,那么进行跳过前置节点的操作,做链表的基本删除节点操作
      do {
          node.prev = pred = pred.prev;
      } while (pred.waitStatus > 0);
      pred.next = node;
  } else {
      //如果前置节点还是0,表示前置节点Node的waitStatus是初始值,需要设置为-1,然后外层循环重新执行shouldParkAfterFailedAcquire方法,即可挂起当前线程。
      compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}
private final boolean parkAndCheckInterrupt() {
  //阻塞挂起线程,等待唤醒
  LockSupport.park(this);
  //唤醒后,重置中断标记,线程中断标记位为不中断
  return Thread.interrupted();
}

FairSync分析

之前提到

**非公平锁与公平锁:**对于NonfairSync非公平锁来说,线程只要执行lock请求,就会立马尝试获取锁,不会管AQS当前管理的等待队列中有没有正在等待的线程,这种操作是不公平的,没有先来后到;而稍后介绍的FairSync公平锁,则会在lock请求进行时,先判断AQS管理的等待队列中是否已经有正在等待的线程,有的话就是不尝试获取锁,直接进入等待队列,保证了公平性。

所以两者的实现区别在于第一次尝试lock的动作不一样。

FairSync#lock

final void lock() {
  acquire(1);
}
public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
      selfInterrupt();
}

最终差异体现在FairSync#tryAcquire的实现(第一次尝试获取lock)

protected final boolean tryAcquire(int acquires) {
  final Thread current = Thread.currentThread();
  int c = getState();
  if (c == 0) {
      //hasQueuedPredecessors判断队列是否还有别的线程在等待锁,没有的话就尝试获取lock
      //如果有别的线程在等待锁,就不会尝试获取lock;下面如果也不是重入的情况的话就直接进入等待队列
      if (!hasQueuedPredecessors() &&
          compareAndSetState(0, acquires)) {
          setExclusiveOwnerThread(current);
          return true;
      }
  }
  else if (current == getExclusiveOwnerThread()) {
      int nextc = c + acquires;
      if (nextc < 0)
          throw new Error("Maximum lock count exceeded");
      setState(nextc);
      return true;
  }
  return false;
}

AbstractQueuedSynchronizer#release --AQS解锁操作

AQS中定义了解锁操作的模板方法,tryRelease(arg)是不同AQS子类实现,对state的多样化操作。例如ReentrantLock中的tryRelease(arg)操作比较明显的就是对state减一。

tryRelease(arg)返回结果表示本次操作后是否需要唤醒下一个等待节点。对于ReentrantLock就是state减一之后是否变为0了。如果需要唤醒下一个节点的线程,那么判断一下Head有没有下一个要唤醒的节点线程,有的话就进行唤醒操作unparkSuccessor(h);

public final boolean release(int arg) {
  if (tryRelease(arg)) {
      Node h = head;
      if (h != null && h.waitStatus != 0)
          unparkSuccessor(h);
      return true;
  }
  return false;
}

ReentrantLock.Sync#tryReleaseC解锁实现

看一下ReentrantLock.Sync的tryRelease实现.是如何为state减一 的。。

protected final boolean tryRelease(int releases) {
  //获取state减掉realease,对于ReentrantLock就是默认减一
  int c = getState() - releases;
  if (Thread.currentThread() != getExclusiveOwnerThread())
      throw new IllegalMonitorStateException();
  boolean free = false;
  if (c == 0) {
      //如果减到0了,那么久释放锁
      free = true;
      //设置持有线程为null
      setExclusiveOwnerThread(null);
  }
  //设置state为新的
  setState(c);
  return free;
}

至于这里设置state为啥不同cas操作,因为

  if (Thread.currentThread() != getExclusiveOwnerThread())
      throw new IllegalMonitorStateException();

所以永远只有持有锁的线程会做解锁减一的操作,state设置是线程安全的。

 

注意一下

其实这里还没分析Condition的实现原理,篇幅太长,下次另开文章分析。以上为个人经验,希望能给大家一个参考,也希望大家多多支持服务器之家。

原文链接:https://blog.csdn.net/qq_20597727/article/details/86263237

延伸 · 阅读

精彩推荐