Java源码解析之可重入锁ReentrantLock

本文基于jdk1.8进行分析。

reentrantlock是一个可重入锁，在concurrenthashmap中使用了reentrantlock。

首先看一下源码中对reentrantlock的介绍。如下图。reentrantlock是一个可重入的排他锁，它和synchronized的方法和代码有着相同的行为和语义，但有更多的功能。reentrantlock是被最后一个成功lock锁并且还没有unlock的线程拥有着。如果锁没有被别的线程拥有，那么一个线程调用lock方法，就会成功获取锁并返回。如果当前线程已经拥有该锁，那么lock方法会立刻返回。这个可以通过isheldbycurrentthread方法和getholdcount方法进行验证。除了这部分介绍外，类前面的javadoc文档很长，就不在这里全部展开。随着后面介绍源码，会一一涉及到。

/**
 * a reentrant mutual exclusion {@link lock} with the same basic
 * behavior and semantics as the implicit monitor lock accessed using
 * {@code synchronized} methods and statements, but with extended
 * capabilities.
 * <p>a {@code reentrantlock} is <em>owned</em> by the thread last
 * successfully locking, but not yet unlocking it. a thread invoking
 * {@code lock} will return, successfully acquiring the lock, when
 * the lock is not owned by another thread. the method will return
 * immediately if the current thread already owns the lock. this can
 * be checked using methods {@link #isheldbycurrentthread}, and {@link
 * #getholdcount}.

首先看一下成员变量，如下图。reentrantlock只有一个成员变量sync，即同步器，这个同步器提供所有的机制。sync是abstractqueuedsynchronizer的子类，同时，sync有2个子类，nonfairsync和fairsync，分别是非公平锁和公平锁。sync，nonfairesync和fairsync的具体实现后面再讲。

  /** synchronizer providing all implementation mechanics **/
  private final sync sync;

下面看一下构造函数。如下图。可以看到，reentrantlock默认是非公平锁，它可以通过参数，指定初始化为公平锁或非公平锁。

  /**
   * creates an instance of {@code reentrantlock}.
   * this is equivalent to using {@code reentrantlock(false)}.
   **/
  public reentrantlock() {
    sync = new nonfairsync();
  }
  /**
   * creates an instance of {@code reentrantlock} with the
   * given fairness policy.
   * @param fair {@code true} if this lock should use a fair ordering policy
   **/
  public reentrantlock(boolean fair) {
    sync = fair ? new fairsync() : new nonfairsync();
  }

下面看一下reentrantlock的主要方法。首先是lock方法。如下图。lock方法的实现很简单，就是调用sync的lock方法。而sync的lock方法是个抽象的，具体实现在nonfairsync和fairsync中。这里我们先不展开讲，而是先读一下lock方法的注释，看看它的作用。lock方法的作用是获取该锁。分为3种情况。

1，如果锁没有被别的线程占有，那么当前线程就可以获取到锁并立刻返回，并把锁计数设置为1。

2，如果当前线程已经占有该锁了，那么就会把锁计数加1，立刻返回。

3，如果锁被另一个线程占有了，那么当前线程就无法再被线程调度，并且开始睡眠，直到获取到锁，在获取到到锁时，会把锁计数设置为1。

lockinterruptibly方法与lock功能类似，但lockinterruptibly方法在等待的过程中，可以响应中断。

  /**
   * acquires the lock.
   * <p>acquires the lock if it is not held by another thread and returns
   * immediately, setting the lock hold count to one.
   * <p>if the current thread already holds the lock then the hold
   * count is incremented by one and the method returns immediately.
   * <p>if the lock is held by another thread then the
   * current thread becomes disabled for thread scheduling
   * purposes and lies dormant until the lock has been acquired,
   * at which time the lock hold count is set to one.
   **/
  public void lock() {
    sync.lock();
  }
  public void lockinterruptibly() throws interruptedexception {
    sync.acquireinterruptibly(1);
  }

下面，详细看一下非公平锁和公平锁中对lock函数的实现。如下图。下图同时列出了公平锁和非公平锁中lock的实现逻辑。从注释和代码逻辑中，都可以看出，非公平锁进行lock时，先尝试立刻闯入（抢占），如果成功，则获取到锁，如果失败，再执行通常的获取锁的行为，即acquire(1)。

    /**
     * 非公平锁中的lock
     * performs lock. try immediate barge, backing up to normal
     * acquire on failure.
     **/
    final void lock() {
      if (compareandsetstate(0, 1))
        setexclusiveownerthread(thread.currentthread());
      else
        acquire(1);
    }
    //公平锁中的lock
    final void lock() {
      acquire(1);
    }

那么，我们首先了解下，非公平锁“尝试立刻闯入”，究竟做了什么。稍后再继续讲解通常的获取锁的行为。下图是立即闯入行为compareandsetstate(0, 1)的实现。从compareandsetstate函数的注释中，可以知道，如果同步状态值与期望值相等，那么就把它的值设置为updated值。否则同步状态值与期望值不相等，则返回false。这个操作和volatile有着相同的内存语义，也就是说，这个操作对其他线程是可见的。compareandsetstate函数注释里描述的功能，是通过unsafe.compareandswapint方法实现的，而unsafe.compareandswapint是一个native方法，是用c++实现的。那么继续追问，c++底层是怎么实现的？c++底层是通过cas指令来实现的。什么是cas指令呢？来自*的解释是，cas，比较和交换，compare and swap，是用用于实现多线程原子同步的指令。它将内存位置的内容和给定值比较，只有在相同的情况下，将该内存的值设置为新的给定值。这个操作是原子操作。那么继续追问，cas指令的原子性，是如何实现的呢？我们都知道指令时cpu来执行的，在多cpu系统中，内存是共享的，内存和多个cpu都挂在总线上，当一个cpu执行cas指令时，它会先将总线lock位点设置为高电平。如果别的cpu也要执行cas执行，它会发现总线lock位点已经是高电平了，则无法执行cas执行。cpu通过lock保证了指令的原子执行。

现在来看一下非公平锁的lock行为，compareandsetstate(0, 1)，它期望锁状态为0，即没有别的线程占用，并把新状态设置为1，即标记为占用状态。如果成功，则非公平锁成功抢到锁，之后setexclusiveownerthread，把自己设置为排他线程。非公平锁这小子太坏了。如果抢占失败，则执行与公平锁相同的操作。

  /**
   * atomically sets synchronization state to the given updated
   * value if the current state value equals the expected value.
   * this operation has memory semantics of a {@code volatile} read
   * and write.
   * @param expect the expected value
   * @param update the new value
   * @return {@code true} if successful. false return indicates that the actual
   *     value was not equal to the expected value.
   **/
  protected final boolean compareandsetstate(int expect, int update) {
    // see below for intrinsics setup to support this
    return unsafe.compareandswapint(this, stateoffset, expect, update);
  }
  public final native boolean compareandswapint(object var1, long var2, int var4, int var5);

下面看一下公平锁获取锁时的行为。如下图。这部分的逻辑有些多，请阅读代码中的注释进行理解。

  /**
   * 公平锁的lock
   **/
  final void lock() {
    acquire(1);
  }
  /**
   * acquires in exclusive mode, ignoring interrupts. implemented
   * by invoking at least once {@link #tryacquire},
   * returning on success. otherwise the thread is queued, possibly
   * repeatedly blocking and unblocking, invoking {@link
   * #tryacquire} until success. this method can be used
   * to implement method {@link lock#lock}.
   * @param arg the acquire argument. this value is conveyed to
   *    {@link #tryacquire} but is otherwise uninterpreted and
   *    can represent anything you like.
   **/
  public final void acquire(int arg) {
    /**
     * acquire首先进行tryacquire()操作。如果tryacquire()成功时则获取到锁，即刻返回。
     * 如果tryacquire()false时，会执行acquirequeued(addwaiter(node.exclusive), arg)
     * 操作。如果acquirequeued(addwaiter(node.exclusive), arg)true时，则当前线程中断自己。
     * 如果acquirequeued(addwaiter(node.exclusive), arg)false，则返回。
     * 其中tryacquire()操作在nonfairsync中和fairsync中实现又有所区别。
     **/
    if (!tryacquire(arg) &&
        acquirequeued(addwaiter(node.exclusive), arg))
      selfinterrupt();
  }
  /**
   * nonfairsync中的tryacquire。
   * @param acquires
   * @return
   **/
  protected final boolean tryacquire(int acquires) {
    return nonfairtryacquire(acquires);
  }
  /**
   * performs non-fair trylock. tryacquire is implemented in
   * subclasses, but both need nonfair try for trylock method.
   **/
  final boolean nonfairtryacquire(int acquires) {
    final thread current = thread.currentthread();
    //首先获取当前同步状态值
    int c = getstate();
    if (c == 0) {
      //c为0，表示目前没有线程占用锁。没有线程占用锁时，当前线程尝试抢锁，如果抢锁成功，则返回true。
      if (compareandsetstate(0, acquires)) {
        setexclusiveownerthread(current);
        return true;
      }
    }
    else if (current == getexclusiveownerthread()) {
      //c不等于0时表示锁被线程占用。如果是当前线程占用了，则将锁计数加上acquires，并返回true。
      int nextc = c + acquires;
      if (nextc < 0) // overflow
        throw new error("maximum lock count exceeded");
      setstate(nextc);
      return true;
    }
    //以上情况都不是时，返回false，表示非公平抢锁失败。
    return false;
  }
  /**
   * fair version of tryacquire. don't grant access unless
   * recursive call or no waiters or is first.
   * 这个是公平版本的tryacquire
   **/
  protected final boolean tryacquire(int acquires) {
    final thread current = thread.currentthread();
    int c = getstate();
    if (c == 0) {
      //c=0时表示锁未被占用。这里是先判断队列中前面是否有别的线程。没有别的线程时，才进行cas操作。
      //公平锁之所以公平，正是因为这里。它发现锁未被占用时，首先判断等待队列中是否有别的线程已经在等待了。
      //而非公平锁，发现锁未被占用时，根本不管队列中的排队情况，上来就抢。
      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;
  }
  /**
   * acquires in exclusive uninterruptible mode for thread already in
   * queue. used by condition wait methods as well as acquire.
   * 当抢锁失败时，先执行addwaiter(node.exclusive)，将当前线程加入等待队列，再执行该方法。
   * 该方法的作用是中断当前线程，并进行检查，知道当前线程是队列中的第一个线程，并且抢锁成功时，
   * 该方法返回。
   * @param node the node
   * @param arg the acquire argument
   * @return {@code true} if interrupted while waiting
   **/
  final boolean acquirequeued(final node node, int arg) {
    boolean failed = true;
    try {
      boolean interrupted = false;
      for (;;) {
        final node p = node.predecessor();
        if (p == head && tryacquire(arg)) {
          sethead(node);
          p.next = null; // help gc
          failed = false;
          return interrupted;
        }
        if (shouldparkafterfailedacquire(p, node) &&
            parkandcheckinterrupt())
          interrupted = true;
      }
    } finally {
      if (failed)
        cancelacquire(node);
    }
  }

接下来是trylock方法。代码如下。从注释中我们可以理解到，只有当调用trylock时锁没有被别的线程占用，trylock才会获取锁。如果锁没有被另一个线程占用，那么就获取锁，并立刻返回true，并把锁计数设置为1. 甚至在锁被设置为公平排序的情况下，若果锁可用，调用trylock会立刻获取锁，而不管有没有别的线程在等待锁了。从这里我们总结出，不管可重入锁是公平锁还是非公平锁，trylock方法只会是非公平的。

/**
   * acquires the lock only if it is not held by another thread at the time
   * of invocation.
   * <p>acquires the lock if it is not held by another thread and
   * returns immediately with the value {@code true}, setting the
   * lock hold count to one. even when this lock has been set to use a
   * fair ordering policy, a call to {@code trylock()} <em>will</em>
   * immediately acquire the lock if it is available, whether or not
   * other threads are currently waiting for the lock.
   * this "barging" behavior can be useful in certain
   * circumstances, even though it breaks fairness. if you want to honor
   * the fairness setting for this lock, then use
   * {@link #trylock(long, timeunit) trylock(0, timeunit.seconds) }
   * which is almost equivalent (it also detects interruption).
   * <p>if the current thread already holds this lock then the hold
   * count is incremented by one and the method returns {@code true}.
   * <p>if the lock is held by another thread then this method will return
   * immediately with the value {@code false}.
   * @return {@code true} if the lock was free and was acquired by the
   *     current thread, or the lock was already held by the current
   *     thread; and {@code false} otherwise
   **/
  public boolean trylock() {
    return sync.nonfairtryacquire(1);
  }
  public boolean trylock(long timeout, timeunit unit)
      throws interruptedexception {
    return sync.tryacquirenanos(1, unit.tonanos(timeout));
  }

接下来是释放锁的方法unlock。代码如下。unlock方式的实现，是以参数1来调用sync.release方法。而release方法是如何实现的呢？release方法首先会调用tryrelease方法，如果tryrelease成功，则唤醒后继者线程。而tryrelease的实现过程十分清晰，首先获取锁状态，锁状态减去参数（放锁次数），得到新状态。然后判断持有锁的线程是否为当前线程，如果不是当前线程，则抛出illegalmonitorstateexception。然后判断，如果新状态为0，说明放锁成功，则把持有锁的线程设置为null，并返回true。如果新状态不为0，则返回false。从tryrelease的返回值来看，它返回的true或false，指的是否成功的释放了该锁。成功的释放该锁的意思是彻底释放锁，别的线程就可以获取锁了。这里要认识到，即便tryrelease返回false，它也只是说明了锁没有完全释放，本次调用的这个释放次数值，依然是释放成功的。

  /**
   * attempts to release this lock.
   * <p>if the current thread is the holder of this lock then the hold
   * count is decremented. if the hold count is now zero then the lock
   * is released. if the current thread is not the holder of this
   * lock then {@link illegalmonitorstateexception} is thrown.
   * @throws illegalmonitorstateexception if the current thread does not
   *     hold this lock
   **/
  public void unlock() {
    sync.release(1);
  }
  /**
   * releases in exclusive mode. implemented by unblocking one or
   * more threads if {@link #tryrelease} returns true.
   * this method can be used to implement method {@link lock#unlock}.
   * @param arg the release argument. this value is conveyed to
   *    {@link #tryrelease} but is otherwise uninterpreted and
   *    can represent anything you like.
   * @return the value returned from {@link #tryrelease}
   **/
  public final boolean release(int arg) {
    if (tryrelease(arg)) {
      node h = head;
      if (h != null && h.waitstatus != 0)
        unparksuccessor(h);
      return true;
    }
    return false;
  }
  protected final boolean tryrelease(int releases) {
      int c = getstate() - releases;
      if (thread.currentthread() != getexclusiveownerthread())
        throw new illegalmonitorstateexception();
      boolean free = false;
      if (c == 0) {
        free = true;
        setexclusiveownerthread(null);
      }
      setstate(c);
      return free;
    }
  /**
   * wakes up node's successor, if one exists.
   * @param node the node
   **/
  private void unparksuccessor(node node) {
    /**
     * if status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling. it is ok if this
     * fails or if status is changed by waiting thread.
     **/
    int ws = node.waitstatus;
    if (ws < 0)
      compareandsetwaitstatus(node, ws, 0);
    /**
     * thread to unpark is held in successor, which is normally
     * just the next node. but if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     **/
    node s = node.next;
    if (s == null || s.waitstatus > 0) {
      s = null;
      for (node t = tail; t != null && t != node; t = t.prev)
        if (t.waitstatus <= 0)
          s = t;
    }
    if (s != null)
      locksupport.unpark(s.thread);
  }

接下来是newcondition方法。关于condition这里不展开介绍，只是了解下该方法的作用。如下图。该方法返回一个和这个锁实例一起使用的condition实例。返回的condition实例支持和object的监控方法例如wait-notify和notifyall相同的用法。

• 1，如果没有获取锁，调用condition的await，signal，signalall方法的任何一个时，会抛出illegalmonitorstateexception异常。
• 2，调用condition的await方法时，锁也会释放，在await返回之前，锁会被重新获取，并且锁计数会恢复到调用await方法时的值。
• 3，如果一个线程在等待的过程中被中断了，那么等待就会结束，并抛出interruptedexception异常，线程的中断标志位会被清理。
• 4，等待的线程以fifo的顺序被唤醒。
• 5，从await方法返回的线程们的获取到锁的顺序，和线程最开始获取锁的顺序相同，这是未指定情况下的默认实现。但是，公平锁更钟爱那些已经等待了最长时间的线程。

  /**
   * returns a {@link condition} instance for use with this
   * {@link lock} instance.
   * <p>the returned {@link condition} instance supports the same
   * usages as do the {@link object} monitor methods ({@link
   * object#wait() wait}, {@link object#notify notify}, and {@link
   * object#notifyall notifyall}) when used with the built-in
   * monitor lock.
   * <ul>
   * <li>if this lock is not held when any of the {@link condition}
   * {@linkplain condition#await() waiting} or {@linkplain
   * condition#signal signalling} methods are called, then an {@link
   * illegalmonitorstateexception} is thrown.
   * <li>when the condition {@linkplain condition#await() waiting}
   * methods are called the lock is released and, before they
   * return, the lock is reacquired and the lock hold count restored
   * to what it was when the method was called.
   * <li>if a thread is {@linkplain thread#interrupt interrupted}
   * while waiting then the wait will terminate, an {@link
   * interruptedexception} will be thrown, and the thread's
   * interrupted status will be cleared.
   * <li> waiting threads are signalled in fifo order.
   * <li>the ordering of lock reacquisition for threads returning
   * from waiting methods is the same as for threads initially
   * acquiring the lock, which is in the default case not specified,
   * but for <em>fair</em> locks favors those threads that have been
   * waiting the longest.
   * </ul>
   * @return the condition object
   **/
  public condition newcondition() {
    return sync.newcondition();
  }

可重入锁还有一些其他的方法，这里就不一一介绍了。this is the end.

总结

以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，谢谢大家对的支持。如果你想了解更多相关内容请查看下面相关链接