ReentrantLock、Semaphore 和 CountDownLatch 的功能不同,却都可以借助 AQS 组织等待线程。AQS 的全称是 AbstractQueuedSynchronizer,子类负责定义资源状态,AQS 负责组织线程的排队和等待。
state 和队列
Java 8 的 AQS 用一个 volatile int 状态值表示资源状态,子类通过 getState、setState、compareAndSetState 访问它。
state 的含义由子类决定。例如 ReentrantLock 用它记录重入次数,Semaphore 用它记录许可数,CountDownLatch 用它记录剩余计数。具体怎么修改这个值,由各自的同步器实现。
获取资源失败时,线程可以进入同步队列。节点记录线程、前后关系和等待状态,队列协调何时需要挂起、何时可以被唤醒。具体阻塞和唤醒会用到 LockSupport。
一个只开一次的门
下面用共享模式实现一个简单的等待通知功能:调用 await() 的线程先等待,调用 open() 后,所有等待线程都可以继续执行:
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
public class OneShotGateDemo {
private static class Gate {
private static class Sync extends AbstractQueuedSynchronizer {
@Override
protected int tryAcquireShared(int ignored) {
return getState() == 1 ? 1 : -1;
}
@Override
protected boolean tryReleaseShared(int ignored) {
setState(1);
return true;
}
}
private final Sync sync = new Sync();
void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
void open() {
sync.releaseShared(1);
}
}
public static void main(String[] args) throws InterruptedException {
Gate gate = new Gate();
Thread worker = new Thread(() -> {
try {
gate.await();
System.out.println("通过");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
worker.start();
gate.open();
worker.join();
}
}输出 通过。无论 open 在 await 前还是后发生,都不会丢失“已经打开”这个状态。这里的状态只从 0 变成 1,重复 open 也是相同结果,所以不需要做计数型的 CAS 减一。
tryAcquireShared() 返回负数表示失败,零表示成功但不表示后继还能成功,正数表示成功并允许继续传播。这个门打开后所有人都能通过,所以返回正数。
独占与共享
独占模式通过 tryAcquire()、tryRelease() 定义获取和释放。共享模式则使用对应的 Shared 方法。子类负责快速尝试,AQS 负责失败后的入队、重试和等待,不应在 try 方法里自行无限阻塞。
排队并不自动保证公平。子类如果允许新线程在队列外直接获取资源,就仍然可以插队。公平策略需要在获取资源的逻辑中检查前驱等待者。
Condition 的队列
Condition 等待队列与同步队列不是一回事。线程 await 时先进入条件等待并释放锁;signal 后再转移到同步队列,重新竞争锁。
要使用 AQS 的 ConditionObject,独占同步器必须正确实现所有权判断、完全释放和重新获取等规则。上面的 Gate 只实现了共享模式下的等待和放行,没有实现这些独占锁的规则,因此没有提供 Condition。