ReentrantLock源码分析
一、引言
ReentrantLock作为concurrent包一下的一员，有着比Synchronized更加直观灵活的使用方式；
private static void lockTest() {
try {
lock.lock();
//dosometing
} catch (Exception e) {
} finally {
//unlock一定要放在finally语句中
lock.unlock();
}
}
ReentrantLock是通过CAS+AQS队列实现（后面会分析），具有以下特性：
1、可重入锁
2、可响应中断
3、可尝试加锁（tryLock）
4、限时等待尝试加锁（tryLock(long timeout, TimeUnit unit)）
5、公平锁、非公平锁（通过构造函数的boolean指定）
ReentrantLock与Synchronized的异同：
不同：
1、synchronized是Java的关键字，使用较为方便；ReentrantLock是类，使用起来需要创建调用方法等
2、ReentrantLock使用较为灵活（可指定是否为公平锁、可尝试加锁等），Synchronized默认只能是非公平锁，没有其它丰富的功能
3、Synchronized在jdk1.6优化之前的性能是远远不如ReentrantLock的，但是优化后（可根据使用场景变为无锁、偏向锁、轻量级锁、重量级锁）；但Synchronized锁的粒度灵活度还是不如ReentrantLock
相同：
1、ReentrantLock与synchronized都是阻塞式锁
2、两者都是可重入锁
常用方法：
1、lock 加锁
2、unLock 解锁
3、tryLock 尝试加锁，成功返回true，失败返回false，不会等待
4、 tryLock(long timeout,TimeUnit unit) 与tryLock大致相同，尝试获取锁，如果超过这段时间依然没有获取锁，返回false
5、 lockInterruptibly 获取锁，跟lock不一样的地方是，过程中会检测是否中断（interrupt），若是会抛出异常
6、 构造函数 public ReentrantLock(boolean fair) 可设置公平锁还是非公平锁
二、源码分析
1、构造函数
ReentrantLock lock = new ReentrantLock();//里面的参数可指定是公平还是非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
//指定是当前是公平锁 还是非公平锁
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
sync继承自AbstractQueuedSynchronizer（即传说中的AQS）
sync.png
2、AQS数据结构分析
AQS（java.util.concurrent.locks.AbstractQueuedSynchronizer）是ReentrantLock实现线程排队等待的数据结构，本质是一个双向的FIFO队列；结构如下：
AQS结构.png
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;//独占模式，表示只有一个线程能执行例如ReentrantLock；共享模式 多个线程可同时执行 例如Semaphore/CountDownLatch
/** 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;//说明这个结点因为被某一个condition挂起了
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate.
*/
static final int PROPAGATE = -3;//在共享模式下，下一次获取锁后可以无限传播
还有以下的变量：
waitStatus 当前结点的状态，默认是0，可以是CANCELL、SIGNAL 等
prev 前继结点
next 后继结点
thread 对应的线程
nextWaiter 下一个等待condition的结点
state 状态；是一个int值，表示当前线程占用资源的数量；0表示空闲，没有线程占用；ReentrantLock的state表示线程重入锁的次数
3、lock
先来看非公平锁的情况下逻辑
3.1、非公平锁的lock
public void lock() {
//当为非公平锁时 这里的lock会走NonfairSync下的lock函数
sync.lock();
}
final void lock() {
//利用CAS更新当前锁的状态 分析1
if (compareAndSetState(0, 1))
//设置当前线程为独占锁的拥有者 分析2
setExclusiveOwnerThread(Thread.currentThread());
else
//如果CAS更新失败 获取锁 分析3
acquire(1);
}
===>分析1
//AbstractQueuedSynchronizer下方法；通过CAS更新当前锁的状态
protected final boolean compareAndSetState(int expect, int update) {
return U.compareAndSwapInt(this, STATE, expect, update);
}
===>分析2
//如果CAS更新成功，设置当前线程为独占锁的拥有者
//AbstractOwnableSynchronizer中方法；
//AbstractOwnableSynchronizer把AQS包了一层；主要提供了一些获取、设置当前独占锁拥有者
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
===>分析3
//如果CAS更新失败 尝试获取锁
//AQS中的方法
public final void acquire(int arg) {
//如果获取锁失败 阻塞排队
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
//如果排队也失败了 中断自己
selfInterrupt();
}
//tryAcquire是AQS中的方法，由子类具体实现；这里看NonfairSync的tryAcquire
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
//acquires 是1
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();//获取state值
if (c == 0) {
//如果当前没有线程占用锁 通过CAS更新state标志位
//(这里也是非公平锁的一个原因 不判断前面是否有其它节点 直接CAS尝试拿锁)
if (compareAndSetState(0, acquires)) {
//将当前线程设置为独占锁的拥有者
setExclusiveOwnerThread(current);
return true;
}
}
//如果state不为0 并且当前线程就是独占锁的拥有者（表示重入）
else if (current == getExclusiveOwnerThread()) {
//累加state
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
//新节点入队方法
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (;;) {//通过阻塞的方式将当前的node入队
Node oldTail = tail;//获取尾结点
if (oldTail != null) {
//将新节点插入到队列尾部
U.putObject(node, Node.PREV, oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else {
//队列为空 初始化队列
initializeSyncQueue();
}
}
}
//将当前线程入队阻塞
final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) {
//获取node的前节点
final Node p = node.predecessor();
//如果p的前节点就是头结点 尝试获取锁
if (p == head && tryAcquire(arg)) {
//设置当前node为头结点
setHead(node);
p.next = null; // help GC
return interrupted;
}
//做一些阻塞前的准备操作
if (shouldParkAfterFailedAcquire(p, node) &&
//阻塞 唤醒后检查中断标志位
parkAndCheckInterrupt())
interrupted = true;
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
3.2、公平锁的lock
final void lock() {
acquire(1);
}
//acquire的实现与非公平锁相同
//看一下公平锁的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;
}
//判断头结点后的节点的线程是否为当前线程
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());
}
3.3、lock的流程总结
调用lock方法，当为非公平锁时，只要有线程过来就尝试获取锁，如果获取成功（AQS的state==0 并且CAS写入成功）将自身设置为独占锁的拥有者 或者state != 0但当前独占锁就是自身（表示重入），获取锁成功并将state累加；如果获取失败就将自己设置到AQS队列的尾部，等待唤醒；当为公平锁时，与非公平锁的主要区别在于公平锁在执行tryAcquire时，需要加一个判断（当前节点是否还有其他节点），如果有其它节点则将自身添加到队列中等待唤醒；具体lock的流程图如下：
ReentrantLock lock流程.png
4、tryLock
直接尝试获取，获取不到返回false，不会阻塞线程
public boolean tryLock() {
//直接尝试获取
return sync.nonfairTryAcquire(1);
}
5、tryLock(long timeout, TimeUnit unit)
直接尝试获取锁，如果失败则在阻塞时间内不断尝试获取锁
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
//判断是否中断了
throw new InterruptedException();
return tryAcquire(arg) || //尝试获取锁；否则进入doAcquireNanos
doAcquireNanos(arg, nanosTimeout);
}
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
//如果时间小于等于0 直接返回失败
return false;
final long deadline = System.nanoTime() + nanosTimeout;//算出超时的时刻
final Node node = addWaiter(Node.EXCLUSIVE);//将当前线程节点加入FIFO队列
try {
for (;;) {
final Node p = node.predecessor();//获取当前节点的前继节点
if (p == head && tryAcquire(arg)) {
//如果前继结点是头结点 尝试获取锁 如果成功的话将自身置为头结点
setHead(node);
p.next = null; // help GC
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) {
//超出指定
cancelAcquire(node);
return false;
}
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
//时间差过大要超过自旋的限定值了 直接挂起线程 nanosTimeOut时间后再唤醒
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
6、unLock
解锁并唤醒队列中下一个不为null的节点
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;//当前执行的节点，当前的head结点指向的是在执行的线程，
//如果后面还有其他结点需要唤醒，此时的head的status应该会是SINGAL，会继续走到unparkSuccessor()
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;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);
Node s = node.next;
//s.waitStatus > 0说明节点为Cancel节点
if (s == null || s.waitStatus > 0) {
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
//从后往前遍历节点 找出不为null的节点
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
//唤醒下一个线程
LockSupport.unpark(s.thread);
}
7、ReentrantLock的唤醒通知机制
同Object的wait notify类似，ReentrantLock也有一套自己的唤醒等待机制；通过ReentrantLock的newCondition创建condition对象，调用condition的await实现等待、调用signal或者signalAll实现唤醒，例如下面的一个生产者消费者例子：
public class ProducterAndConsumerTest {
private static Queue<Integer> queue = new ArrayDeque<>();
private static ReentrantLock lock = new ReentrantLock();
private static Condition condition = lock.newCondition();
public static void main(String[] args) {
new Thread(new ProducterRunnable()).start();
new Thread(new ConsumerRunnable()).start();
}
private static class ProducterRunnable implements Runnable {
@Override
public void run() {
try {
lock.lock();
while (true) {
if (queue.size() >= 10) {
//当生产者队列已满 等待 并释放锁
System.out.println("生产者车间已满");
condition.await();
}
queue.offer(1);
condition.signal();
System.out.println("生产者生产了一个苹果");
Thread.sleep(1000);
}
} catch (Exception e) {
} finally {
lock.unlock();
}
}
}
private static class ConsumerRunnable implements Runnable {
@Override
public void run() {
try {
lock.lock();
while (true) {
if (queue.size() <= 0) {
//当队列为空 将消费者等待 并释放锁
System.out.println("消费者都吃完了");
condition.await();
}
System.out.println("消费者吃了一个苹果");
condition.signal();
queue.poll();
Thread.sleep(1000);
}
} catch (Exception e) {
} finally {
lock.unlock();
}
}
}
}
