AQS 主要有两种具体实现，共享锁和排他锁，排他锁简单点，我先讲共享锁
1、共享锁
关键方法 AbstractQueuedSynchronizer#acquireShared
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0) // 尝试获取arg个资源，AQS的state - arg > 0并且赋值成功以后, 返回一个剩余量，若大于等于0，则获取成功
doAcquireShared(arg); //要开始让请求的线程获取资源了
}
关键方法 AbstractQueuedSynchronizer#doAcquireShared
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
/** 其实这里蛮关键的，多个线程同时跑进来，然后会尝试唤醒头结点的后继节点的线程，然后线程才能进入代码的临界区，
*不过还要再详细讲一下 {@link setHeadAndPropagate#setHeadAndPropagate}的细节才好理解
*/
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
/**
*如果当前线程现在所在的节点的前继不是头节点，
*则无法获取锁(资源) 那么会把当前节点前面所有状态为CANCEL的节点去掉，
*循环往复，知道当前节点的前继节点状态为 SIGNAL为止
*然后会挂起当前线程如果不能唤醒的话，然后记录下中断状态啥的
*/
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
关键方法 AbstractQueuedSynchronizer#setHeadAndPropagate
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
/**以上是原作者的注释，接下来我讲下我自己的理解
* propagate 说白了就是剩余的资源数目(总资源数肯定要大于0，要不然怎么玩？)
* 如果propagate == 0 的时候，其实肯定有线程进入了 {@link doReleaseShared}
* 在propagate == 0 && h.waitStatus == 0 的时候，其实没有必要去唤醒后继节点，没有资源了，当前线程可以直接进入临界区，跳过就好了，如果
* propagate > 0 的话，有剩余资源，所以进去一下唤醒后继节点的比较好。。。
* h.waitStatus < 0 的话，如果为{@link Node.PROPAGATE}辣么，进入 {@link AbstractQueuedSynchronizer#doReleaseShared}
* 里面的for循环基本就直接出来了，如果是 {@link Node.SIGNAL}会唤醒后继的节点
* 所以嘛，还是要进去，就算是 PROPAGATE的状态，很快也会出来
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}
AbstractQueuedSynchronizer#doReleaseShared
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
/**
* 这里比较好理解，如果头结点是SIGNAL状态，唤醒后继节点
*/
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
/**
* 这里之所以要把状态先设置为0，就是为了上个方法
* {@link AbstractQueuedSynchronizer#setHeadAndPropagate}
* 中的状态判断，头结点状态为0的话，那么一定有个线程在这里运行，
* 会唤醒后继节点，并且至少有一个线程会把状态设置为 PROPAGATE
* 这样后面进来的线程在头结点不变的情况下，会很快的跑出这个for循环的
*/
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
共享锁的释放锁
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
比较简单，跟获取锁相比，只需要自定义这么具体地分配资源 tryReleaseShared 就好了
ConditionObject的队列
这个队列和和AQS的CLH队列有密切的联系，当需要await的时候我们添加一个节点到CondtionObject的队列末尾，然后阻塞住相应的线程，当需要signal释放的时候，transferForSignal会把ConditionObject的首个节点添加到AQS队列尾部，然后把倒数第二个节点(enq方法保证一定会有的)状态设置为SIGNAL，方便后续唤醒。
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
// 在这里，我们会阻塞线程
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*/
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
/*
* 这里返回倒是第二个节点的指针或者说是引用，如果状态是CANCEL或者
* 无法设置为SIGNAL，直接唤醒最后一个节点的线程，
* 这样一样可以把倒是第二个节点的状态设置为SIGNAL
* {@see AbstractQueuedSynchronizer#shouldParkAfterFailedAcquire}
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
网友评论