文章目录
简介
条件锁，是指在获取锁之后发现当前业务场景自己无法处理，而需要等待某个条件的出现才可以继续处理时使用的一种锁。
注意，这里的条件，必须是在获取锁之后去等待，对应到ReentrantLock的条件锁，就是获取锁之后才能调用condition.await()方法。
在java中，条件锁的实现都在AQS的ConditionObject类中，ConditionObject实现了Condition接口
使用示例
public class ReentrantLockTest {
public static void main(String[] args) throws InterruptedException {
// 声明一个重入锁
ReentrantLock lock = new ReentrantLock();
// 声明一个条件锁 ReentrantLockTest没有实现AQS，但是ReentrantLockTest中的sync实现了AQS，所以调用的sync来初始化
/** ReentrantLock
public Condition newCondition() {
return sync.newCondition();
}
* sync
final ConditionObject newCondition() {
return new ConditionObject();
}
*/
Condition condition = lock.newCondition();
new Thread(()->{
try {
lock.lock(); // 1
try {
System.out.println("before await"); // 2
// 等待条件 其实就是阻塞了
condition.await(); // 3
System.out.println("after await"); // 10
} finally {
lock.unlock(); // 11
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
// 这里睡1000ms是为了让上面的线程先获取到锁
Thread.sleep(1000);
lock.lock(); // 4
try {
// 这里睡2000ms代表这个线程执行业务需要的时间
Thread.sleep(2000); // 5
System.out.println("before signal"); // 6
// 通知条件已成立 唤醒
condition.signal(); // 7
System.out.println("after signal"); // 8
} finally {
lock.unlock(); // 9
}
}
}
上面的代码很简单，一个线程等待条件，另一个线程通知条件已成立，后面的数字代表代码实际运行的顺序，其实就是阻塞和唤醒，就是Object的wait和notify
源码分析
ConditionObject的主要属性
public class ConditionObject implements Condition, java.io.Serializable {
/** First node of condition queue. 条件队列的第一个结点*/
private transient Node firstWaiter;
/** Last node of condition queue.条件队列的最后一个结点 */
private transient Node lastWaiter;
}
可以看到条件锁中也维护了一个队列，为了和AQS的队列区分，我这里称为条件队列，firstWaiter是队列的头节点，lastWaiter是队列的尾节点
lock.newCondition()方法
新建一个条件锁，ReentrantLockTest没有实现AQS，但是ReentrantLockTest中的sync实现了AQS，所以调用的sync来初始化
// ReentrantLock.newCondition()
public Condition newCondition() {
return sync.newCondition();
}
// ReentrantLock.Sync.newCondition()
final ConditionObject newCondition() {
return new ConditionObject();
}
// AbstractQueuedSynchronizer.ConditionObject.ConditionObject()
public ConditionObject() { }
新建一个条件锁最后就是调用的AQS中的ConditionObject类来实例化条件锁。
condition.await()方法
- condition.await()方法，表明现在要等待条件的出现，也就是阻塞此线程，并且会把线程获取的锁全部释放。 其中提到
addConditionWaiter、fullyRelease、isOnSyncQueue
// AbstractQueuedSynchronizer.ConditionObject.await()
public final void await() throws InterruptedException {
// 如果线程中断了，抛出异常
if (Thread.interrupted())
throw new InterruptedException();
// 添加节点到Condition的队列中，并返回该节点
Node node = addConditionWaiter();
// 完全释放当前线程获取的锁
// 因为锁是可重入的，所以这里要把获取的锁全部释放
int savedState = fullyRelease(node);
int interruptMode = 0;
// 是否在同步队列中，在 就阻塞
while (!isOnSyncQueue(node)) {
// 阻塞当前线程
LockSupport.park(this);
// 上面部分是调用await()时释放自己占有的锁，并阻塞自己，等待条件的出现
// *************************分界线************************* //
// 下面部分是条件已经出现，尝试去获取锁
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
// 尝试获取锁，注意第二个参数savedState，就是占有多少把锁
// 如果没获取到会再次阻塞
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
// 清除取消的节点
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
// 线程中断相关
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
- condition.
addConditionWaiter方法把结点入队，并且等待状态是CONDITION，返回此线程节点
// AbstractQueuedSynchronizer.ConditionObject.addConditionWaiter
private Node addConditionWaiter() {
Node t = lastWaiter;
// 如果条件队列的尾节点已取消，从头节点开始清除所有已取消的节点
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
// 重新获取尾节点
t = lastWaiter;
}
// 新建一个节点，它的等待状态是CONDITION
Node node = new Node(Thread.currentThread(), Node.CONDITION);
// 如果尾节点为空，则把新节点赋值给头节点（相当于初始化队列）
// 否则把新节点赋值给尾节点的nextWaiter指针
if (t == null)
// 头结点是此线程结点node，所以头结点含有线程 这里和AQS队列不同
firstWaiter = node;
else
t.nextWaiter = node;
// 尾节点指向新节点
lastWaiter = node;
// 返回新节点
return node;
}
- AQS.
fullyRelease释放次线程结点获取的所有锁，返回获取锁的次数
// AbstractQueuedSynchronizer.fullyRelease
final int fullyRelease(Node node) {
boolean failed = true;
try {
// 获取状态变量的值，重复获取锁，这个值会一直累加
// 所以这个值也代表着获取锁的次数
int savedState = getState();
// 一次性释放所有获得的锁
if (release(savedState)) {
failed = false;
// 返回获取锁的次数
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
- AQS.
isOnSyncQueue方法判断该结点是否在AQS队列中
// AbstractQueuedSynchronizer.isOnSyncQueue
final boolean isOnSyncQueue(Node node) {
// 如果等待状态是CONDITION，或者前一个指针为空，返回false
// 说明还没有移到AQS的队列中
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
// 如果next指针有值，说明已经移到AQS的队列中了
if (node.next != null) // If has successor, it must be on queue
return true;
// 从AQS的尾节点开始往前寻找看是否可以找到当前节点，找到了也说明已经在AQS的队列中了
return findNodeFromTail(node);
}
解释：
（1）Condition的队列和AQS的队列不完全一样；
// AQS的队列头节点是不存在任何值的，是一个虚节点；
// Condition的队列头节点是存储着实实在在的元素值的，是真实节点。 在addConditionWaiter中有提到
（2）各种等待状态（waitStatus）的变化；
// 首先，在条件队列中，新建节点的初始等待状态是CONDITION(-2)；
// 其次，移到AQS的队列中时等待状态会更改为0(AQS队列节点的初始等待状态为0)；
// 然后，在AQS的队列中如果需要阻塞，会把它上一个节点的等待状态设置为SIGNAL(-1),在AQS中的shouldParkAfterFailedAcquire提到
// 最后，不管在Condition队列还是AQS队列中，已取消的节点的等待状态都会设置为CANCELLED(1)；
（3）相似的名称；
// AQS中下一个节点是next，上一个节点是prev；Condition中下一个节点是nextWaiter，没有上一个节点。
await()方法的大致流程：
（1）新建一个节点加入到条件队列中去 addConditionWaiter
（2）完全释放当前线程占有的锁；fullyRelease
（3）阻塞当前线程，并等待条件的出现；addConditionWaiter的while中
（4）条件已出现（此时节点已经移到AQS的队列中），尝试获取锁；addConditionWaiter的while后
也就是说await()方法内部其实是先释放锁->等待条件->再次获取锁的过程。
condition.signal()方法
- condition.
signal()方法通知条件已经出现，先做判断，如果条件队列中有头结点，则去通知头结点，但不一定就是唤醒头结点
// AbstractQueuedSynchronizer.ConditionObject.signal
public final void signal() {
// 如果不是当前线程占有着锁，调用这个方法抛出异常
// 说明signal()也要在获取锁之后执行
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
// 条件队列的头节点
Node first = firstWaiter;
// 如果有等待条件的节点，则通知它条件已成立
if (first != null)
doSignal(first);
}
- condition.
doSignal方法将条件队列头结点转移到AQS队列中，但是并不一定会加入，如果此结点是取消状态则直接寻找下一个结点
// AbstractQueuedSynchronizer.ConditionObject.doSignal
private void doSignal(Node first) {
do {
// 移到条件队列的头节点往后一位，头结点没有后续结点，则直接将尾结点设为空
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
// 相当于把头节点从队列中出队
first.nextWaiter = null;
// 转移节点到AQS队列中，转移失败则寻找下一个结点
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
- AQS.
transferForSignal方法将结点入AQS队列
// AbstractQueuedSynchronizer.transferForSignal
final boolean transferForSignal(Node node) {
// 把节点的状态更改为0(也就是默认状态)，也就是说即将移到AQS队列中
// 如果失败了，说明节点已经被改成取消状态了
// 返回false，通过上面的循环可知会寻找下一个可用节点
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
// 调用AQS的入队方法把节点移到AQS的队列中
// 注意，这里enq()的返回值是node的上一个节点，也就是旧尾节点
Node p = enq(node);
// 上一个节点的等待状态
int ws = p.waitStatus;
// 如果上一个节点已取消了，或者更新状态为SIGNAL失败（也是说明上一个节点已经取消了）
// 则直接唤醒当前节点对应的线程
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
// 如果更新上一个节点的等待状态为SIGNAL成功了
// 则返回true，这时上面的循环不成立了，退出循环，也就是只通知了一个节点
// 此时当前节点还是阻塞状态
// 也就是说调用signal()的时候并不一定真正唤醒一个节点
// 只是把节点从条件队列移到AQS队列中
return true;
}
signal()方法的大致流程为：
（1）从条件队列的头节点开始寻找一个非取消状态的节点；
（2）把它从条件队列移到AQS队列；
（3）且只移动一个节点；
注意，这里调用signal()方法后并不一定真正唤醒一个节点，那么，唤醒一个节点是在啥时候呢？
开头例子，signal()方法后，最终会执行lock.unlock()方法，此时才会真正唤醒一个节点，唤醒的这个节点如果曾经是条件节点的话又会继续执行await()方法“分界线”下面的代码。
用一个图来表示的话，下面这个图可以大致表示一下
总结
（1）重入锁是指可重复获取的锁，即一个线程获取锁之后再尝试获取锁时会自动获取锁；
（2）在ReentrantLock中重入锁是通过不断累加state变量的值实现的；
（3）ReentrantLock的释放要跟获取匹配，即获取了几次也要释放几次；
（4）ReentrantLock默认是非公平模式，因为非公平模式效率更高；
（5）条件锁是指为了等待某个条件出现而使用的一种锁；
（6）条件锁比较经典的使用场景就是队列为空时阻塞在条件notEmpty上；
（7）ReentrantLock中的条件锁是通过AQS的ConditionObject内部类实现的；
（8）await()和signal()方法都必须在获取锁之后释放锁之前使用；
（9）await()方法会新建一个节点放到条件队列中，接着完全释放锁，然后阻塞当前线程并等待条件的出现；
（10）signal()方法会寻找条件队列中第一个可用节点移到AQS队列中；
（11）在调用signal()方法的线程调用unlock()方法才真正唤醒阻塞在条件上的节点（此时节点已经在AQS队列中）；
（12）之后该节点会再次尝试获取锁，后面的逻辑与lock()的逻辑基本一致了。