ReentrantLock几乎拥有了synchronized的一切基本功能，而且它还做了相应的扩展。比如，它继承了Lock接口，对资源的保护更加精细化。它是AQS的派生类，同时支持公平锁FairSync和非公平锁NonFairSync两种锁模式。它总是以Lock()和unlock()的形式，成对使用。
private void ReentrantLockDemo() {
ReentrantLock lock = new ReentrantLock();
try {
lock.lock();
Thread.sleep(100);
} catch (InterruptedException e) {
Log.d(TAG,"ReentrantLock#InterruptedException = " + e.getMessage());
} finally {
lock.unlock();
}
}
上面的demo就是我们经常用的一种ReentrantLock方式，我们从构造方法开始分析。
/**
* Creates an instance of {@code ReentrantLock}.
* This is equivalent to using {@code ReentrantLock(false)}.
*/
public ReentrantLock() {
sync = new NonfairSync();
}
默认的构造方法，生成了一个非公平锁，其实还有一个构造方法可以生成公平锁。NonfairSync的继承关系如下：
NonfairSync继承关系
NonfairSync是ReentrantLock的内部类，继承了同样是内部类的Sync。这也就不难理解，返回的实例是Sync了。而Sync又继承了AbstractQueuedSynchronizer类，也就是我们前面提到的AQS。
请求锁
demo中调用了lock.lock()方法，而NonfairSync中正好有对该方法的实现：
/**
* 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);
}
方法中直接执行锁操作，这里的同步策略采用的CAS方式，如果设置成功，代表得到锁，将当前线程设为独占式线程。如果请求失败，执行acquire请求方法。
CAS是一种直接操作JVM内存的方式，其工作原理：给出一个期望值expect与内存中的值value进行比较，如果两者相等。说明内存没有被修改过，当前线程就可以修改这块内存，写入自己携带的另一个值expect。这样内存区被修改后，别的持有expect的线程就无法修改这块内存，直到当前线程释放掉这块内存，并将其改为持有之前的值为止。
但是需要注意的是，这种方法虽然避免了加锁造成的时间浪费，但是它容易运行失败引起性能问题，而且还要通过AutomicStampedReference处理ABA问题
知道了CAS的原理，我们就可以接着分析。compareAndSetState方法实际上是AbstractQueuedSyncronizer提供的一个方法，其源码如下：
/**
* 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) {
return U.compareAndSwapInt(this, STATE, expect, update);
}
从注释中我们就可以知道这是CAS的一种实现方式，内存读写操作是基于volatile修饰符的操作，换句话说，能够保证对内存的可见性和有序性操作。
它真正的操作是Unsafe类的compareAndSwapInt方法。该方法才是可以直接操作JVM内存的后门。this代表当前对象，STATE代表的是当前对象的state字段在内存中存储的位置相对于，对象起始位置的偏移量。起始位置+偏移量，就是state在内存中的保存地址。而state就是我们刚刚在解释CAS原理时提到的value这个值。如果你对Unsafe和CAS特别感兴趣，可以看看这两篇文章：《Java Unsafe 类》和《Java中Unsafe使用详解》
我们继续说刚刚的state字段，它同样是在AQS中声明的一个同步变量，
/**
* The synchronization state.
*/
private volatile int state;
看到了吧，它不但被volatile修饰，而且是private的字段，也许这并没有什么稀奇！但是：
/**
* Returns the current value of synchronization state.
* This operation has memory semantics of a {@code volatile} read.
* @return current state value
*/
protected final int getState() {
return state;
}
/**
* Sets the value of synchronization state.
* This operation has memory semantics of a {@code volatile} write.
* @param newState the new state value
*/
protected final void setState(int newState) {
state = newState;
}
但是他的set和get方法全部加了final和protected修饰，换句话用户想使用它只能在其派生类里，想要覆盖它门都没有。当然这也避免了用户在使用锁时调用修改该状态，从一定程度上降低了同步操作的复杂性。
我们接着聊NonfairSync.lock()方法，如果请求锁成功了，compareAndSetState(0, 1)将state内存块的值设为1，别的线程就访问不了，那当前线程就对这个内存区域在一定时间内独占，直接调用AbstractOwnableSynchronizer【注意，不是AQS。它是AQS的父类】的setExclusiveOwnerThread方法将当前线程设置为独占线程。
如果请求锁失败，compareAndSetState(0, 1)方法返回false，那就通过调用AQS#acquire方法继续发起请求。
/**
* 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) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
该方法以独占方式发起请求锁，忽略中断。发起请求是通过调用至少一次tryAcquire方法完成的。如果tryAcquire也请求失败了，系统就会把当前线程加入到等待队列里，这中间可能会重复的阻塞和解阻塞，直到调用tryAcquire请求成功。
step1：调用tryAcquire。
该方法在ReentrantLock#NonfairSync具体实现，tryAcquire将请求任务交给了nonfairTryAcquire完成请求。
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
来看看nonfairTryAcquire方法的源码：
/**
* 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) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
代码中通过getState获取当前锁的状态，如果是0说明锁空闲将状态设为acquires这个请求数，并将当前线程设为独占线程，请求锁成功。否则的话，就要对比当前线程current和持有锁的独占线程是否为同一线程，如果是，直接调用setState()设置锁状态。如果以上两个条件都无法满足，不好意思请求锁失败。
注意：请求锁的条件2满足，请求成功，state = c + acquires。它是大于1的，而且条件2没有调用CAS方法改变state，速度自然快很多。这就是锁的可重入性！也就是说同一个Thread联系调用了N次lock，一旦获取锁成功，后续请求锁state依次叠加不用排队竞争。同样，每一个lock都要对应一个unlock的调用。
那么问题来了，可重入锁可以被一个线程独占state次，那些没有获取到锁的线程怎么办？
step2：排队等待。
对于暂时无法获取到锁的线程，调用tryAcquire必然返回false，代码就要执行acquireQueued方法，该方法是通过addWaiter(Node.EXCLUSIVE)方法传入了一个独占式final节点。我们先来看看Node节点：
- 该节点构成的等待队列是一个“CLH(Craig, Landin, and Hagersten)”锁队列的变体，CLH锁队列通常用于自旋锁。Node仅使用其基本功能，并作了相应的改造。
- Node有两个字段prev和next分别指向节点的前任节点和后续节点，足见这是一个双向链表。
- Node在首次构造时，会创建一个哑结点作为头结点。
- Node构成的等待队列既有头节点，又有尾节点。在入队时，仅仅把新节点拼接作为尾节点。出队时，仅仅设置头节点。
- Node提供了一个waitStatus代表当前节点的等待状态：
CANCELLED==1，因为超时或中断等，线程取消了锁请求；
SIGNAL==-1，表示继任节点被阻塞，当前节点在释放锁或取消时需要通知它的继任节点解除等待，发起锁请求；
CONDITION==-2，节点当前处于一个竞争队列里，不能用于同步队列，直到状态发生变化；
PROPAGATE==-3，用于共享模式，通知其他节点；
0，以上状态都不是。 - Node支持SHARED和EXCLUSIVE两种模式的节点，通过nextWaiter字段存储模式值。
- thread：存贮当前线程
- 所有字段均被volatile修饰，保证内存操作的可见性和有序性
知道了节点的功能，我们就可以继续分析addWaiter(Node.EXCLUSIVE)方法了，源码如下：
/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new node
*/
private Node addWaiter(Node mode) {
Node node = new Node(mode); // 代码1
for (;;) { // 代码2
Node oldTail = tail;
if (oldTail != null) { // 代码3
U.putObject(node, Node.PREV, oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else { // 代码4
initializeSyncQueue();
}
}
}
该方法毫不含糊，直接表明通过当前线程和指定的模式创建和入队节点。
代码1处，通过Node构造方法创建了一个目标节点node，构造方法如下：
/** Constructor used by addWaiter. */
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
U.putObject(this, THREAD, Thread.currentThread());
}
nextWaiter就是我们传入的请求模式EXCLUSIVE，当前线程被保存到Node的thread字段中。
由于等待队列是一个双向链表，代码2处通过一个无限for循环轮询链表，而进入等待队列的当前节点是第一次使用队列，代码3处oldTail自然是null，代码4处的initializeSyncQueue()方法被执行。看看整个方法的源码：
/**
* Initializes head and tail fields on first contention.
*/
private final void initializeSyncQueue() {
Node h;
if (U.compareAndSwapObject(this, HEAD, null, (h = new Node())))
tail = h;
}
好简单有木有，直接创建一个哑结点（设为h）作为当前队列的头结点，由于目前只有一个节点所有head、tail这两个头尾指针都指向它。对了，你是不是一直疑惑，volatile不支持原子性操作，为啥能赋值对象？答案是CAS操作是原子性的。
继续回到addWaiter方法的代码2，循环继续。此时oldTail = tail不再为null，代码3处的条件为真， U.putObject(node, Node.PREV, oldTail);这一行代码的意思是：把我们刚刚创建的持有当前线程的目标节点node的prev字段用oldTail赋值。由于oldTail指向哑结点h，那么node.prev = 哑结点。接着调用compareAndSetTail(oldTail, node)方法将目标节点node设为尾节点。如果设置成功，oldTail也就是哑结点或者当前的头结点的下一个节点next就是我们刚刚创建的目标节点node，这一步是通过oldTail.next = node完成的。
这样，当前线程所在节点node插入等待队列测操作就完成了。
如果这个时候，又来了新的线程thread2、thread3……他们都会被依次插入队列中等待。
节点创建好了，什么时候调用呢？还记得我们的acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法吗？它就是用来发起二次锁请求的。源码如下：
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) { // 代码1
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) { // 代码2
setHead(node);
p.next = null; // help GC
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) // 代码3
interrupted = true;
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
代码1处，通过一个for循环不间断的发起请求。代码2有两个条件，虽然条件1目前满足p == head。但是条件2tryAcquire在锁被占用的情况下是无法请求成功的。
代码走入代码3的逻辑，同样有两个条件：shouldParkAfterFailedAcquire(p, node) 和parkAndCheckInterrupt()。
- **shouldParkAfterFailedAcquire(p, node) **
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false;
}
该方法是用来核对和修改请求锁失败的线程的等待状态waitStatus的，由于外层的for循环和acquire获取锁失败，所以pred节点也就是当前节点的前一个节点的状态必然会被改为SIGNAL，并且返回true。而Node节点被设置为SIGNAL就是为了告诉持有锁的线程，释放锁时请唤醒队列中拥有SIGNAL状态的节点的继任节点线程。整个继任在挂起状态。
条件2 parkAndCheckInterrupt
既然继任线程设置了闹钟，他就高枕无忧了，系统是通过parkAndCheckInterrupt方法来完成挂起线程的。源码如下：
/**
* Convenience method to park and then check if interrupted.
*
* @return {@code true} if interrupted
*/
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
当前线程先通过LockSupport.park(this)方法挂起，通常只有LockSupport.unpark方法或中断才能把它唤醒，然后执行Thread.interrupted()中断操作。
park方法的源码如下：
/**
* Disables the current thread for thread scheduling purposes unless the
* permit is available.
*
* <p>If the permit is available then it is consumed and the call returns
* immediately; otherwise
* the current thread becomes disabled for thread scheduling
* purposes and lies dormant until one of three things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @since 1.6
*/
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
U.park(false, 0L);
setBlocker(t, null);
}
注意它仍然是通过Unsafe方法直接操作JVM内存将线程刮起，0L代表无限挂起。
Thread.interrupted()不但可以判断当前线程是否中断，而且可以清除中断状态。如果连续调用偶数次该方法，相当于没有调用。另外，如果线程已经处于中断状态，该方法返回true，否则返回false。所以，parkAndCheckInterrupt方法第一次调用会返回false，使得acquireQueued继续循环调用tryAcquire(arg)方法获取锁。
释放锁
整个请求锁的逻辑，到此就算告一段落了。那么，要想请求成功，持有当前锁的线程必须执行ReentrantLock#unlock方法，释放锁。源码如下：
/**
* 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);
}
看看人家怎么说的，如果当前线程是锁的持有者，持有的数量减1，接着看AQS#release的源码：
/**
* 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;
}
以独占模式释放锁，如果tryRelease方法返回true，等待队列中的头结点的继任节点会被唤醒。ReentrantLock#tryRelease的源码如下：
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;
}
没什么新鲜的，就是通过getState和setState设置锁的状态，当state为0时，说明锁被彻底释放，等待队列中线程可以访问锁。
一旦锁释放成功。release方法就会执行，unparkSuccessor(h);代码唤醒继任节点中的线程。源码如下：
/**
* 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)
node.compareAndSetWaitStatus(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 p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
代码中写的很清晰，如果头节点的等待状态waitStatus是负值，比如SIGNAL，就把置为0还原。通常情况下，头结点的下一个节点就是继任节点successer，但是如果该节点取消等待了或者值为null，就通过反向从tail尾节点查找的办法，找到待唤醒的节点s。其中，唤醒任务是LockSupport.unpark(s.thread);来完成的。
这样，被唤醒的线程就可以通过acquire方法获取到锁了。
公平锁
我们上面分析的加锁和释放锁都是基于NonfairSync这把非公平锁的，那么公平锁FairSync到底公平在何处？其实主要是tryAcquire方法的实现不一样：
NonfairSync#nonfairTryAcquire
/**
* 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) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
FairSync#tryAcquire源码
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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;
}
}
注意：在请求锁的过程中，公平锁多了AQS的hasQueuedPredecessors()方法的判断，该方法源码如下：
/**
* Queries whether any threads have been waiting to acquire longer
* than the current thread.
*
* <p>An invocation of this method is equivalent to (but may be
* more efficient than):
* <pre> {@code
* getFirstQueuedThread() != Thread.currentThread()
* && hasQueuedThreads()}</pre>
*
* <p>Note that because cancellations due to interrupts and
* timeouts may occur at any time, a {@code true} return does not
* guarantee that some other thread will acquire before the current
* thread. Likewise, it is possible for another thread to win a
* race to enqueue after this method has returned {@code false},
* due to the queue being empty.
*
* <p>This method is designed to be used by a fair synchronizer to
* avoid <a href="AbstractQueuedSynchronizer.html#barging">barging</a>.
* Such a synchronizer's {@link #tryAcquire} method should return
* {@code false}, and its {@link #tryAcquireShared} method should
* return a negative value, if this method returns {@code true}
* (unless this is a reentrant acquire). For example, the {@code
* tryAcquire} method for a fair, reentrant, exclusive mode
* synchronizer might look like this:
*
* <pre> {@code
* protected boolean tryAcquire(int arg) {
* if (isHeldExclusively()) {
* // A reentrant acquire; increment hold count
* return true;
* } else if (hasQueuedPredecessors()) {
* return false;
* } else {
* // try to acquire normally
* }
* }}</pre>
*
* @return {@code true} if there is a queued thread preceding the
* current thread, and {@code false} if the current thread
* is at the head of the queue or the queue is empty
* @since 1.7
*/
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
该方法检查队列中是否存在等待获取锁的线程，比当前线程的等待时间还长。注意(s = h.next) == null这句代码，它表示队列中刚入队了一个新的等待节点node，只是还未赋值给head.next。
通过该方法可以防止如下非公平现象的发生：如果一个新的线程发起请求的时刻，正好当前持有锁的线程释放了锁，那对于正在唤醒的等待了很长时间的线程是不公平的。
但是，使用公平锁比较耗性能，两害相侵取其轻，官方把非公平锁作为ReentrantLock的默认锁，说明这种非公平情况在大部分情况下是可以接受的。
AQS作为一种工具，除了能够实现ReentrantLock之外，还实现了CyclicBarrier、CountdownLatch、ReentrantReadWriteLock、ThreadPoolExecutor.Worker、Semaphore等。此外，它还提供了可以代替wait/notify的Condition接口封装，后期我们逐一分析。
网友评论