ReentrantLock源码解析

什么是ReentrantLock

javadoc中的描述如下：

一种可重入互斥锁，具有与使用synchronized方法和语句访问隐式监视器锁相同的基本行为和语义，同时具有额外的功能。

最后成功加锁并且没有释放该锁的线程拥有一个ReentrantLock。当锁不属于其他线程时，调用锁的线程将返回并成功获取锁。如果当前线程已经拥有锁，该方法将立即返回。可以使用isHeldByCurrentThread和getHoldCount方法进行检查。

该类的构造函数接受一个可选的公平性参数。当设置为true时，在争用下，锁倾向于向等待时间最长的线程授予访问权。否则，锁并不保证任何特定的访问顺序。使用许多线程访问的公平锁的程序可能会显示较低的总体吞吐量（即更慢），比起使用默认设置的程序，但在获得锁和保证不会饿死方面，时间上的差异更小时间。但是请注意，锁的公平性并不保证线程调度的公平性。因此，使用公平锁的多个线程中的一个可能会在其他活动线程没有进展或当前没有持有锁的情况下连续多次获得该锁。还要注意，非定时tryLock()方法不支持公平设置。如果锁可用，即使其他线程正在等待，它也会成功。

建议在调用获取锁之后，总是立即使用try语句块，最典型的是在构造之前/之后：

 1  class X {
 2    private final ReentrantLock lock = new ReentrantLock();
 3    // ...
 4 
 5    public void m() {
 6      lock.lock();  // block until condition holds
 7      try {
 8        // ... method body
 9      } finally {
10        lock.unlock()
11      }
12    }
13  }

除了实现锁接口之外，该类还定义了一些用于检查锁状态的公共和受保护的方法。其中一些方法仅用于检测和监视。

该类的序列化与内置锁的行为方式相同:反序列化锁处于解锁状态，而与序列化时的状态无关。

此锁最多支持同一线程的2147483647个递归锁。试图超过此限制将导致锁定方法抛出错误。

公平锁和非公平锁

ReentrantLock类中定义了抽象静态类java.util.concurrent.locks.ReentrantLock.Sync，继承自java.util.concurrent.locks.AbstractQueuedSynchronizer。

ReentrantLock实例属性中有Sync类型的sync。

 1     /** Synchronizer providing all implementation mechanics */
 2     private final Sync sync;
 3 
 4     /**
 5      * Base of synchronization control for this lock. Subclassed
 6      * into fair and nonfair versions below. Uses AQS state to
 7      * represent the number of holds on the lock.
 8      */
 9     abstract static class Sync extends AbstractQueuedSynchronizer {
10     ...
11     }

ReentrantLock有两个构造函数，默认使用非公平策略。

 1     /**
 2      * Creates an instance of {@code ReentrantLock}.
 3      * This is equivalent to using {@code ReentrantLock(false)}.
 4      */
 5     public ReentrantLock() {
 6         sync = new NonfairSync();
 7     }
 8 
 9     /**
10      * Creates an instance of {@code ReentrantLock} with the
11      * given fairness policy.
12      *
13      * @param fair {@code true} if this lock should use a fair ordering policy
14      */
15     public ReentrantLock(boolean fair) {
16         sync = fair ? new FairSync() : new NonfairSync();
17     }

获取锁的实现

ReentrantLock.lock()方法其实是依赖于其内部sync对象的实现，下面以非公平策略（即内部sync对象是java.util.concurrent.locks.ReentrantLock.NonfairSync类型）为例

1     public void lock() {
2         sync.lock();
3     }

NonfairSync类的lock()实现中，先调用AbstractQueuedSynchronizer.compareAndSetState(int expect, int update)，设置AbstractQueuedSynchronizer的state为1。如果设置成功，调用AbstractOwnableSynchronizer.setExclusiveOwnerThread(Thread thread)设置exclusiveOwnerThread属性为当前线程。如果不成功，调用AbstractQueuedSynchronizer#acquire(int arg)尝试获取锁。

 1         /**
 2          * Performs lock.  Try immediate barge, backing up to normal
 3          * acquire on failure.
 4          */
 5         final void lock() {
 6             if (compareAndSetState(0, 1))
 7                 setExclusiveOwnerThread(Thread.currentThread());
 8             else
 9                 acquire(1);
10         }

在AbstractQueuedSynchronizer#acquire(int arg)方法中，if的第一个判断条件调用ReentrantLock.NonfairSync#tryAcquire(int acquires)方法尝试获取锁，该方法直接调用了抽象静态类ReentrantLock.Sync#nonfairTryAcquire(int acquires)方法。如果返回false，则添加进FIFO等待队列中。

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

抽象静态类ReentrantLock.Sync#nonfairTryAcquire(int acquires)方法中，如果此时state为0，说明锁已经被其他线程释放，则再次尝试获取锁。如果当前线程已经持有了锁，则会把state+1，当state<0时，则抛出错误，因此最多能重复加锁Int.MAX_VALUE次。如果仍旧获取不到锁，则返回false。

 1        final boolean nonfairTryAcquire(int acquires) {
 2             final Thread current = Thread.currentThread();
 3             int c = getState();
 4             if (c == 0) {
 5                 if (compareAndSetState(0, acquires)) {
 6                     setExclusiveOwnerThread(current);
 7                     return true;
 8                 }
 9             }
10             else if (current == getExclusiveOwnerThread()) {
11                 int nextc = c + acquires;
12                 if (nextc < 0) // overflow
13                     throw new Error("Maximum lock count exceeded");
14                 setState(nextc);
15                 return true;
16             }
17             return false;
18         }

如果没有获取到锁，则线程应该被阻塞，首先调用AbstractQueuedSynchronizer#addWaiter(Node mode) 将当前线程创建为Node并添加到队列中。方法内部先判断是否已经存在尾部节点。如果已经存在，将这个node添加到原尾部节点的尾部。如果不存在，调用AbstractQueuedSynchronizer#enq(final Node node)创建队列。

 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     }

在AbstractQueuedSynchronizer#enq(final Node node)方法中，通过自旋和CAS操作保证能写入队列的尾部。

 1     private Node enq(final Node node) {
 2         for (;;) {
 3             Node t = tail;
 4             if (t == null) { // Must initialize
 5                 if (compareAndSetHead(new Node()))
 6                     tail = head;
 7             } else {
 8                 node.prev = t;
 9                 if (compareAndSetTail(t, node)) {
10                     t.next = node;
11                     return t;
12                 }
13             }
14         }
15     }

与该线程相关联的Node对象创建之后，调用AbstractQueuedSynchronizer#acquireQueued(final Node node, int arg)方法。如果当前node是头部节点，则再次尝试获取锁。

 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     }

如果仍然无法获取到，则调用AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire(Node pred, Node node)检测及更新node的status为Node.SIGNAL状态。

 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     }

如果成功，调用java.util.concurrent.locks.AbstractQueuedSynchronizer#parkAndCheckInterrupt()方法将调用LockSupport#park(java.lang.Object)方法，将该线程设置为等待状态

1     private final boolean parkAndCheckInterrupt() {
2         LockSupport.park(this);
3         return Thread.interrupted();
4     }

释放锁的实现

ReentrantLock#unlock实际上是调用了AbstractQueuedSynchronizer#release(int arg) 方法

1     public void unlock() {
2         sync.release(1);
3     }

在AbstractQueuedSynchronizer#release(int releases)方法中，先调用ReentrantLock.Sync#tryRelease(int releases)方法尝试释放锁

1     public final boolean release(int arg) {
2         if (tryRelease(arg)) {
3             Node h = head;
4             if (h != null && h.waitStatus != 0)
5                 unparkSuccessor(h);
6             return true;
7         }
8         return false;
9     }

在ReentrantLock.Sync#tryRelease(int releases)方法尝试释放锁时，只有当c == 0时才将锁释放

 1         protected final boolean tryRelease(int releases) {
 2             int c = getState() - releases;
 3             if (Thread.currentThread() != getExclusiveOwnerThread())
 4                 throw new IllegalMonitorStateException();
 5             boolean free = false;
 6             if (c == 0) {
 7                 free = true;
 8                 setExclusiveOwnerThread(null);
 9             }
10             setState(c);
11             return free;
12         }

锁释放后，调用AbstractQueuedSynchronizer#unparkSuccessor(Node node)方法将队列的头部节点唤醒

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

由于线程是在调用lock()方法时，在AbstractQueuedSynchronizer#acquireQueued(final Node node, int arg)方法的第14行被挂起，唤醒后会继续执行后续的代码，再次进入for循环，获取到锁后，将本身节点设置为head节点，原head节点的next设置为null，以便进行回收。