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。

最后修改:2018 年 05 月 13 日
如果觉得我的文章对你有用,请随意赞赏