Java AQS条件队列Condition的await与signal实现

Java并发包中的AbstractQueuedSynchronizer（AQS）是构建锁和其他同步组件的基础框架。Condition是AQS的内部类，提供条件等待功能，类似于Object的wait/notify，但更灵活且功能更强大。本文将深入分析Condition的await和signal方法的实现机制，帮助读者理解线程间的协调原理。

AQS与Condition基础

AQS通过一个FIFO队列管理线程的获取和释放，而Condition则通过一个独立的等待队列实现条件等待。每个Condition实例对应一个等待队列，线程调用await方法会进入该队列等待，直到被signal或signalAll唤醒。

Condition的await方法实现

await方法使当前线程进入等待状态，释放持有的锁，并在被唤醒后重新获取锁。其核心流程包括：进入等待队列、释放锁、挂起线程、被唤醒后重新获取锁。

1. 源码分析

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    Node node = addWaiter(Node.CONDITION);
    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)
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

2. 流程步骤

1. 检查中断：如果当前线程已被中断，抛出InterruptedException。
2. 添加到等待队列：调用addWaiter将线程封装为Node节点，加入Condition的等待队列。
3. 完全释放锁：调用fullyRelease释放当前线程持有的锁，确保其他线程可以获取锁。
4. 进入自旋等待：通过isOnSyncQueue检查节点是否在同步队列（AQS的主队列）中。如果不在，说明线程仍在等待状态，调用LockSupport.park挂起线程。
5. 检查中断状态：在等待过程中，检查是否被中断，更新中断模式。
6. 重新获取锁：当线程被唤醒后，调用acquireQueued尝试重新获取锁。
7. 处理中断：根据中断模式处理中断情况，可能重新中断线程或抛出异常。

3. 状态转换流程

Condition的signal方法实现

signal方法从等待队列中移除一个节点，并将其转移到同步队列（AQS主队列），使线程可以尝试获取锁。

1. 源码分析

public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

2. 流程步骤

1. 检查锁持有状态：确保当前线程持有独占锁，否则抛出IllegalMonitorStateException。
2. 获取等待队列首节点：获取Condition等待队列的第一个节点。
3. 唤醒节点：调用doSignal将首节点从等待队列移除，并转移到同步队列。

3. doSignal方法细节

private void doSignal(Node first) {
    do {
        if ((firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) && (first = firstWaiter) != null);
}

• 移除节点：将首节点从等待队列中移除。
• 转移节点：调用transferForSignal将节点添加到同步队列，并唤醒线程。

4. 转移到同步队列流程

示例：生产者-消费者模型

通过示例代码展示Condition的await和signal的实际应用。

1. 代码实现

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionExample {
    private final Lock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();
    private final Object[] items = new Object[10];
    private int putptr, takeptr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}

2. 代码说明

• put方法：生产者线程调用，当队列满时调用notFull.await()等待，否则添加元素并唤醒消费者线程。
• take方法：消费者线程调用，当队列空时调用notEmpty.await()等待，否则取出元素并唤醒生产者线程。

注意事项

1. 虚假唤醒：必须使用while循环检查条件，而不是if，因为线程可能被意外唤醒。
2. 锁释放：await会自动释放锁，signal不会释放锁，需手动解锁。
3. 中断处理：await方法会响应中断，需正确处理中断异常。

总结

Condition的await和signal方法通过独立的等待队列实现线程间的条件等待和唤醒，结合AQS的同步队列，提供了高效且灵活的线程协调机制。理解其实现原理有助于编写正确的并发程序。