简介
Java 并发编程离不开锁， Synchronized 是常用的一种实现加锁的方式，使用比较简单快捷。在 Java 中还有另一种锁，即 Lock 锁。 Lock 是一个接口，提供了超时阻塞、可响应中断以及公平非公平锁等特性，相比于 Synchronized，Lock 功能更强大，可以实现更灵活的加锁方式。
Lock 的主要实现类是 ReentrantLock，而 ReetrantLock 中具体的实现方式是利用另外一个类 AbstractQueuedSynchronizer，所有的操作都是委托给这个类完成。AbstractQueuedSynchronizer 是 Lock 锁的重要组件，本文从 AbstractQueuedSynchronizer 来分析 ReetrantLock 的实现原理。
基本用法
先看一下 Lock 的基本用法：
Lock lock = ...;
lock.lock();
try{
//处理任务
}catch(Exception ex){
}finally{
lock.unlock(); //释放锁
}
lock.lock() 即是加锁， lock.unolck() 是释放锁，为了保证所能够释放，unlock() 应该放到 finally 中。
下面分别从 lock() 和 unlock() 方法来分析加锁和解锁到底做了什么。
lock
下面是 lock() 的代码：
public void lock() {
sync.lock();
}
可以看到，只是简单调用了 sync 对应的 lock() 方法。那么这个 sync 是什么呢？其实这个就是 AbstractQueuedSynchronizer 的实现类。可以看一下 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 有两个方法，主要目的是选择是公平锁还是非公平锁。公平锁指的是先来后到，先争抢锁的线程先获得锁，而非公平锁则不一定。ReentrantLock 默认使用的是非公平锁，也可以通过构造参数选择公平锁。选择哪个锁其实是生成了一个对象并赋值给变量 sync，下面是涉及到的代码：
/**
* Base of synchronization control for this lock. Subclassed
* into fair and nonfair versions below. Uses AQS state to
* represent the number of holds on the lock.
*/
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* Performs {@link Lock#lock}. The main reason for subclassing
* is to allow fast path for nonfair version.
*/
abstract void lock();
/**
* 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;
}
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;
}
protected final boolean isHeldExclusively() {
// While we must in general read state before owner,
// we don't need to do so to check if current thread is owner
return getExclusiveOwnerThread() == Thread.currentThread();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
// Methods relayed from outer class
final Thread getOwner() {
return getState() == 0 ? null : getExclusiveOwnerThread();
}
final int getHoldCount() {
return isHeldExclusively() ? getState() : 0;
}
final boolean isLocked() {
return getState() != 0;
}
/**
* Reconstitutes the instance from a stream (that is, deserializes it).
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
setState(0); // reset to unlocked state
}
}
// 非公平锁
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* 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);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
/**
* Sync object for fair locks
* 公平锁
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
/**
* 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;
}
}
在 ReentrantLock 中有一个抽象的内部类 Sync，继承于 AbstractQueuedSynchronizer 并实现了一些方法。另有两个类 FairSync 和 NoFairSync 继承了 Sync，它们自然就是公平锁以及非公平锁的实现。下面分析将从公平锁出发，非公平锁与公平锁差别并不是很多。
公平锁 FairSync 加锁的代码如下：
final void lock() {
acquire(1);
}
只有一行，调用了 acquire，这是 AbstractQueuedSynchronizer 中的一个方法，代码如下：
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
acquire 的实现也很短，不多在其中却包含了加锁的具体实现，关键就在内部调用的几个方法中。
为了理解加锁和解锁的过程，下面具体介绍一下 AbstractQueuedSynchronizer（以下简称 AQS）。
AbstractQueuedSynchronizer
AQS 中使用一个同步队列来实现线程同步状态的管理，当一个线程获取锁失败的时候， AQS将此线程构造成一个节点(Node)并加入同步队列并且阻塞线程。当锁释放时，会从同步队列中将第一个节点唤醒并使其再次获取锁。
同步队列中的节点用来保存获取锁失败的线程的相关信息，包含如下属性：
static final class Node {
/** Marker to indicate a node is waiting in shared mode */
// 标识共享模式
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
// 标识独占模式
static final Node EXCLUSIVE = null;
/** waitStatus value to indicate thread has cancelled. */
// 线程取消
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking. */
// 需要唤醒后继节点
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition. */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate.
*/
static final int PROPAGATE = -3;
// 节点状态，为上面的几个状态之一
volatile int waitStatus;
// 前置节点
volatile Node prev;
// 后继节点
volatile Node next;
// 节点所表示的线程
volatile Thread thread;
Node nextWaiter;
...
}
Node 是 AQS 的内部类，其中包含一些属性标识一个阻塞线程的节点，包括是独占模式还是共享模式、节点的状态、前驱结点、后继结点以及节点所代表的线程。
同步队列是一个双向列表，在 AQS 中有这样几个属性：
/**
* Head of the wait queue, lazily initialized. Except for
* initialization, it is modified only via method setHead. Note:
* If head exists, its waitStatus is guaranteed not to be
* CANCELLED.
*/
// 头结点
private transient volatile Node head;
/**
* Tail of the wait queue, lazily initialized. Modified only via
* method enq to add new wait node.
*/
// 尾节点
private transient volatile Node tail;
/**
* The synchronization state.
*/
// 锁的状态
private volatile int state;
其中，head 和 tail 分别指向同步队列的头结点和尾节点，state 标识锁当前的状态，为 0 时表示当前锁未被占用，大于 1 表示被占用，之所以是大于 1 是因为锁可以重入，每重入一次增加 1。同步队列的结构大致如下图：
lock_queue
了解了同步队列后，下面具体看看加锁和解锁的过程。
加锁
final void lock() {
acquire(1);
}
public final void acquire(int arg) {
// 加锁的主要代码
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
主要逻辑其实就是一行代码：if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))，tryAcquire 是尝试获取一下锁，为什么说是尝试呢？看代码：
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;
}
tryAcquire 可以分为三条分支：
-
当前锁未被占用（
getState() == 0），则判断是否有前驱结点，没有的话就用 CAS 加锁(compareAndSetState(0, acquires))，加锁成功则调用setExclusiveOwnerThread(current)标示一下并返回true。 -
当前线程已经获取过这个锁，则此时是重入，改变
state的计数即可，返回true表示加锁成功。 -
如果不是上面两种情况，那么说明锁被占用或者 CAS 没有抢过其它线程，则需要进入同步队列，返回
false表示尝试加锁失败。
回到 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 这一行，如果 tryAcquire(arg) 返回 false 将会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。先看一下 addWaiter(Node.EXCLUSIVE) ，这个方法的代码如下所示：
private Node addWaiter(Node mode) {
// 将线程包装成 Node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
// 尾节点为空
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
其中主要逻辑是将当前线程包装为一个 Node 节点并加入同步队列。如果尾节点为空，则用 CAS 设置尾节点，如果入队失败则调用 enq(node)，这个方法内部是一个循环，利用自旋 CAS 把节点加入同步队列，具体代码就不分析了。
在节点加入队列之后，执行的是 acquireQueued 方法，代码如下：
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
// 这是一个无限循环
for (;;) {
final Node p = node.predecessor();
// 如果前驱节点是 head，则尝试获取锁
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);
}
}
acquireQueued 实现了线程的睡眠与唤醒。在内部是一个无限循环，每次获取前驱节点，如果前驱结点是 HEAD，那么尝试去获取锁，获取成功则将此节点变为新的头结点并将原先的头结点出队。如果前驱节点不是头结点或者获取锁失败，那么就会进入 shouldParkAfterFailedAcquire 方法，判断是否进入睡眠，如果这个方法返回 true，则调用 parkAndCheckInterrupt 让线程进入睡眠状态。下面是 parkAndCheckInterrupt 的代码：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 线程在这一步进入阻塞状态
return Thread.interrupted();
}
对于 shouldParkAfterFailedAcquire 来说，如果前驱节点正常，那么会返回 true，表示当前线程应该挂起，如果前驱结点取消了排队，那么当前线程有机会抢锁，此时返回 false，并继续 acquireQueued 中的循环。
解锁
相比于加锁，解锁稍微简单一点，看一下 unlock 的代码：
public void unlock() {
sync.release(1);
}
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 解锁，如果解锁成功则唤醒后继结点，返回值表示是否成功释放锁。那为什么会解锁不成功，其实是因为重入，看一下 tryRelease 的代码：
protected final boolean tryRelease(int releases) {
// 更新 state 计数值
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;
}
将 state 减去对应的值，如果 state == 0，那么说明锁已经完全释放。
在 release 中，如果锁已经完全释放，那么将调用 unparkSuccessor 唤醒后继节点，唤醒的节点所代表的线程阻塞在 parkAndCheckInterrupt 中：
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 线程在这一步进入阻塞状态
return Thread.interrupted();
}
线程被唤醒后，将继续 acquireQueued 中的循环，尝试获取锁。
总结
本文简要分析了 Lock 锁的原理，主要是利用 AbstractQueuedSynchronizer这个关键的类。AQS 的核心在于使用 CAS 更新锁的状态，并利用一个同步队列将获取锁失败的线程进行排队，当前驱节点解锁后再唤醒后继节点，是一个几乎纯 Java 实现的加锁与解锁。
网友评论