概述
- 基于AQS（AbstractQueuedSynchronizer，concurrent包中的一个抽象类）
- AQS基于CAS（Compare and Swap，一种并发技术），CAS在java中靠Unsafe实现(native，依靠C++和汇编)
- Unsafe
- Access restriction: The type 'Unsafe' is not API (restriction on required library 'C:\Program Files\Java\jdk1.8.0_111\jre\lib\rt.jar')
- 已测试，确实跑不动
- @CallerSensitive，被注解的方法，需要方法调用者具备一定权限
- bootstrap class loader加载的类可调用
- extension class loader加载的类可调用
private final Sync sync; // 持有Sync，锁的功能来自于Sync
private volatile int state; // Sync中属性
// 默认非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
// 实际调用sync的方法，Sync是个抽象类，实现类有FairSync和NonfairSync，默认非公平
lock.lock()
sync.lock()
// NonfairSync实现
NonfairSync
final void lock() {
if (compareAndSetState(0, 1)) // 试着抢一把（这就是非公平的特点）
setExclusiveOwnerThread(Thread.currentThread()); // 设置当前线程独占锁
else
acquire(1); // 抢夺不成，排队去（公平锁直接来这）
}
源码解析 - lock()
thread-003.jpg
第1个线程
- compareAndSetState检查当前lock的state属性
- 采用unsafe.compareAndSwapInt(this, stateOffset, expect, update)
- 检查this对象的偏移地址为stateOffset的属性
- stateOffset：unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state"))
- 是否与expect一致
- 若是一致，就更新为update
- 意味着，原state值为0才表示这个lock没有被别人持有，当前线程才可以改值
- 检查this对象的偏移地址为stateOffset的属性
- NonfairSync持有一个从AbstractOwnableSynchronizer继承过来的exclusiveOwnerThread属性，表示独占锁线程
- 第一个获取锁的线程做了两件事情
- 设置AQS(即Sync)的state为1
- 设置exclusiveOwnerThread为当前线程
第2个线程
-
线程2也尝试获取同一个锁，在线程1没有释放锁的情况下必然是行不通的，所以线程2就要阻塞
-
那么，线程2如何被阻塞
thread-004.jpg
-
线程2执行compareAndSetState(0, 1)返回false
public final void acquire(int arg) {
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
- 先走第一个判断条件尝试获取锁
- 如果获取失败，走第二个判断条件添加FIFO同步队列
- 先看下tryAcquire方法做了什么？
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
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); //这里无需CAS，偏向锁
return true;
}
return false;
}
- state是volatile的，对线程2具有可见性
- 线程2拿到state，再次判断一下能否持有锁（可能线程1同步代码执行得比较快，这会儿已经释放了锁），不可以就返回false
- 一个锁最多可重入Integer.MAX_VALUE次，也就是2147483647
- 上述第二个判断条件中addWaiter逻辑
private Node addWaiter(Node mode) {
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;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
- 先用当前线程创建一个Node
- 判断tail有没有，有的话就以compareAndSetTail方式将当前node插在tail后面，并将node谁为tail
- 若是tail不存在，进入enq(node)
- 以compareAndSetHead的方式创建head
- 将当前node作为head.next，以compareAndSetTail方式将node设为tail
- 若是因为别的线程也在插入导致插入失败，则进入enq(node)一直插入直到成功
- 若是tail不存在，进入enq(node)
- head是个dummy节点，不对应线程
- 上述第二个判断条件中的acquireQueued逻辑
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);
}
}
- 如果线程是双向队列第一个真Node（前面还有一个dummy head），那么尝试获取锁
- 如果依旧失败，调用AQS的shouldParkAfterFailedAcquire(p, node)
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus; // node的前置节点
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.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
- head的waitStatus是0
- 把head的waitStatus设置为Noed.SIGNAL即-1，并返回false
- 下一次for循环，还是先尝试获取锁，不成功
- 继续走shouldParkAfterFailedAcquire，此时waitStatus为-1，返回true
- 执行parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted(); // 返回中断标识，并清除
}
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L); // 进入阻塞，醒来时也在这醒来
setBlocker(t, null);
}
- 至此，lock()告一段落
源码解析 - 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
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;
}
- 只有c等于0才会让free=true，即重入情况下需要多次释放
- 若c为0，置exclusiveOwnerThread为null，表示没有线程占有锁，返回true
- head不为空，且waitStatus为-1，所以执行unparkSuccessor(h)
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);
}
- s就是线程2的Node，不为null，并且s.waitStatus是0（未被赋过值）
- 线程2被unpark
- 有一个很重要的问题是：锁被解了怎么保证FIFO队列减少一个Node呢？回到AQS的acquireQueued方法
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 // 猜测是便于node本身之后的回收
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
- 线程2被唤醒后，依然进行for循环
- 线程2所在Node的前驱节点p其实就是那个dummy head
- 线程2尝试tryAcquire，成功
- 线程2节点置为head
- head这个局部变量在之前enq结束时就被回收，指向原head节点那块内存的引用就只剩下线程2节点
- 在setHead(node)中，node.prev被置空，意味着指向原head内存引用全部解除，这样它就被GC自动回收了
- 然后setHead中很关键的一句，node.thread=null，其实就是把现任head（即线程2节点）变成一个dummy head
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
- 为什么需要dummy head
- 因为一个线程随时可能因为中断而取消，而取消的话，Node自然就要被GC了，也就意味着head可能就没了。那GC前必然要把头Node的后继Node变为一个新的头而且要应对多种情况，这样就很麻烦。（虽然这个解释有点勉强，但貌似也可以接受）用一个没有thread的Node作为头，相当于起了一个引导作用，因为head没有线程，自然也不会被取消。
- 看一下上面unparkSuccessor的14行~20行，是为了应对在head的下一个node可能已被取消的情况；从尾到头遍历，找出离head最近的一个真实等待的node，对这个node进行unpark操作
- lockInterruptibly作用
- 获取锁阶段可以被外界interrupt，这样就可以在某种死锁的情况下获得退出
- 记得，它的作用不是在线程运行过程中接受interrupt并处理
网友评论