抽象同步队列AQS

Posted 2021-03-10 cmg219

AQS是什么

AbstractQueuedSynchronizer抽象同步队列简称AQS，它是实现同步器的基础组件，并发包中锁的底层就是使用AQS实现的。
先看下一下AbstractQueuedSynchronizer的说明文档

/**
 * Provides a framework for implementing blocking locks and related
 * synchronizers (semaphores, events, etc) that rely on
 * first-in-first-out (FIFO) wait queues.  This class is designed to
 * be a useful basis for most kinds of synchronizers that rely on a
 * single atomic {@code int} value to represent state. Subclasses
 * must define the protected methods that change this state, and which
 * define what that state means in terms of this object being acquired
 * or released.  Given these, the other methods in this class carry
 * out all queuing and blocking mechanics. Subclasses can maintain
 * other state fields, but only the atomically updated {@code int}
 * value manipulated using methods {@link #getState}, {@link
 * #setState} and {@link #compareAndSetState} is tracked with respect
 * to synchronization.

通过官方文档可以知道它是用来构建锁或者其他同步组件的基础框架，它使用了一个int成员变量表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队工作。其中Node中的thread 变量用来存放进入AQS队列里面的线程，Node节点内部的SHARED用来标记该线程是获取共享资源时被阻塞。ConditionObject里维护的是调用条件变量的await()方法后被阻塞的线程。

AQS的用途

* <p>To use this class as the basis of a synchronizer, redefine the
 * following methods, as applicable, by inspecting and/or modifying
 * the synchronization state using {@link #getState}, {@link
 * #setState} and/or {@link #compareAndSetState}:
 *
 * <ul>
 * <li> {@link #tryAcquire}
 * <li> {@link #tryRelease}
 * <li> {@link #tryAcquireShared}
 * <li> {@link #tryReleaseShared}
 * <li> {@link #isHeldExclusively}
 * </ul>

可以通过重写以上的方法来完成一个自定义的锁。ReentrantLock,ReentrantReadWriteLock都是用AQS实现的。
为了便于理解，可以先用AQS实现一个自定义的锁独占锁，需要做的事是实现上面的部分方法。注意一下内容都是以独占锁来举例说明，共享锁的原理和次类似。

public class mutex implements Lock {
    private static  class Sync extends AbstractQueuedSynchronizer{
        //是否被占用
        protected  boolean isHeldExclusively(){
            return getState()==1;
        }
        //当状态为0时获取锁
        public boolean tryAcquire(int acquires){
            final Thread current = Thread.currentThread();
            //compareAndSetState是CSA操作
            if(compareAndSetState(0,1)){
                setExclusiveOwnerThread(current);
                return true;
            }
            return false;
        }
        protected  boolean tryRelease(int releases){
            //线程不持有锁而调用此方法会报异常
            if(getState()==0){
                throw new IllegalMonitorStateException();
            }
            setExclusiveOwnerThread(null);
            return true;

        }

        Condition newCondition() { return new ConditionObject();
    }
    //把操作代理到sync上即可
    private final  Sync sync=new Sync();
    public void lock() {
        sync.acquire(1);
    }

    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

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

    public Condition newCondition1() {
        return sync.newCondition();
    }
}

通过重写上面的方法可以完成一个自定义的锁。下面通过介绍各种函数来认识AQS。
在AQS中维持了一个单一的状态信息state，可以通过 getState() 、 setState()、compareAndSetState()函数修改其值 。对于ReentrantLock的实现来说，state 可以用来表示当前线程获取锁的可重入次数；对于读写锁 ReentrantReadWriteLock 来说，state的高16位表示读状态，也就是获取该读锁的次数，低 16位表示获取到写锁的线程的可重入次数；所以根据state的含义不同，而来实现不同的锁。
独占锁通过acquire(),release()来获取与释放资源。
newCondition()方法返回ConditionObject，用来结合锁实现线程同步。ConditionObject 可以直接访问AQS对象内部的变量 ，比如 state 状态值和 AQS 队列。每个条件变量对应 一个条件队列（单向链表队列），其用来存放调用条件变量的await 方法后被阻塞的线程。
结合下图，可以说明线程在获取锁的流程。
假设有线程A线程B，lock对象。A与B同时争取lock对象，A成功后，B会被放入AQS阻塞队列，等待。
1.当线程A区获取lock锁时，调用 lock.lock()方法，lock()方法中写有acquire(1)方法，而此方法会调用我们重写的tryAcquire(arg)方法尝试获取资源， 具体是设置状态变量 state 的值，成功则直接返回，失败则将当前线程封装为类型为 Node. EXCLUSIVE 的 Node 节点后插入到 AQS 阻塞队列的尾部，并调用LockSupport. park( this ） 方法挂起自己。

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

2.当线程A运行到一段时间，若使用条件变量的condition.await()方法，则会阻塞挂起当前线程，并被放入条件队列中。这时候，若B线程获取到了锁，并调用条件变量的signal 方法时，被阻塞的线程才会从await 处返回，然后加入AQS阻塞队列。
3.当线程B调用lock.unlock会释放锁，它会调用release（1）中tryRelease()arg,这个方法也是我们重写的。这里是设置状态变量 state 的值，然后调用LockSupport.unpark(thread ）方法激活 AQS 队列 里面被阻塞的一个线程（thread）被激活的线程则使用 tryAcquire尝试，看当前状态变量 state的值是否能满足自己的需要，满足则该线程被激活，然后继续向下运行，否则还是会被放入AQS队列并被挂起。

 public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }