java条件队列—Condition

前言

 

Condition是java1.5在引入显示锁Lock的同时一起引入的，它是一个接口，定义了一些列的await方法和signal、signalAll方法。其作用与Object内置锁对象的wait、notify、notifyAll方法类似，用于在某种条件下对线程进行阻塞和唤醒，这些条件就构成了条件队列。只是Condition方法在java API中进行了显式的实现；而内置锁使用的Object类中的相关方法都是native方法，是由jvm实现的。

 

Condition一般都是配合一个显式锁Lock一起使用，Lock接口的方法中有一个newCondition()方法用于生成Condition对象。在java1.5以后常用的显式锁有ReentrantLock和ReentrantReadWriteLock，他们都是基于AQS实现的，而在AQS中有一个内部类ConditionObject实现了Condition接口。所谓条件队列，其实是一个单向链表；在讲解AQS的实现原理时只讲解了AQS队列，AQS队列前面讲过（点这里）是双向链表结构。也就是说在AQS整体实现中维护了两个链表：一个是“同步队列”双向链表(这里简称AQS队列)，另一个是“条件队列”单向链表。

 

这里以ReentrantLock为例先简单讲解下这两个队列的关系：通过ReentrantLock的lock方法，如果获取不到锁当前线程会进入AQS队列阻塞；被唤醒后继续获取锁，如果获取到锁，移出AQS队列，继续执行；遇到Condition的await方法，加入“条件队列”，阻塞线程；被其他现象的signal方法唤醒，从“条件队列”中删除，并加入到AQS队列，如果获取到锁就继续执行。可以看到上述操作，线程节点（Node）其实在两个队列之间切换，由于“条件队列”在被唤醒时 都是从头开始遍历，所以只需要使用单向链表实现即可。在探究原理之前，先来看看Condition的相关方法、以及基本使用方式。

 

Condition的主要方法

 

Condition接口中一共定义了5个await方法，一个signal方法，一个signalAll方法。

 

5个await方法：

1、最基本的await方法： void await() throws InterruptedException; 使用这个方法必须在一个显式锁的lock和unlock包围的代码块之间；调用该方法后，当前线程会释放锁并被阻塞，直到其他线程通过调用同一个Condition对象的signal或者signalAll方法，再次被唤醒（唤醒后继续抢锁）。该方法会抛出InterruptedException异常，也就是说是可中断的（内置锁使用Object对象的三个wait方法也是可中断的）。

2、不可中断的await方法：void awaitUninterruptibly();该方法与await方法的作用相同，区别就是awaitUninterruptibly是不可中断的。也就是说，只能通过其他线程调用同一个Condition对象的signal或者signalAll方法，才能被唤醒。

3、延时wait方法（返回long）：long awaitNanos(long nanosTimeout) throws InterruptedException;这个方法基本作用与await方法相同，区别就是通过awaitNanos方法阻塞的线程，如果在指定的时间内还没有被signal或者signalAll方法唤醒，则会阻塞指定时间后自动取消阻塞，并返回；返回值 nanostimeout 值减去花费在等待此方法的返回结果的时间的估算值，如果返回值如果小于等于0说明超时（在指定时间内没有被signal或者signalAll方法唤醒）。

4、延时wait方法（返回boolean）：boolean await(long time, TimeUnit unit) throws InterruptedException;这个方法与awaitNanos作用完全一样，区别只在返回值。如果返回true，相当于awaitNanos返回大于0，如果返回false相当于awaitNanos返回小于等于0。

5、延时到指定时间wait方法: oolean awaitUntil(Date deadline) throws InterruptedException;这个方法与第4个方法作用相同，只是参数有区别，这个方法依赖服务器的时钟。

这5个方法除了第二以外，其他的都是支持中断的。

 

signal方法：唤醒条件队列中的1个线程（await时间最长的线程）。

signalAll方法：唤醒条件队列中所有的线程，去竞争。

 

可见使用signal方法的性能肯定会好些，但有可能有些线程会被忘了唤醒（延时await可以解决这个问题）。signalAll方法性能差些，但能保证所有的线程最终都会被唤醒，使用方便。这两个方法可以根据具体业务情况使用。

 

ConditionObject的实现原理

 

AQS的内部类ConditionObject对Condition的所有接口方法进行了实现，5个await方法的核心实现基本相同，这里只对第一个await方法进行分析。另外在对signal、signalAll方法的实现原理进行分析。在对这三个方法进行分析之前，首先来看下条件队列，在ConditionObject中定了条件队列的第一个节点和最后一个节点：

/** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;

在Node中还有一个指向下一个节点的指针：nextWaiter，这就构成了一个单向链表的“条件队列”。

 

await方法

public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        Node node = addConditionWaiter();//加入条件队列，注意不是AQS队列
        long savedState = fullyRelease(node);//释放当前线程占用的排它锁
        int interruptMode = 0;
        while (!isOnSyncQueue(node)) {//判断当前节点是否在AQS队列中，如果不在就进行阻塞
            LockSupport.park(this);//阻塞等待signal
            //判断中断标记在阻塞等待期间 是否改变
            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);
    }
 

 

这里的重点就是如何跳出while循环，一般有两种情况：被外部调用interrupt方法中断，这时会在break处跳出；在Condition上调用signal，这时会把该节点从“条件队列”移到AQS队列，whlie循环继续调用isOnSyncQueue方法检查，这时当前线程已经在AQS队列中存在，跳出while循环。另外unlinkCancelledWaiters方法会清除已经清理已经处理过的节点，即从“条件队列”中移除。

 

signal方法

signal本质上就是把“条件队列”的第一个节点移除，并加入到“AQS队列”:

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;
        //添加到AQS队列
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);//唤醒线程，在await内部park阻塞的
        return true;
}

在transferForSignal方法中可以看到节点会被添加到“AQS队列”中，排队获取锁。如果取得锁，就可以执行执行，否则还是会在AQS队列中继续被阻塞。

 

signalAll方法

signalAll方法与signal的区别是，signalAll会遍历整个“条件队列”，唤醒所有线程加入到“AQS队列”中：

public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);//区别
        }
private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);//遍历到队列末尾
        }
 

 

至此三个核心方法分析完毕，其他await方法实现大同小异。

 

总结

 

AQS中本质上有两个队列，一个是AQS队列，一个是条件队列。AQS队列一般用于各种锁的实现，条件队列必须结合锁一起使用，通过await方法加入条件队列，通过signal方法可以把节点移动到“AQS队列”，并触发从条件队列中移除（在await方法返回前）。

 

 

最后简单提下,Thread.sleep与await（或Object的wait方法）的区别, sleep方法本质上不会放弃锁；而await会放弃锁，并在signal后，还需重新获得锁 才能继续执行。