AQS条件队列及中断机制
Condition条件队列
当我们进行线程间的通信时，可以使用ReetrantLock与Condition相结合，其中的await()和signal()方法进行线程间的阻塞与唤醒。我将详细的解释其中的机制。
ConditionObject
ConditionObject是实现条件队列的关键，每个ConditionObject对象都维护一个单独的条件等待对列。一个AQS中可以有多个条件队列，但是只有一个同步队列。
该博客的描述挺好的，有的部分是引用里面的。AQS解析与实战
条件队列与同步队列的联系
1） 调用了
await()方法的线程，会被加入到ConditionObject等待队列中，并且唤醒同步队列中head节点的下一个节点2） 线程在某个ConditionObject对象上调用
signal()方法后，等待队列中的firstWaiter会被加入到同步队列中，等待被唤醒。3）当线程调用
unLock()方法解锁后，同步队列中的head节点的下一个节点会被唤醒。
public class ConditionObject implements Condition, java.io.Serializable {
private transient Node firstWaiter;
private transient Node lastWaiter;
每个ConditionObject对应一个条件队列，它记录该队列的头节点和尾节点。
条件队列是单向的，而同步队列是双向的，会有前驱指针。
await()方法
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);
}
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;
}
整体的流程如下：
- 执行await()时，会新创建一个节点并放入到该条件队列尾部。
- 然后释放锁，并唤醒同步队列中的Head节点的后一个节点。
- 然后while循环，将该节点阻塞，直达该节点被放入到同步节点或者被中断了，才退出循环。
- 退出循环后，开始调用
acquireQueued()不断尝试拿锁。 - 拿到锁后，会清空条件队列中被取消的节点。
public final void signal() {
//如果当前线程不是持有该锁的线程，抛出异常
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
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;
}
}
}
}
主要过程：
将条件队列中的头节点放入到同步队列尾部，并获取它在同步队列中的前驱节点
如果他前驱节点的状态是取消状态或者设置设置前驱节点为Signal失败，则唤醒当前节点，
唤醒后会执行在await()方法阻塞后的代码，会进行不断尝试获取锁。
线程的中断
当我们想执行线程中断时，一般会调用interrupt方法，对于初学者可能以为线程会立即中断执行，但是如果线程处于运行状态，不会受影响，会继续执行下去。为什么会出现这种情况呢？
java的线程中断机制，主要是依靠中断标志位实现的，当调用interrupt方法时，会将中断标志位设置为ture，然后针对线程此时的不同线程状态，会有不同的处理。具体的处理情况可以阅读这篇文章: java中断详细介绍及其对各种线程状态的影响分析
执行中断后，当线程处于waiting状态会抛出中断异常并将清空中断标志位。
前面写了一篇AQS的文章，介绍了acqure()方法，最后一步调用了selfInterrupt()，这是当前线程进行中断。一般我们加锁会调动lock()方法，但是实际上还有一个方法lockInterruptibly()。我将通过ReetrantLock类介绍它俩的区别。
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
首先分析下lock()方法，首先如果state为0，那么通过CAS机制将state设置为1，然后将当前线程设置为锁的拥有者；其他情况则调用acquire(1)尝试加锁，具体的尝试加锁机制就不介绍了，在之前文章写过，它是一个自旋操作，如果获取失败，会将线程park挂起，进入阻塞状态，当其他线程执行完，会唤醒该线程，执行向下执行，这时会执行Thread.interrupted(),会判断当前线程是否被中断，如果该线程拿到锁，会执行selfInterrupt()方法，将当前线程进行中断，但是之后自己设计的逻辑中没有如果没有处理中断的代码，那么它会一直执行下去，或者将当前线程设置为waitiing状态，就会抛出中断异常，并消除中断标志。
public void lockInterruptibly() throws InterruptedException {
sync.acquireInterruptibly(1);
}
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
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();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
现在再介绍一下lockInterruptibly的机制。
首先它会判断当前线程是否中断，如果有，就马上抛出异常；没有中断，则开始尝试获得锁，获取失败，将该节点放入到同步队列中，并执行自旋操作：
1）如果该节点的前驱节点头节点，并且该节点获取到了锁，则设置当前节点是头节点，并返回
2）其他情况，就需要将它的前置节点设置为Siginal,并将当前节点所属的线程挂起，当重新唤醒后，如果发生中断，会直接抛出异常，并将该节点从同步队列中删除
