目录
- AQS结构
- Node结构
- AQS阻塞链表
- ReentrantLock
- ReentrantLock FairSync
- ReentrantLock NonfairSync
- Condition
- CountDownLatch
为了搞清楚AQS到底是有什么特性需要先看看AQS有哪些属性
AQS结构
通过结构图可以看到主要的属性有head,tail,state, exclusiveOwnerThread 下面分别对其进行说明
// 头结点，你直接把它当做 当前持有锁的线程 可能是最好理解的
private transient volatile Node head;
// 阻塞的尾节点，每个新的节点进来，都插入到最后，也就形成了一个隐视的链表
private transient volatile Node tail;
// 这个是最重要的，不过也是最简单的，代表当前锁的状态，0代表没有被占用，大于0代表有线程持有当前锁
// 之所以说大于0，而不是等于1，是因为锁可以重入嘛，每次重入都加上1
private volatile int state;
// 代表当前持有独占锁的线程，举个最重要的使用例子，因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次，所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer
Node结构
//用来表示在共享模式下等待的标记
static final Node SHARED = new Node();
//用来表示在独占模式下等待的标记
static final Node EXCLUSIVE = null;
/** waitStatus 的值，用来表示线程已经被取消 */
static final int CANCELLED = 1;
/** waitStatus的值，用来表示后继线程都需要被挂起*/
static final int SIGNAL = -1;
/** waitStatus 的值，表示线程在一个条件上等待 */
static final int CONDITION = -2;
/**
* waitStatus 的值，表示下一个acquireShared应该无条件传播
*/
static final int PROPAGATE = -3;
volatile int waitStatus;
//前驱节点
volatile Node prev;
//后继节点
volatile Node next;
// 当前线程
volatile Thread thread;
AQS阻塞链表
通过分析Node,AQS的链表结构就很清晰了，注意head节点是持有锁的，所说的阻塞队列是从第二个节点算起的，通过后面代码可以发现如果节点的直接前驱是可以直接
尝试获取锁的。
ReentrantLock
ReentrantLock的公平锁
lock
static final class FairSync extends Sync {
// 争锁
final void lock() {
acquire(1);
}
}
acquire是在AQS中实现的，其代码如下：public final void acquire(int arg) {
//尝试获取锁失败后,这个时候需要把当前线程挂起，放到阻塞队列中;
//如果在获取锁的等待过程中发生中断，则执行selfInterrupt
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire又是在FairSync中实现的
protected final boolean tryAcquire(int acquires) {
//获取当前线程
final Thread current = Thread.currentThread();
//获取当前节点的状态
int c = getState();
//state == 0 此时此刻没有线程持有锁
if (c == 0) {
//如果没有前驱节点在排队，则CAS获取锁；如果获得了锁则将独占锁的持有者设置为当前线程
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;
}
//如果没有获取到锁则返回false
return false;
}
如果tryAcquire(1)没有获取到锁，则继续执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
private Node addWaiter(Node mode) {
//对当前线程创建一个新节点
Node node = new Node(Thread.currentThread(), mode);
//把当前node加到链表的最后面去，也就是进到阻塞队列的最后
Node pred = tail;
//如果tail节点不为null
if (pred != null) {
//将当前节点的前驱节点设置为目前的队尾节点
node.prev = pred;
//CAS将当前节点设置为队尾，如果设置成功则将之前的队尾的后继节点设置为当前节点，并返回
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//pred==null(队列是空的) 或者 CAS失败(有线程在竞争入队)，则执行enq(node)
//采用自旋的方式入队
enq(node);
return node;
}
// 采用自旋的方式入队
// 之前说过，到这个方法只有两种可能：阻塞队列为空，或者有线程竞争入队，
// 自旋在这边的语义是：CAS设置tail过程中，竞争一次竞争不到，就多次竞争，总会排到的
private Node enq(final Node node) {
//自旋
for (;;) {
//队尾节点
Node t = tail;
//如果队尾为空
if (t == null) {
//初始化head节点，原来head和tail初始化的时候都是null，
//此时初始化阻塞队列，CAS设置一个新节点作为头节点
if (compareAndSetHead(new Node()))
//这个时候有了head，但是tail还是null,需要把tail指向head
tail = head;
} else {
//队尾节点不为空，这个套在无限循环里，反正就是将当前线程排到队尾，
//有线程竞争的话排不上重复排
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
//下面这个方法，参数node，经过addWaiter(Node.EXCLUSIVE)，此时已经进入阻塞队列
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
//是否中断标记
boolean interrupted = false;
//自旋
for (;;) {
//获取前驱节点
final Node p = node.predecessor();
// p == head 说明当前节点虽然进到了阻塞队列，但是是阻塞队列的第一个，因为它的前驱是head
// 注意，阻塞队列不包含head节点，head一般指的是占有锁的线程，head后面的才称为阻塞队列；
//如果前驱节点是头节点则可以执行tryAcquire,因为初始化队列的时候头节点并不占有锁，
//tryAcquire方法执行成功表示已经抢到锁；则将该节点设置为头节点，后继节点为null
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);
}
}
/*
* 节点获取锁失败后，检查，更新节点waitstatus
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//获取前驱节点的waitStatus的值
int ws = pred.waitStatus;
//前驱节点的 waitStatus == -1 ，说明前驱节点状态正常，当前线程需要挂起，直接可以返回true
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
/*
* 前驱节点 waitStatus大于0 ，大于0 说明前驱节点取消了排队
* 进入阻塞队列排队的线程会被挂起，而唤醒的操作是由前驱节点完成的。
* 所以下面这块代码说的是将当前节点的prev指向waitStatus<=0的节点，跳过已经取消的节点
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
*前驱节点的waitStatus不等于-1和1；都没有看到有设置waitStatus的，
*所以每个新的node入队时，waitStatu都是
* 用CAS将前驱节点的waitStatus设置为Node.SIGNAL(也就是-1)
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
shouldParkAfterFailedAcquire(Node pred, Node node)
这个方法结束根据返回值我们简单分析下：
如果返回true, 说明前驱节点的waitStatus==-1，是正常情况，那么当前线程需要被挂起，等待以后被唤醒
如果返回false, 说明当前不需要被挂起，为什么呢？ 因为如果返回false,则表示pred已经是头节点了；也就是当前节点是head的直接后继，再次循环的时候就可以直接尝试获取锁了，没有必要在被挂起了。
如果返回true则执行 parkAndCheckInterrupt()
private final boolean parkAndCheckInterrupt() {
//当前线程挂起
LockSupport.park(this);
//中断检查
return Thread.interrupted();
}
至此，线程已经被挂起了，下面在看一下解锁相关代码
unlock
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
//尝试释放锁，如果释放成功
if (tryRelease(arg)) {
Node h = head;
//头节点不为null且waitStatus不是0则唤醒后继节点
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
//当前状态减去releases;
int c = getState() - releases;
//判断当前线程是否是持有锁的线程，该方法可以放在第一句执行
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
// 是否完全释放锁
boolean free = false;
//如果c=0表示锁已经完全释放了
if (c == 0) {
free = true;
//将持有锁的线程设置为null
setExclusiveOwnerThread(null);
}
//将c的值更新到state
setState(c);
return free;
}
/**
*如果存在后继节点则唤醒后继节点
*/
private void unparkSuccessor(Node node) {
//获取当前节点的值
int ws = node.waitStatus;
//如果当前节点的status为负数则CAS修改为0
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//下面的代码就是唤醒后继节点，但是有可能后继节点取消了等待（waitStatus==1）
//获取当前节点的后继节点
Node s = node.next;
//如果后继节点为null或后继节点已经被取消
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);
}
此时线程被唤醒，则阻塞的线程则从这个 parkAndCheckInterrupt() 方法的中断检查开始执行
如果线程没有中断，则会在 acquireQueued中自旋进行锁竞争
在看看指定超时时间的场景
/**
* 在指定的时间内线程没有中断则获取没有被其他线程持有的锁；如果竞争到锁则立即返回true,
*
**/
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(arg, nanosTimeout);
}
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
//如果指定的超时时间小于或等于0则直接返回false
if (nanosTimeout <= 0L)
return false;
//根据当前时间加上指定的超时时间得到超时时间点
final long deadline = System.nanoTime() + nanosTimeout;
//向阻塞队列中添加节点
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
//自旋
for (;;) {
//获取当前节点的直接前驱节点
final Node p = node.predecessor();
//如果直接前驱是head节点则尝试获取锁，如果可以获取锁，则将该节点设置为头节点，同时将该节点的后继节点设置为null
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return true;
}
//计算剩余等待时间
nanosTimeout = deadline - System.nanoTime();
//如果剩余可等待时间小于或等于0则返回false
if (nanosTimeout <= 0L)
return false;
//还可以等待一段时间，则检查并更新线程状态，如果需要被挂起且剩余等待时间大于自旋优先的阈值，则将该线程最多挂起nanosTimeout 纳秒
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
//最多挂起 nanosTimeout 纳秒
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
//如果超时或线程中断抛出InterruptedException异常时failed为true
if (failed)
cancelAcquire(node);
}
}
/**
* 取消正在获取锁的尝试
*/
private void cancelAcquire(Node node) {
// 如果节点不存在则直接返回
if (node == null)
return;
//将node节点持有的线程设置为null
node.thread = null;
// 跳过被取消的前驱节点
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
// predNext is the apparent node to unsplice. CASes below will
// fail if not, in which case, we lost race vs another cancel
// or signal, so no further action is necessary.
//前驱节点的后继节点
Node predNext = pred.next;
//将当前节点的等待状态设置为取消状态
node.waitStatus = Node.CANCELLED;
// 如果当前节点就是队尾，将该节点删除
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
//如果自己不是队尾节点
int ws;
//前驱节点也不是头节点且 等待状态为-1或cas更新为-1同时前驱节点线程不为null
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
//当前节点的后继节点
Node next = node.next;
//后继节点不为null且没有持有锁，则将前驱节点的直接后继执行后继节点，进而将该节点从链表中删除
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
//如果前驱节点不满足上述条件则唤醒前驱节点
unparkSuccessor(node);
}
//将该节点的后继执行自己，断开链接
node.next = node; // help GC
}
}
通过上面的源码可以发现虽然线程中断了但是该节点依然会参与锁竞争，下面我们在看一下可中断的加锁实现/**
* 竞争独占锁，线程中断则终止
*/
public final void acquireInterruptibly(int arg)
throws InterruptedException {
//首先检查线程是否中断，如果中断则抛出异常
if (Thread.interrupted())
throw new InterruptedException();
//尝试获取锁，如果获取锁失败则执行doAcquireInterruptibly
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
//将当前节点添加到阻塞队列
final Node node = addWaiter(Node.EXCLUSIVE);
boolean failed = true;
try {
//自旋
for (;;) {
//获取当前节点的直接前驱节点
final Node p = node.predecessor();
//如果直接前驱是head节点则尝试获取锁，如果可以竞争到锁则将当前节点设置为头节点
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
//判断需要挂起，同时线程已经中断则抛出中断异常
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
// 如果通过 InterruptedException 异常出去，那么 failed 就是 true 了
if (failed)
cancelAcquire(node);
}
}
上面主要讲述了公平锁，如果是非公平锁又是怎样的呢？
NonfairSync
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
final void lock() {
//直接进行CAS尝试进行获取锁，这个比公平锁多出来的操作
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
final boolean nonfairTryAcquire(int acquires) {
//获取当前线程
final Thread current = Thread.currentThread();
//获取state值
int c = getState();
//c为0，表示没有线程持有锁，则进行cas操作获取锁
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;
}
公平锁和非公平锁只有两处不同：
- 非公平锁在调用 lock 后，首先就会调用 CAS 进行一次抢锁，如果这个时候恰巧锁没有被占用，那么直接就获取到锁返回了。
- 非公平锁在 CAS 失败后，和公平锁一样都会进入到 tryAcquire 方法，在 tryAcquire 方法中，如果发现锁这个时候被释放了（state == 0），非公平锁会直接 CAS 抢锁，但是公平锁会判断等待队列是否有线程处于等待状态，如果有则需要排队等待。
公平锁和非公平锁就这两点区别，如果这两次 CAS 都不成功，那么后面非公平锁和公平锁是一样的，都要进入到阻塞队列等待唤醒。
Condition
上面主要走读了独占锁的实现逻辑，下面主要走读Condition相关的实现逻辑
public class BoundedBuffer {
final Lock lock = new ReentrantLock();
final Condition notFull = lock.newCondition();
final Condition notEmpty = lock.newCondition();
final Object[] items = new Object[100];
int putptr, takeptr, count;
public void put(Object obj) throws InterruptedException {
lock.lock();
try {
//如果当前数量已经达到上限则在notFull条件上等待
while (count == items.length) {
notFull.await();
}
items[putptr] = obj;
if (++putptr == items.length) putptr = 0;
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (count == 0) {
notEmpty.await();
}
Object obj = items[takeptr];
if (++takeptr == items.length) takeptr = 0;
--count;
notFull.signal();
return obj;
} finally {
lock.unlock();
}
}
}
public Condition newCondition() {
return sync.newCondition();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
下面看下ConditionObject结构
// 条件队列的第一个节点
private transient Node firstWaiter;
// 条件队列的最后一个节点
private transient Node lastWaiter;
首先看在条件上阻塞操作public final void await() throws InterruptedException {
//如果线程已经中断，则抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
//向条件等待队列中添加一个新的Waiter节点
Node node = addConditionWaiter();
//完全释放该节点的锁，只有完全释放才可以避免重入问题，并将释放前的值返回；后续再次获取锁的时候需要用到
int savedState = fullyRelease(node);
int interruptMode = 0;
//如果当前节点已经在阻塞队列中或者在等待过程中中断过，如果该节点不在阻塞队列中则线程在该处挂起
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
//线程唤醒后会进行中断检查
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
//线程唤醒后将该节点加入到阻塞队列 如果在唤醒前就发生了中断则interruptMode 设置为REINTERRUPT
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
//如果后继节点不为null 则进行清理取消的节点
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
//interruptMode不为0则会根据interruptMode的值决定是抛出异常还是执行中断或则什么都不做
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
将节点添加到等待队列 /**
* 向等待队列中添加一个新的waiter,并返回这个新的等待节点
*/
private Node addConditionWaiter() {
//等待队列的最后一个节点
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
//如果最后一个节点不为null但是状态又不是condition,则需要清理等待队列
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
//新建一个等待节点
Node node = new Node(Thread.currentThread(), Node.CONDITION);
//如果最后一个节点为null,则将该节点设置为最后节点，否则将该节点添加到最后节点之后并将该节点设置为最后节点
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
//等待队列是一个单向链表，遍历链表将已经取消等待的节点清除出去
private void unlinkCancelledWaiters() {
//获取第一个等待节点
Node t = firstWaiter;
Node trail = null;
//如果第一个节点不为null
while (t != null) {
//获取下一个节点
Node next = t.nextWaiter;
//如果第一个节点的等待状态不是condition
if (t.waitStatus != Node.CONDITION) {
//将下一个节点的直接后继设置为null
t.nextWaiter = null;
//如果trail为null则将第二个节点提升为第一个节点，否则trail的直接后继节点指向next,将当前节点从链表中删除
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
//遍历到队尾，将trial指向到节点设置为最后一个等待节点
if (next == null)
lastWaiter = trail;
}
else//如果第一个等待节点是condition状态，则trail指向第一个节点，t指向下一个节点
trail = t;
t = next;
}
}
final int fullyRelease(Node node) {
boolean failed = true;
try {
//获取当前同同步状态
int savedState = getState();
//这里使用了当前的 state 作为 release 的参数，也就是完全释放掉锁，将 state 置为 0
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
final boolean isOnSyncQueue(Node node) {
//如果 waitStatus 还是 Node.CONDITION，也就是 -2，那肯定就是还在条件队列中
//如果 node 的前驱 prev 指向还是 null，说明肯定没有在 阻塞队列
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
//如果存在后继节点则一定在阻塞队列中
//node.prev() != null 来推断出 node 在阻塞队列？不能是因为在addWaiter方法中会将node.prev指向为队尾，但是此时
//节点上位加入到阻塞队列中
if (node.next != null)
return true;
// 这个方法从阻塞队列的队尾开始从后往前遍历找，如果找到相等的，说明在阻塞队列，否则就是不在阻塞队列
return findNodeFromTail(node);
}
private boolean findNodeFromTail(Node node) {
Node t = tail;
for (;;) {
//如果t==node表示该节点在阻塞队列中
if (t == node)
return true;
//如果全部遍历完还是没有找到相同的节点则返回false
if (t == null)
return false;
t = t.prev;
}
}
如果当前等待节点不在阻塞队列中，则需要执行LockSupport.park(this)将线程挂起；那么什么时候唤醒呢？后面会详细说；假设此时线程被唤醒，唤醒后需要进行中断检查，如果中断检查当返回值不是0，即发生了中断则从while循环中跳出。 /**
* 中断检查，如果在唤醒前中断则返回 THOW_IE,如果在唤醒后中断则返回
* REINTERRUPT；如果没有中断则返回0
*/
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}
从while循环中跳出后，执行 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)，acquireQueued是做什么的呢？上文已经有介绍过了，主要是用来锁竞争和线程挂起。如果竞争到了锁且未发生中断，则await阻塞结束，执行后续业务代码。但是我们知道acquireQueued要求当前Node是在阻塞队列的，那么什么时候加入阻塞队列的呢？下面我们看一下signal方法/**
* 如果等待队列存在线程，将等待时间最长的线程从等待队列转移到锁的阻塞队列中
*
*/
public final void signal() {
//没有独占锁抛出异常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
//condition等待队列的第一个节点
Node first = firstWaiter;
//如果第一个节点不为null
if (first != null)
doSignal(first);
}
private void doSignal(Node first) {
do {
//当前节点的下一个节点设置为第一个节点，如果这个节点为null则将最后一个节点设置为null，当前节点的下一个节点也设置为null
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
//循环推出的条件是：将condition等待队列的第一个节点转移到阻塞队列成功或等待队列中元素为空
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
/*
*CAS 设置节点waitStatus的值为0，如果设置失败,即当前节点的状态不是CONDITION,标示该节点已经被取消了
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
//将该节点自旋加入阻塞队列的队尾，并返回该节点的前驱节点
Node p = enq(node);
//获取前驱节点的状态
int ws = p.waitStatus;
// ws > 0 说明 node 在阻塞队列中的前驱节点取消了等待锁，直接唤醒 node 对应的线程
//如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用，节点入队后，需要把前驱节点的状态设为 Node.SIGNAL(-1)，如果设置失败则唤起线程
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
如果发生了中断会怎么处理内？
/**
* 如果interruptMode是 THROW_IE 则抛出中断异常
* 如果interruptMode是 REINTERUPT 则再次执行中断
* 其他情况什么都不做
*/
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}
至此已经将排它锁相关逻辑介绍完成，下面介绍共享锁相关逻辑。
CountDownLatch
CountDownLatch结构
首先看一下await()方法
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* 在共享模式下获取锁，线程中断则终止
*/
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//首先进行线程中断检查，如果中断则抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
//尝试获取锁，如果获取锁失败则执行doAcquireSharedInterruptibly()方法，
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
//只有state为0的时候才返回1，否则就返回-1
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
/**
* 在共享可中断模式下获取锁
*/
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//为当前线程创建一个共享模式节点并加入到阻塞队列中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
//自旋
for (;;) {
//获取当前新建节点的直接前驱节点
final Node p = node.predecessor();
//如果直接前驱节点为head节点
if (p == head) {
//尝试获取锁
int r = tryAcquireShared(arg);
//获取到锁
if (r >= 0) {
//将当前节点设置为头节点和传播
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//如果当前节点的直接前驱不是head或没有获取到锁，则需要判断当前节点是否需要挂起，如果需要则通过parkAndCheckInterrput()方法挂起
//如果发生中断则抛出异常
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
/**
*
* 设置队列头，并检查后继节点是否可以在共享模式下等待，如果不需要等待则可以唤醒后继节点
*/
private void setHeadAndPropagate(Node node, int propagate) {
//当前头节点
Node h = head;
//将当前节点设置为新的头节点
setHead(node);
//如果propagate大于0，通过上面代码可以知道propagate的值为1,如果下一个节点不为null且只共享模式则执行doReleaseShared()方法
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
/**
* 共享模式下释放锁的操作,唤醒后继节点同时保证传播功能
*/
private void doReleaseShared() {
for (;;) {
//头节点
Node h = head;
//如果头节点不为null同时也不能与尾节点（初始化的时候头节点等于尾节点）
if (h != null && h != tail) {
//头节点的等待状态
int ws = h.waitStatus;
//如果头节点的状态时SINGLE(-1),则将头节点设置为0，失败重试,成功则唤醒头节点
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
unparkSuccessor(h);
}//如果头节点是新加入的节点，则将其状态设置为-3
//为什么不可以是-1呢？这是因为这个方法执行结束后需要执行chouldParkAfterFail,-1在此会被挂起
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
//如果头节点没有变化则跳出循环，反之则继续循环
if (h == head)
break;
}
}
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected boolean tryReleaseShared(int releases) {
// 对state值进行自减，如果结果为0则返回true
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//如果没有获取锁则执行doAcquireSharedNanos方法
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
//计算出deadLine
final long deadline = System.nanoTime() + nanosTimeout;
//为当前线程创建一个共享模式节点并添加到阻塞队列
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return true;
}
}
//计算剩下等待时间
nanosTimeout = deadline - System.nanoTime();
//如果剩下等待时间小于或等于0则表示超时了，返回false
if (nanosTimeout <= 0L)
return false;
//判断是否需要挂起，如果剩余等待时间小于阈值则不挂起而是通过自旋处理；
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
相关推荐
在本篇中，我们将深入分析AQS的条件队列，它是实现高级同步机制如`ReentrantLock`和`CountDownLatch`的关键部分。 条件队列是AQS中与`Condition`接口相关的部分，它允许线程在满足特定条件时等待，而不是简单地阻塞...
**Java并发系列之AbstractQueuedSynchronizer源码分析概要** **1. AbstractQueuedSynchronizer（AQS）的定义与作用** AbstractQueuedSynchronizer（AQS）是Java并发编程中的核心组件，它是一个抽象的、基于FIFO...
【Java并发系列之AbstractQueuedSynchronizer源码分析（独占模式）】 AbstractQueuedSynchronizer（AQS）是Java并发编程中一个重要的工具，它是Java并发包`java.util.concurrent.locks`中的核心抽象类，用于构建锁...
《Java并发系列之AbstractQueuedSynchronizer源码分析（共享模式）》 AbstractQueuedSynchronizer（AQS）是Java并发编程中一个重要的工具，它是Java并发包`java.util.concurrent.locks`中的核心抽象类，用于构建锁...
接下来，我们来具体分析一下AQS的源码。AQS中定义了一个名为state的volatile变量，用于表示同步状态。这个变量有三种操作方法：getstate()、setstate()和compareAndSetState()，分别用于获取、设置和原子性地更新...
### ReentrantLock源码分析 #### 一、ReentrantLock简介 ReentrantLock是一个基于`AbstractQueuedSynchronizer`(AQS)实现的高级锁工具类。与传统的synchronized关键字相比，ReentrantLock提供了更多控制手段，比如...
java8 源码 AQS补充材料 美团技术团队《从ReentrantLock的实现看AQS的原理及应用》 ...《AbstractQueuedSynchronizer源码分析(基于Java8)》 waterstone《Java并发AQS详解》 英文论文的中文翻译： AQS作者的英文论文:
ReentrantLock Lock 加锁过程源码分析图，AQS 源码分析
根据给定文件的信息，我们可以深入理解AQS（AbstractQueuedSynchronizer）独占锁之ReentrantLock的源码分析及其实现原理。这不仅包括ReentrantLock本身的特性，还包括了其背后的AQS框架是如何工作的。 ### 一、管程...
Java并发包源码分析(JDK1.8)：囊括了java.util.concurrent包中大部分类的源码分析，其中涉及automic包，locks包(AbstractQueuedSynchronizer、ReentrantLock、ReentrantReadWriteLock、LockSupport等)，queue...
本文将详细分析AQS的源码，探讨其工作机制，以及在Java中如何实现不同类型的锁。 首先，我们需要了解锁的基本类型。在Java中，锁主要分为两类：悲观锁和乐观锁。悲观锁认为并发操作会导致数据不一致，因此在操作...
Java并发编程之Condition源码分析 Condition是Java并发编程中的一种同步机制，用于实现异步通信的同步。Condition主要有两个方法：await()和signal()。await()方法可以阻塞当前线程，并释放锁，而signal()方法可以...
Java并发结合源码分析AQS原理 Java并发编程中，AQS（AbstractQueuedSynchronizer）是一个核心组件，它提供了一个基于FIFO队列和状态变量的基础框架，用于构建锁和其他同步装置。在这篇文章中，我们将深入探讨AQS的...
Java并发系列之ReentrantLock源码分析 ReentrantLock是Java 5.0中引入的一种新的加锁机制，它实现了Lock接口，并提供了与synchronized相同的互斥性和内存可见性。ReentrantLock的底层实现是通过AQS来实现多线程同步...
Java并发系列之CountDownLatch源码分析 CountDownLatch是一种非常有用的工具类，用于拦截一个或多个线程，使其在某个条件成熟后再执行。它的内部提供了一个计数器，在构造闭锁时必须指定计数器的初始值，且计数器的...
描述：Worker类实现Runnable接口、继承AbstractQueuedSynchronizer类 Thread thread : 工作线程，用于处理任务 Runnable firstTask : 第一个任务，当线程池worker对象达到corePoolSize且workQueue满时，worker对象的...
源码分析能帮助我们理解这些数据结构的底层实现，如扩容策略、查找效率等。 5. **IO与NIO** - JDK 1.7包含传统的IO流和非阻塞I/O（NIO）系统，如`BufferedReader`、`FileInputStream`以及`Channels`和`Selectors`...
CountDownLatch源码解析之await() CountDownLatch是Java并发编程中常用的同步工具，通过await()方法可以让当前线程处于阻塞状态，直到锁存器计数为零（或者线程中断）。下面我们将详细解析CountDownLatch源码之...
本文将深入分析Worker线程的执行流程，特别是其源码实现。 首先，Worker类是ThreadPoolExecutor的一个内部类，它继承了AbstractQueuedSynchronizer（AQS），这是一个抽象的同步队列，用于实现锁和其他同步组件的...
文档内容丰富，既包括了Java的基本语法、源码分析、多线程处理、IO流操作、设计模式、常用框架、数据库技术、数据结构与算法、JVM原理、Web开发技术，也包括了Linux操作系统、Redis数据库、UML绘图以及JDK的新特性等...