Condition

Condition

Condition是一种多线程通信工具，表示多线程下参与数据竞争的线程的一种状态，主要负责多线程环境下对线程的挂起和唤醒工作。

方法

// 造成当前线程在接到信号或被中断之前一直处于等待状态。

void await() throws InterruptedException;

// 造成当前线程在接到信号之前一直处于等待状态。【注意：该方法对中断不敏感】。

void awaitUninterruptibly();

// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。

//返回值表示剩余时间，如果在`nanosTimeout` 之前唤醒，那么返回值 `= nanosTimeout - 消耗时间` ，

//如果返回值 `<= 0` ,则可以认定它已经超时了

long awaitNanos(long nanosTimeout) throws InterruptedException;

// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。

boolean await(long time, TimeUnit unit) throws InterruptedException;

// 造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。

//如果没有到指定时间就被通知，则返回 true ，否则表示到了指定时间，返回返回 false 。

boolean awaitUntil(Date deadline) throws InterruptedException;

// 唤醒一个等待线程。该线程从等待方法返回前必须获得与Condition相关的锁。

void signal();

// 唤醒所有等待线程。能够从等待方法返回的线程必须获得与Condition相关的锁。

void signalAll();

原理

Condition 内部维护一个条件队列，在获取锁的情况下，线程调用 await，线程会被放置在条件队列中并被阻塞。直到调用 signal、signalAll 唤醒线程，此后线程唤醒，会放入到 AQS 的同步队列，参与争抢锁资源。

AQS是AbstractQueuedSynchronizer的简称，翻译过来就是抽象队列同步器.

await

调用condition.await()方法会使当前线程进入等待队列并释放锁，同时线程变为等待状态。当从await()方法返回时，一定是获得与condition相关联的锁。

此时AQS主要执行以下动作：

线程1把自己包装成节点，waitStatus设为CONDITION(-2)，追加到ConditionObject中的条件队列(每个ConditionObject有一个自己的条件队列)；

线程1释放锁，把state设置为0；

然后唤醒等待队列中head节点的下一个节点；

//await

public final void await() throws InterruptedException {

    // 1、线程如果中断，那么抛出异常

    if (Thread.interrupted())

        throw new InterruptedException();

    // 2、将当前线程包装成为一个Node节点，加入FIFO队列中

    Node node = addConditionWaiter();

    // 3、释放锁

    int savedState = fullyRelease(node);

    int interruptMode = 0;

    // 4、判断节点是否在同步队列（注意非Condition队列）中，如果没有，则挂起当前线程，因为该线程尚未具备数据竞争资格

    while (!isOnSyncQueue(node)) {

        // 5、挂起线程

        LockSupport.park(this);

        // 6、中断直接返回

        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)

            break;

    }

    // 7、参与数据竞争（非中断时执行）

    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)

        interruptMode = REINTERRUPT;

    // 清理条件队列中状态为cancelled的节点

    if (node.nextWaiter != null) // clean up if cancelled

        unlinkCancelledWaiters();

    if (interruptMode != 0)

reportInterruptAfterWait(interruptMode);

    }

signal

另一个线程执行了 condition1.signal之后，主要是做了以下事情：

把条件队列中的第一个节点追加到等待队列中；

把等待队列原来尾节点的waitStatus设置为SIGNAL。

然后继续处理自己的事情，自己的事情处理完成之后，会释放锁，唤醒等待队列中head节点的下一个节点线程进行工作。

 public final void signal() {

 // 调用signal的线程必须持有独占锁

            if (!isHeldExclusively())

                throw new IllegalMonitorStateException();

            Node first = firstWaiter;

            if (first != null)

                doSignal(first);

        }

 private void doSignal(Node first) {

            do {

    // 因为first马上就要被转移到同步队列了，所以将first.nextWaiter，作为新的firstWatier。

                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;

    // 调用 enq 添加到 同步队列的尾部

    Node p = enq(node);

    int ws = p.waitStatus;

    // node 的上一个节点 修改为 SIGNAL 这样后续就可以唤醒自己了

    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))

        LockSupport.unpark(node.thread);

    return true;

}

使用示例

Await signal