详解Java并发包基石AQS

我会根据以下结构给出一份详细讲解“详解Java并发包基石AQS”的完整攻略。

什么是AQS

AQS是AbstractQueuedSynchronizer的缩写，翻译成中文可以叫做“抽象队列同步器”。它是java.util.concurrent包中的核心组成部分，也是各种同步组件（如ReentrantLock、Semaphore、CountDownLatch等）的基础。它提供了一种实现同步的可扩展的框架，并且大多数情况下不需要使用者直接使用AQS的核心方法，只需要继承它，然后实现一些简单的钩子方法，即可得到基本的同步功能。

AQS原理

AQS的核心回调方法是tryAcquire(int)和tryRelease(int)，它们分别会在AQS需要获取同步状态和释放同步状态的时候被调用。这两个方法的实现非常简单，只需要加几行CAS操作即可，如下所示：

protected boolean tryAcquire(int arg) {
   if (compareAndSetState(0, arg)) {
      setExclusiveOwnerThread(Thread.currentThread());
      return true;
   }
   return false;
}

protected boolean tryRelease(int arg) {
   if (getState() == 0) throw new IllegalMonitorStateException();
   setExclusiveOwnerThread(null);
   setState(0);
   return true;
}

其中getState()和setState(int)方法是操作同步状态的方法，compareAndSetState(int, int)利用了Java中的CAS原子操作，尝试将同步状态从0改为传入的arg。如果同步状态为0，那么当前线程会获取同步状态并且成为同步状态的拥有者；如果同步状态不为0，那么当前线程尝试获取同步状态失败，并且需要加入同步队列中排队等待。

同步队列中的节点是继承自AbstractQueuedSynchronizer.Node的类，它包含了一个元素，主要用于表示在等待队列中处于节点的锁状态。包含如下字段：

volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;

其中，waitStatus表示节点的状态，prev和next表示前驱节点和后继节点，thread表示当前节点代表的线程，nextWaiter表示都是在等待队列中节点的链表结构。等待队列是一个FIFO的队列，被锁住的线程可能被插入到队列的尾部，等待其他线程释放锁。

AQS使用示例

下面代码是一个简单的使用AQS实现的锁的例子。它首先通过继承AbstractQueuedSynchronizer类并实现tryAcquire()和tryRelease()方法，得到一个自定义的同步组件，然后通过这个同步组件来实现一个基于排它锁的缓存：

public class CustomLock extends AbstractQueuedSynchronizer {
    public CustomLock() {
        super();
    }

    @Override
    protected boolean tryAcquire(int arg) {
        if (compareAndSetState(0, arg)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    @Override
    protected boolean tryRelease(int arg) {
        if (getState() == 0) throw new IllegalMonitorStateException();
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock() {
        acquire(1);
    }

    public void unlock() {
        release(1);
    }

    private final HashMap<String, Object> cache = new HashMap<>();

    public Object getValue(String key) {
        lock();
        try {
            return cache.get(key);
        } finally {
            unlock();
        }
    }

    public void setValue(String key, Object value) {
        lock();
        try {
            cache.put(key, value);
        } finally {
            unlock();
        }
    }
}

在这个例子中，我们定义了一个CustomLock类，它继承自AbstractQueuedSynchronizer类，并实现了tryAcquire()和tryRelease()方法，最后定义了一个基于该锁的缓存。在getValue()和setValue()方法中，我们调用了lock()和unlock()方法来获取和释放锁。这样就可以保证在修改缓存的时候只有一个线程能够访问它。

下面给出另外一个简单的使用AQS的例子：

public class Semaphore {
    private final Sync sync;

    abstract static class Sync extends CustomLock {
        abstract void reducePermits(int reduction);
        abstract int tryAcquireShared(int acquires);
        abstract boolean tryReleaseShared(int releases);

        Sync(int permits) {
            setState(permits);
        }
    }

    private static final class NonfairSync extends Sync {
        NonfairSync(int permits) {
            super(permits);
        }

        @Override
        int tryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 || compareAndSetState(available, remaining)) {
                    return remaining;
                }
            }
        }

        @Override
        boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (next < current)
                    throw new Error("Maximum permit count exceeded");
                if (compareAndSetState(current, next))
                    return true;
            }
        }

        @Override
        void reducePermits(int reduction) {
            for (;;) {
                int current = getState();
                int next = current - reduction;
                if (next > current)
                    throw new IllegalArgumentException("Permit count underflow");
                if (compareAndSetState(current, next))
                    return;
            }
        }
    }

    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public void release() {
        sync.releaseShared(1);
    }
}

在这个例子中，我们定义了一个简单的信号量（Semaphore），它包含了一个内部类Sync，它继承自上面的CustomLock类，同时包含了三个抽象方法reducePermits(int)、tryAcquireShared(int)和tryReleaseShared(int)。在NonfairSync类中，我们实现了这三个抽象方法，来实现信号量的功能。在Semaphore类的acquire()和release()方法中，我们调用了acquireSharedInterruptibly(int)和releaseShared(int)方法，来获取和释放信号量。

这些示例说明了如何通过继承AQS类，并实现一些简单的钩子方法，来实现基本的同步和语义。这些同步组件包括锁、读/写锁、信号量、倒计时锁等，都是用AQS框架来实现的。在使用这些组件的时候，我们一般不需要直接和AQS的方法打交道，只需要调用这些同步组件的公开API即可。