Java并发编程-AQS源码之条件队列-一一网

这是我参与8月更文挑战的第11天，活动详情查看：8月更文挑战

前言

前面介绍了CLH同步队列,但是AQS内部类还有一个 ConditionObject, 条件队列就是由ConditionObject 维护的队列.

Contition是一种广义上的条件队列，它利用await()和signal()为线程提供了一种更为灵活的等待/通知模式。

1. Condition接口

我们知道任意一个对象,都有一组监视器方法,定义在java.lang.Object类上,主要包括wait()、wait(long timeout)、notify()以及notifyAll()方法，这些方法与synchronized同步关键字配合可以实现等待/通知模式。

Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式.
但是这两者在使用方式以及功能特性上还是有差别的。

对比Object的监视器方法和Condition接口

1.1 Condition 条件队列使用

Condition必须要配合Lock一起使用，因为对共享状态变量的访问发生在多线程环境下。

public class ConditionUseTest {

    public static void main(String[] args) {
        Lock lock = new ReentrantLock();
        Condition condition = lock.newCondition();

        Thread thread1 = new Thread(()->{
            String name = Thread.currentThread().getName();
            lock.lock();
            System.out.println(name + "==>成功获取到锁" + lock);
            try {
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(name + "==>被唤醒");
            lock.unlock();
            System.out.println(name + "==>释放锁");

        }, "线程1");
        thread1.start();

        Thread thread2 = new Thread(()->{
            String name = Thread.currentThread().getName();
            lock.lock();
            System.out.println(name + "==>成功获取到锁" + lock);
            try {
                System.out.println("========== 线程1：条件队列await,没有被signal的时候会一直等着 ===========");
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(name + "==>通知等待队列的线程");
            condition.signal();
            lock.unlock();
            System.out.println(name + "==>释放锁");

        }, "线程2");
        thread2.start();
    }
}
复制代码

执行结果:

1.2 源码分析

AQS，Lock,Condition，ConditionObject之间的关系：

以ReentrantLock为例,
ConditionObject是AQS的内部类，实现了Condition接口，Lock中提供newCondition()方法，委托给内部AQS的实现Sync来创建ConditionObject对象，享受AQS对Condition的支持。

从上面的例子可以看出,条件队列的流程大概是这样
lock -> await -> signal -> unlock

waitStatus可以取五种状态：

初始化为0，啥也不表示，之后会被置signal。

1表示cancelled，取消当前线程对锁的争夺。

-1表示signal，表示当前节点释放锁后需要唤醒后面可被唤醒的节点。

–2表示condition，我们这篇的重点，表示当前节点在条件队列中。

-3表示propagate，表示释放共享资源的时候会向后传播释放其他共享节点。

1.2.1 等待队列

等待队列是一个FIFO的队列，在队列中的每个节点都包含了一个线程引用，该线程就是在Condition对象上等待的线程，如果一个线程调用了Condition.await()方法，那么该线程将会释放锁、构造成节点加入等待队列并进入等待状态。

Condition拥有首尾节点的引用，而新增节点只需要将原有的尾节点nextWaiter指向它，并且更新尾节点即可。

同步队列和等待队列之间的关系

Condition的实现是同步器的内部类，因此每个Condition实例都能够访问同步器提供的方法，相当于每个Condition都拥有所属同步器的引用

1.2.2 等待

await代码

public final void await() throws InterruptedException {
    // 1. 当前线程被中断,直接抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 2. 添加到条件队列中    
    Node node = addConditionWaiter();
    // 3. 释放同步资源,锁
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 4. 如果这个节点的线程不在同步队列中，说明该线程还不具备竞争锁的资格
    while (!isOnSyncQueue(node)) {
        // 挂起线程
        LockSupport.park(this);
        // 线程中断退出
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 退出循环，被唤醒之后，进入阻塞队列，等待获取锁 acquireQueued
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}
复制代码

Node addConditionWaiter() 添加到条件队列

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        // 遍历整个条件队列，将已取消的所有节点清除出列
        unlinkCancelledWaiters();
        // t 重新赋值
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // t == null 表示队列此时为空，初始化firstWaiter
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    // 将尾指针指向新建的node
    lastWaiter = node;
    return node;
}
复制代码

上面其实也就是一个单链表的操作.
void unlinkCancelledWaiters() 清除队列中已经取消等待的节点

private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    // trail这里表示取消节点的前驱节点
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;
        t = next;
    }
}
复制代码