AbstractQueuedSynchronizer的基本数据结构为Node
Node SHARED 代表一个标识位,指示节点使用共享模式等待
Node EXCLUSIVE 代表一个标识位,指示接的使用独占模式等待
int CANCELLED = 1 等待状态值,指示线程已被取消(节点因为超市或者被打断而取消,一个界面不会持有这种状态,取消节点持有的线程不会重新阻塞)
int SIGNAL = -1 等待状态值,指示后继的线程需要取消阻塞(当前节点的后继节点已经通过park阻塞,因此当前节点释放或者取消的时候必须unpark它的后继节点,为了避免竞争,acquire()必须指示需要signal信号，然后重试院子acquire方法,最后再失败的情况下阻塞)
int CONDITION = -2 等待状态值,指示线程由于阻塞而处于等待状态(节点当前处于等待队列中,直到状态在某个时间点被转化为0前它都不会作为同步队列的一个节点被使用)
int PROPAGATE = -3 等待状态值,指示线程由于阻塞而处于取消等待(共享模式下的释放，应用被传播到其他节点.这个状态会在doReleaseShared方法中被设置,来确保传播的持续,及时其他操作介入)
volatile Node prev 当前节点的前驱节点
volatile Node next 当前节点的后继节点
Thread thread 当前节点持有的线程,构造方法中初始化,使用完毕之后设置为null
Node nextWaiter 指向下一个处于阻塞等待的节点
ReentrantLock中的内部类Sync实现了AbstractQueuedSynchronizer,而Sync分为公平锁FairSync和非公平锁NonFairSync。
AbstractQueuedSynchronizer里面维护着一个队列,如果是公平锁的话,则强制按照队列的顺序执行,而非公平锁则是每一个新的节点都会去尝试获取锁,获取不成功在进入队列,获取成功则直接使用,因为线程的唤醒相对来说比较耗费时间,非公平锁在性能上优于公平锁,下面我们以非公平锁来切入，其中可能会穿插一些ReentrantLock的方法,但是并不影响
NonFairSync
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);
}
}
lock()
入口为lock()
ReentrantLock#NonFairSync#lock():
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
#AbstractQueuedSynchronizer#compareAndSetState():
protected final boolean compareAndSetState(int expect, int update) {
return U.compareAndSwapInt(this, STATE, expect, update);
}
private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
private static final long STATE;
private static final long HEAD;
private static final long TAIL;
static {
try {
STATE = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
HEAD = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
TAIL = U.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
} catch (ReflectiveOperationException e) {
throw new Error(e);
}
// Reduce the risk of rare disastrous classloading in first call to
// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
Class<?> ensureLoaded = LockSupport.class;
}
首先通过compareAndSetState中的STATE属性判断当前是否可以获取锁,如果值为0代表当前并无线程持有锁,可以获取锁,然后并将STATE设置为1,如果为1则代表当前锁已被其他线程获取
如果可以获取锁的话，则进入setExclusiveOwnerThread()
AbstractQueuedSynchronizer#setExclusiveOwnerThread():
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
exclusiveOwnerThread代表将持有锁的线程,将持有锁的线程对象指向当前线程
如果不可以获取锁的话，则进入acquire()
AbstractQueuedSynchronizer#acquire():
static final Node EXCLUSIVE = null;
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
这是一个判断体,我们先来看看判断体成立的情况:
AbstractOwnableSynchronizer#selfInterrupt():
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
中断当前线程,作用是中断线程,如果线程处于阻塞状态则会抛出一个InterruptedException， 如果是一个正常状态的线程,则不起任何的作用,但是会将当前线程的状态修改为中断停止状态
来看判读条件，
- 第一个条件
最后会通过NonFairSync调用Sync的nonfairTryAcquire():
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;
}
首先获取状态，如果当前状态为0的话,则设置状态为1,并设置持有锁的线程为当前线程,返回true;如果当前状态不为0,但是持有持有锁的线程与当前线程相同,则状态值加1，返回true,否则返回false. 总的来说这个方法的作用就是判断当前线程能否获取到锁,可以就返回ture,不可以就返回false
- 第二个条件
先看addWaiter()
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (;;) {
Node oldTail = tail;
if (oldTail != null) {
U.putObject(node, Node.PREV, oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
return node;
}
} else {
initializeSyncQueue();
}
}
}
这里的入参node可以暂时理解为空，我们知道AbstrantQueuedSynchronizer维护这个以队列,所以这里的Node对象的创建则是设置新建Node的后继节点,接着是一个循环,目的是将新建节点加入到队列中,不过用到了CAS算法,保证同步性，最后返回尾节点,也就是新建节点，所以这个方法的作用就是将新建节点加入到队列中.
AbstrantQueuedSynchronizer#Node#Node()
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
U.putObject(this, THREAD, Thread.currentThread());
}
可以看到设置新建节点的后继节点为入参节点,就是null,然后将Node对象中的THREAD属性,修改为当前线程,也可以说关联,因为THREAD是long类型的,而Thread.currentThread()是Thread类型的
再来看acquireQueued():
AbstractQueuedSynchronizer#acquireQueued():
final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
// 这里会挂起线程，线程进入阻塞状态，直到前一节点唤醒（前一节点是头结点并且执行完毕之后）
parkAndCheckInterrupt())
interrupted = true;
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
通过前面的分析我们知道入参node为队列尾节点,里面是一个循环,首先获取尾节点的前一节点，先看第一个判断,如果尾节点的前一节点是头结点,并且新建节点的线程即当前线程可以获取到锁
,就将当前线程设置为头结点,并使原先的头结点脱离出来。再来看shouldParkAfterFailedAcquire()根据方法名可以简单的做个判断,当当前节点获取锁失败,是否应该将当前线程挂起阻塞
AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire():
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
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.
*/
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false;
}
如果当前节点的前任节点的状态为SINGAL，则表示当前节点应该阻塞
如果当前节点的前任节点的状态>0 即CANCELLED(因为状态值里面只有CANCELLED是大于0的),也可以理解为当前节点的前任节点已被取消,因此需要将前任节点从队列中脱离出去,遍历队列,重新组成双向队列最后返回false
如果前任节点的状态值 < 0，即处于等待状态,则修改其状态值为SINGAL，即当前任节点的线程被释放的时候,当前节点需要被释放的标识，并返回false
parkAndCheckInterrupt则是将当前线程挂起,并返回当前线程的中断状态
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
现将当前线程挂起,然后判断当前线程的中断状态,Thread.interrupted()返回的当前线程的中断状态，正常情况下都是false,而如果返回了true,等轮到该线程获取到锁的时候,就是先中断线程
回到上面的代码
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
大致捋一下这个方法的作用：
首先新建线程通过tryAcquire()尝试获取锁,如果获取到锁直接退出判断体,如果没有，则进行第二个判断；
现将当前线程作为一个节点通过addWaiter加入到等待队列之中,并返回当前节点,即队列尾节点,然后通过acquireQueued去轮询队列使当前节点不断判断,当当前节点的前任节点是头结点时并且当前节点可以获取锁的时候,设置当前节点为头结点,此时当前节点获取到锁，所以这个方法的作用就是轮询队列中的当前节点(注意不是线程)不断的去尝试获取锁,直到达到能获取锁的要求条件为止
上面是线程获取锁的过程,如果获取成功则直接占用锁,如果失败,则线程被挂起,等到被唤醒
unLock
线程被挂起发生在lock()里面,而唤醒则在unLock()里面
ReentrantLock#unlock():
public void unlock() {
sync.release(1);
}
AbstractQueuedSynchronizer#release():
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
先看外层的判断条件
ReentrantLock#Sync#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;
}
获取到当前线程的STATE值,即,当前线程持有的锁数量,减去释放的锁的数量(即1),首先要判断当前线程跟持有锁的线程是否是同一个线程,如果不是则说明要释放锁的线程跟持有锁的线程不一致,抛出异常
根据当前线程在释放一个锁之后持有的锁数量判断,当前线程持有的锁数量为0的话,设置当前持有锁的线程标识为空,并设置当前锁的状态为空闲状态，即setState(0)
所以这个方法的意思就是等到一个线程要释放锁时,先判断当前线程持有这个锁的数量,如果当前线程只持有一个锁,则会去唤醒线程队列中的其他线程(这个下面会说到),如果持有的不止一把锁的话,则只是将当前线程持有的锁的数量减一,但不会真正的释放锁。 回到上面,如果当前线程只持有一把锁的话
获取到头结点,即当前线程所在的节点,判断头结点不为空,并且没有被释放(waitStatus = 0代表当前线程已被释放)，则去释放当前线程的后继线程.
AbstractQueuedSynchronizer#unparkSuccessor():
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
node.compareAndSetWaitStatus(ws, 0);
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
首先获取到当前线程的状态,确保没有被释放(主要是判断不为0),则设置当前节点的状态为0,代表当前线程已释放锁,然后获取到后继节点，如果后继节点为空或者已被取消或释放,则从队列尾部遍历,直到找到靠近队列头部并且没有被取消或释放的节点,以此节点作为下一节点,然后通过LockSupport.unpark(s.thread)唤醒此节点所代表的线程,使其获取锁
ReentrantLock#tryLock():
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
这个方法的参数是一个时间值,方法的作用是,在这个时间段内去等待获取线程,如果获取到话,则正常的走,如果没有获取到的话,就会取消这个线程继续等待,对应CANCELLED
AbstractQueuedSynchronizer#tryAcquireNanos():
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
首先会通过tryAcquire()尝试回去锁,如果没有后去成功,则进入doAcquireNanos()
AbstractQueuedSynchronizer#doAcquireNanos()
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.EXCLUSIVE);
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)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} catch (Throwable t) {
cancelAcquire(node);
throw t;
}
}
首先将当前线程加入等待队列,接着是一个循环,条件是此线程的前任线程是队列的头结点,或者是等待的时间到了和次线程中断了.线程中断和次线程的前任线程是头结点的情况这里就不多说了,来看等待的时间到了的情况
AbstractQueuedSynchronized#cancelAcquire():
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
// Skip cancelled predecessors
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;
// Can use unconditional write instead of CAS here.
// After this atomic step, other Nodes can skip past us.
// Before, we are free of interference from other threads.
node.waitStatus = Node.CANCELLED;
// If we are the tail, remove ourselves.
if (node == tail && compareAndSetTail(node, pred)) {
pred.compareAndSetNext(predNext, null);
} else {
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
if (next != null && next.waitStatus <= 0)
pred.compareAndSetNext(predNext, next);
} else {
unparkSuccessor(node);
}
node.next = node; // help GC
}
}
首先获取到此线程所代表的节点的前任节点和前任节点的后继节点,需要注意的是,如果前任节点已被释放则需要重新遍历队列获取到一个没有处于阻塞状态的线程节点；如果此线程位于队列的队尾,则设置前任节点的后继节点为空,如果不是位于队尾,下面就有两种处理情况,
- 如果前任节点不等于头结点并且前任节点的状态值等于
SIGNAL,则设置前任节点的后继节点指针指向当前节点的后继节点,即当前线程不再被阻塞，但同时也不会在等待锁,而是去处理其他的逻辑了 - 如果前任节点是头结点(主要是这个条件),那么当前节点也不会去获取锁,而是会将当前线程的后继节点唤醒,使后继节点去获取锁
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)
node.compareAndSetWaitStatus(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 p = tail; p != node && p != null; p = p.prev)
if (p.waitStatus <= 0)
s = p;
}
if (s != null)
LockSupport.unpark(s.thread);
}
网友评论