我意识中的AQS（AbstractQueuedSynchronizer）以及几种同步类（器）

曾经终败给现在 2022-12-10 11:26 119阅读 0赞

## 一 概述 ##

AQS(AbstractQueuedSynchronizer)是一个构建锁和同步器的框架，很多同步类（器）可以通过AQS容易且高效的构造出来。

通过AQS构造出来的常见同步类（器）包括以下几种：

1.  ReentrantLock
2.  CountDownLatch
3.  Sempahore
4.  ReentrantReadWriteLock

## 二 AQS(AbstractQueuedSynchronizer)的解析 ##

**AQS拥有三个核心部分：**

1.  state。
2.  控制线程抢锁和配合的FIFO队列。
3.  需要不同同步类（器）实现的获取/释放等方法。

**AQS核心部分的源码：**

public abstract class AbstractQueuedSynchronizer
        extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
    
        //保持内存可见性的state
        private volatile int state;
    
        //提供compareAndSapInt本地原子指令方法的Unsafe类
        private static final Unsafe unsafe = Unsafe.getUnsafe();
        
        protected final boolean compareAndSetState(int expect, int update) {
            //通过调用unsafe.compareAndSwapInt方法，借助指令的原子性保证方法的原子性
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }
    
        //自旋超时时间
        static final long spinForTimeoutThreshold = 1000L;
        
        //获取独占锁
        protected boolean tryAcquire(int arg) {
            throw new UnsupportedOperationException();
        }
        //释放独占锁
        protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }
        //获取共享锁
        protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
        //释放共享锁
        protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
    }

**同步类（器）会根据自己是否需要独占来选择重写AQS中的某些具体方法：**

**独占：**重写tryAcqurie()和tryRelease()

**共享：**重写tryAcquireShared()和tryReleaseShared()

## 三 AQS构造出的常见同步类（器） ##

由AQS构造出来的常见同步类（器）需要通过继承AQS的静态内部类Sync来满足ASQ的低三个部分即不同的同步类（器）需要实现自己获取/释放等方法。

**1.ReentrantLock**

public class ReentrantLock implements Lock, java.io.Serializable {
        private static final long serialVersionUID = 7373984872572414699L;
        /** Synchronizer providing all implementation mechanics */
        private final Sync sync;
    
        /**
         * Base of synchronization control for this lock. Subclassed
         * into fair and nonfair versions below. Uses AQS state to
         * represent the number of holds on the lock.
         */
        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = -5179523762034025860L;
    
            /**
             * Performs {@link Lock#lock}. The main reason for subclassing
             * is to allow fast path for nonfair version.
             */
            //定义锁定的接口
            abstract void lock();
    
            /**
             * Performs non-fair tryLock.  tryAcquire is implemented in
             * subclasses, but both need nonfair try for trylock method.
             */
            //非公平锁的实现
            final boolean nonfairTryAcquire(int acquires) {
                //获取当前线程
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (compareAndSetState(0, acquires)) {
                        //设置持有锁的线程为当前线程
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                //锁重入
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
    
            protected final boolean tryRelease(int releases) {
                //c表示当前线程重入的次数
                int c = getState() - releases;
                //判断当前线程是持有锁的线程，如果不是持有锁的线程就会抛出异常
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                //当可重入的次数大于0的时候，free会一直为false
                boolean free = false;
                //当前c为0时就会释放锁
                if (c == 0) {
                    free = true;
                    //将持有当前锁的线程设置为null,表示没有任何线程持有这把锁
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }
    
            //判断持有锁的线程是否为当前线程
            protected final boolean isHeldExclusively() {
                return getExclusiveOwnerThread() == Thread.currentThread();
            }
    
    
            //通过AbstractOwnableSynchronizer中的方法getExclusiveOwnerThread()返回当前持有锁的线程
            final Thread getOwner() {
                return getState() == 0 ? null : getExclusiveOwnerThread();
            }
    
            //获取重入锁的值，表示次数，当线程持有锁的时候返回重入的次数，否则返回0，即锁未被占有
            final int getHoldCount() {
                return isHeldExclusively() ? getState() : 0;
            }
    
            //判断锁是否被持有
            final boolean isLocked() {
                return getState() != 0;
            }
    
            /**
             * Reconstitutes the instance from a stream (that is, deserializes it).
             */
            private void readObject(java.io.ObjectInputStream s)
                throws java.io.IOException, ClassNotFoundException {
                s.defaultReadObject();
                setState(0); // reset to unlocked state
            }
        }
    }

**非公平锁**

/**
         * Sync object for non-fair locks
         */
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = 7316153563782823691L;
    
            /**
             * Performs lock.  Try immediate barge, backing up to normal
             * acquire on failure.
             */
            //实现非公平锁
            final void lock() {
                //进行CAS操作，表示当前线程持有锁为0，则会将当前线程设置为持有锁
                if (compareAndSetState(0, 1))
                    //使得当前线程获取锁
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    //非公平锁初始化为1，即同一非公平锁只能由单线程获取
                    acquire(1);
            }
            
            //非公平锁
            protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);
            }
        }
    
        //AbstractQueuedSynchronizer中acquire(方法)
        public final void acquire(int arg) {
            //调用ReentrantLock.Sync.NonfariSync中的tryAcquire(arg)方法，
            //EXCLUSIVE表示的当前获取的为互斥锁
            //addWaiter则为将当前线程加入等待队列中
            //acquireQueued必要时会先等待，如果存在获取锁的机会就会尝试获取锁
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    
        //ReentrantLock.Sync.NonfariSync中的tryAcquire(arg)方法
        protected final boolean tryAcquire(int acquires) {
                return nonfairTryAcquire(acquires);
            }

**公平锁**

/**
         * Sync object for fair locks
         */
        static final class FairSync extends Sync {
            private static final long serialVersionUID = -3000897897090466540L;
            
            //实现公平锁，且初始化为1，即同一锁只能由单线程获取
            final void lock() {
                acquire(1);
            }
    
            /**
             * Fair version of tryAcquire.  Don't grant access unless
             * recursive call or no waiters or is first.
             */
            protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (!hasQueuedPredecessors() &&
                        //CAS自旋操作
                        compareAndSetState(0, acquires)) {
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                return false;
            }
        }

**初始化锁类型**

//默认为非公平锁
    public ReentrantLock() {
            sync = new NonfairSync();
        }
    //通过fair字段设置为true，构造公平锁
    public ReentrantLock(boolean fair) {
            sync = fair ? new FairSync() : new NonfairSync();
        }

**加锁**

/*当前线程请求锁，如果该锁未被其他线程获取，则为当前线程加锁（count设置为1），
    如果当前下称已经持有了该锁，则将count进行加1，请求几次就代表重入几次，则会增加几。*/
    public void lock() {
            sync.lock();
        }
    
    public boolean tryLock() {
            return sync.nonfairTryAcquire(1);
        }
    
    public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }

加锁时会先判断当前线程是否已经持有锁，如果已经持有就会增加重入次数state，如果没有持有锁，而锁又被其他线程锁持有，则当前线程进入阻塞队列进行等待，然后等待机会进行重试获取锁的操作。

**解锁**

//ReentrantLock
        public void unlock() {
            sync.release(1);
        }
    
        //AbstractQueuedSynchronizer
        public final boolean release(int arg) {
            //Sync中重写的tryRelease(arg)返回true时，表示没有线程持有当前锁即锁被释放，如果返回false就会跳过下面代码
            if (tryRelease(arg)) {
                //锁被释放之后，就会从后面的线程队列中唤醒等待线程获取该锁
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    //将后面的队列结点中的线程唤醒
                    unparkSuccessor(h);
                return true;
            }
            //当可重入锁的数值不为零，直接退出，不会将后面的队列结点中的线程唤醒
            return false;
        }
    
        //AbstractQueuedSynchronizer中的原始tryRelease(arg)方法
        protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }
    
        //ReentrantLock中Sync中重写的tryRelease(arg)方法
        protected final boolean tryRelease(int releases) {
            //c表示当前线程重入的次数，releases表示减少的次数，默认我1
            int c = getState() - releases;
            //判断当前线程是持有锁的线程，如果不是持有锁的线程就会抛出异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            //当可重入的次数大于0的时候，free会一直为false
            boolean free = false;
            //当前c为0时就会释放锁
            if (c == 0) {
                free = true;
                //将持有当前锁的线程设置为null,表示没有任何线程持有这把锁
                setExclusiveOwnerThread(null);
            }
            setState(c);
            return free;
        }

由于ReentrantLock为重入锁，所以AQS中的state表示的是重入的次数，当持有锁的线程释放锁后，先判断是否为当前持有锁的线程释放的，如果不是则会抛出异常，如果是，则重入次数减一，当重入次数减到0的时候，就说明锁被完全释放，于是free为true，并发state设置为0，从而表明当前锁未被线程锁持有，则会由其他线程来进行竞争。

**等待中断方法**

public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
    
      public final void acquireInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (!tryAcquire(arg))
                doAcquireInterruptibly(arg);
        }
    
      private void doAcquireInterruptibly(int arg)
            throws InterruptedException {
            final Node node = addWaiter(Node.EXCLUSIVE);
            boolean failed = true;
            try {
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    
      private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }
    
      public static void park(Object blocker) {
            Thread t = Thread.currentThread();
            setBlocker(t, blocker);
            //park为native方法，使得当线程成阻塞，并处于挂起状态
            UNSAFE.park(false, 0L);
            setBlocker(t, null);
        }

**锁绑定多个条件**

final ConditionObject newCondition() {
                return new ConditionObject();
            }
    
        public class ConditionObject implements Condition, java.io.Serializable {
            private static final long serialVersionUID = 1173984872572414699L;
            /** First node of condition queue. */
            private transient Node firstWaiter;
            /** Last node of condition queue. */
            private transient Node lastWaiter;
    
            /**
             * Creates a new {@code ConditionObject} instance.
             */
            public ConditionObject() { }
        }

**2. CountDownLatch**

CountDownLatch能够使得某一线程在等待其他线程完成各自工作之后，再继续执行。使用一个计数器(AQS中的state)进行实现，计数器初始值为当前线程需要等待的其他线程的数量。当每个线程完成自己的任务后，计算器就会减一。当计数器的值为0的时候，表示其他线程都已经完成自己的那些任务，然后在CountDownLatch上等待的线程会执行自己的任务。

public class CountDownLatch {
        /**
         * Synchronization control For CountDownLatch.
         * Uses AQS state to represent count.
         */
        private static final class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) {
                //会将AQS内部类的方法setState设置成我们期望的值
                setState(count);
            }
    
            int getCount() {
                return getState();
            }
            //设置为共享的
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
            //设置为共享的
            protected boolean tryReleaseShared(int releases) {
                //通过for无限循环进行CAS的自旋，将state减一
                // Decrement count; signal when transition to zero
                for (;;) {
                    int c = getState();
                    //当count为0时
                    if (c == 0)
                        //不需要释放
                        return false;
                    //当count不为0时，nextc为count减一之后的结果
                    int nextc = c-1;
                    //通过CAS方法在多线程的情况下将当前count更新为nextc
                    //判断c是否为刚开始的值，如果时就通过CAS操作将c更新为nextc
                    //如果不是c说明存在线程修改了c,就会重新进入for循环
                    //直到cas（CompareAndSweap）操作的c为刚开始的值
                    if (compareAndSetState(c, nextc))
                        //如果c等于1，则nextc为0，返回true
                        return nextc == 0;
                }
            }
        }
        //获取count的值
        public long getCount() {
            return sync.getCount();
        }
    }

**初始化CountDownLatch对象，初始化count倒数的数值**

//count为倒数的数量
        public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }
        
        Sync(int count) {
            //会将AQS内部类的方法setState设置成我们期望的值
            setState(count);
           }
        
        //AbstractQueuedSynchronizer中的setState方法
        protected final void setState(int newState) {
            state = newState;
        }

await()方法被调用时会尝试获取"共享锁"，刚开始获取不到锁的时候，线程处于阻塞状态，await同时为使CountDownLatch上的线程等待方法，直到count的数量为0，即需要等待另外执行任务的线程数为0时，CountDownLatch上的线程获取"共享锁"才会进行对应的任务处理。

//等待直到倒数结束
       public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
       public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            //如果条件不成立就表示不同等待，直接放行
            if (tryAcquireShared(arg) < 0)
                //让当前线程进入等待队列，并将其阻塞
                doAcquireSharedInterruptibly(arg);
        }
    
       //CountDownLatch.Sync中重写的tryAcquireShared方法
       protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
       //AbstractQueuedSynchronizer中的tryAcquireShared方法
       protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
        //将线程放入等待队列中，并进行阻塞
        private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            //将线程包装成Node结点，每个Node中存在一个线程
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head) {
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    
        private final boolean parkAndCheckInterrupt() {
            //线程挂起，并进入阻塞状态
            LockSupport.park(this);
            return Thread.interrupted();
        }
    
        public static void park(Object blocker) {
            Thread t = Thread.currentThread();
            setBlocker(t, blocker);
            //park为native方法，使得当线程挂起，并处于阻塞状态
            UNSAFE.park(false, 0L);
            setBlocker(t, null);
        }

"锁计数器"的初始值为count，每个线程使用了CountDownLatch对象的countDown()方法时会将"锁计数器"count进行减一的操作，使得CountDownLatch上线程等待的其他线程数在减少，当count的线程调用countDown()方法之后，"锁计数器"才为0，而等待获取共享锁的线程才能继续运行。

public void countDown() {
            sync.releaseShared(1);
        }
    
        public final boolean releaseShared(int arg) {
            //当tryReleaseShared(arg)返回true时
            if (tryReleaseShared(arg)) {
                //唤醒阻塞队列中所有等待的线程
                doReleaseShared();
                return true;
            }
            return false;
        }
    
        //唤醒阻塞队列中的所有等待线程
        private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }
    
        //AbstarctQueuedSychronizer中的tryReleaseShared方法
        protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
        //CountDownLatch.Sync.tryReleaseShared方法
        protected boolean tryReleaseShared(int releases) {
            //通过for循环进行CAS的自旋，将state减一
            // Decrement count; signal when transition to zero
            for (; ; ) {
                int c = getState();
                //当count为0时
                if (c == 0)
                    //不需要释放
                    return false;
                //当count不为0时，nextc为count减一之后的结果
                int nextc = c - 1;
                //通过CAS方法在多线程的情况下将当前count更新为nextc
                //判断c是否为刚开始的值，如果时就通过CAS操作将c更新为nextc
                //如果不是c说明存在线程修改了c,就会重新进入for循环
                //直到cas（CompareAndSweap）操作的c为刚开始的值
                if (compareAndSetState(c, nextc))
                    //如果c等于1，则nextc为0，返回true
                    return nextc == 0;
            }
        }
    }

**3. Semaphore**

Semaphore是一个用来维护线程的许可证的数量并提供同步机制的同步类（器）。使用Semaphore可以控制线程拥有的许可证数量，当许可证数量state大于线程需要的许可证数量（默认为1，可通过permits的参数进行设置）时无需等待，否则进入AQS中的阻塞队列，处于等待状态。

public class Semaphore implements java.io.Serializable {
        private static final long serialVersionUID = -3222578661600680210L;
        /** All mechanics via AbstractQueuedSynchronizer subclass */
        private final Sync sync;
    
        /**
         * Synchronization implementation for semaphore.  Uses AQS state
         * to represent permits. Subclassed into fair and nonfair
         * versions.
         */
        abstract static class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 1192457210091910933L;
    
            Sync(int permits) {
                setState(permits);
            }
    
            final int getPermits() {
                return getState();
            }
    
            final int nonfairTryAcquireShared(int acquires) {
                //一个for循环的死循环
                for (;;) {
                    //available表示当前许可证的数量
                    int available = getState();
                    //acquires表示线程需要获取的许可证数量
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        //当许可证书剩余数量小于0的时候，将线程加入等待队列中
                        return remaining;
                }
            }
    
            protected final boolean tryReleaseShared(int releases) {
                for (;;) {
                    int current = getState();
                    int next = current + releases;
                    if (next < current) // overflow
                        throw new Error("Maximum permit count exceeded");
                    if (compareAndSetState(current, next))
                        return true;
                }
            }
    
            final void reducePermits(int reductions) {
                for (;;) {
                    int current = getState();
                    int next = current - reductions;
                    if (next > current) // underflow
                        throw new Error("Permit count underflow");
                    if (compareAndSetState(current, next))
                        return;
                }
            }
    
            final int drainPermits() {
                for (;;) {
                    int current = getState();
                    if (current == 0 || compareAndSetState(current, 0))
                        return current;
                }
            }
        }
    }

**非公平版本**

/**
         * NonFair version
         */
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = -2694183684443567898L;
    
            NonfairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }
        }
    
            final int nonfairTryAcquireShared(int acquires) {
                //一个for循环的死循环
                for (;;) {
                    //available表示当前许可证的数量
                    int available = getState();
                    //acquires表示想要获取的许可证数量,通过remaining检查剩余许可证数量是否满足此次需求
                    // remaining>0表示满足，就会通过CompareAndSetState自旋来改变state状态，直到改变成功则返回正数
                    // reminging<0表示不满足，直接返回负数，表示失败，则则线程进入阻塞状态，进入阻塞队列进行排队
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        //当剩余许可证书数量小于0的时候，将线程加入等待队列中
                        return remaining;
                }
            }

**公平版本**

static final class FairSync extends Sync {
            private static final long serialVersionUID = 2014338818796000944L;
    
            FairSync(int permits) {
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
                for (;;) {
                    if (hasQueuedPredecessors())
                        return -1;
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
        }

**初始化Semaphore对象，并初始化对象持有的许可证数**

//默认为非公平的
    public Semaphore(int permits) {
            sync = new NonfairSync(permits);
        }
    //通过将fair参数设置为true来获取公平
    public Semaphore(int permits, boolean fair) {
            sync = fair ? new FairSync(permits) : new NonfairSync(permits);
        }

**获取许可证，许可证数量减少，使得后面的线程获取得不到满足的许可证而进入阻塞状态**

//获取许可证
        public void acquire() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
    
        public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                //让当前线程进入等待队列，并将其阻塞
                doAcquireSharedInterruptibly(arg);
        }
    
       protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
        
       //Semaphore中重写的非公平方法
       //Semaphore.Sync.NonfairSync.tryAcquiredShared
       protected int tryAcquireShared(int acquires) {
                return nonfairTryAcquireShared(acquires);
            }
    
       //Semaphore中重写的公平方法
       //Semaphore.Sync.FairSync.tryAcquiredShared
       protected int tryAcquireShared(int acquires) {
                for (;;) {
                    if (hasQueuedPredecessors())
                        return -1;
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                            compareAndSetState(available, remaining))
                        return remaining;
                }
            }
    
        //判断sync.nonfairTryAcquireShared
        public boolean tryAcquire() {
            return sync.nonfairTryAcquireShared(1) >= 0;
        }
    
        //Semaphore.Sync.nonfairTryAcquireShared
            final int nonfairTryAcquireShared(int acquires) {
                //一个for循环的死循环
                for (;;) {
                    //available表示当前许可证的数量
                    int available = getState();
                    //acquires表示想要获取的许可证数量,通过remaining检查剩余许可证数量是否满足此次需求
                    // remaining>0表示满足，就会通过CompareAndSetState自旋来改变state状态，直到改变成功则返回正数
                    // reminging<0表示不满足，直接返回负数，表示失败
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        //当许可证书数量小于0的时候，将线程加入等待队列中
                        return remaining;
                }
            }

在Semaphore中，state表示许可证的剩余数量，在tryAcquire方法，判断nonfairTryAcquirShared大于等于0时表示满足线程需求的许可证数。

在nonfairTryAcquiredShared方法中，先判断剩余许可证数量remaining是否大于或等于0，即是否满足线程需求许可证数量，当大于或等于0时，就会通过compareAndSetState方法进行自旋来改变state的值，直到修改成功就返回正数remaining。当小于0时，就会返回负数，表示失败。

**释放许可证，使得剩余的许可证数量增加，并唤醒阻塞队线程获取满足数量的许可证**

public void release() {
            sync.releaseShared(1);
        }
    
        public final boolean releaseShared(int arg) {
            //当tryReleaseShared(arg)返回true时，说明许可证数量增加
            if (tryReleaseShared(arg)) {
                //唤醒阻塞队列中所有等待的线程，获取许可证并执行相应的任务
                doReleaseShared();
                return true;
            }
            return false;
        }
    
        protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
        protected final boolean tryReleaseShared(int releases) {
             for (;;) {
                    //获取当前许可证数
                    int current = getState();
                    //当前许可证加上被释放的许可证数，默认数量为1
                    int next = current + releases;
                    if (next < current) // overflow
                        throw new Error("Maximum permit count exceeded");
                    if (compareAndSetState(current, next))
                        return true;
                }
            }
    
        private void doReleaseShared() {
            /*
             * Ensure that a release propagates, even if there are other
             * in-progress acquires/releases.  This proceeds in the usual
             * way of trying to unparkSuccessor of head if it needs
             * signal. But if it does not, status is set to PROPAGATE to
             * ensure that upon release, propagation continues.
             * Additionally, we must loop in case a new node is added
             * while we are doing this. Also, unlike other uses of
             * unparkSuccessor, we need to know if CAS to reset status
             * fails, if so rechecking.
             */
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

**4. ReentrantReadWriteLock**

ReadWriteLock接口表示存在两个锁，一个写入锁（排它锁，独占锁，独享锁，最多同时只有一个线程持有）和一个读取锁（共享锁，可以由多个线程同时持有），但在基于AQS实现的ReentrantWriteLock中，当个AQS子类将同时管理读取加锁和写入加锁。ReentrantReadWriteLock使用了一个16位的状态来表示写入锁的计数，并使用了另一个16位的状态来表示读入锁的计数。在读取锁上的操作加过你使用共享的获取方法于释放方法，在写入锁上的操作将使用独占的获取方法与释放方法。

AQS在内部维护一个等待线程队列，其中记录了某个线程请求的是独占访问还是共享访问。在ReentrantReadWriteLock中，当锁可用时，如果位于队列头部的线程执行写入锁，那么线程会得到该锁，如果位于队列头部的线程执行读取访问，那么队列中在第一个写入线程之前的所有线程都获取该锁。

这种机制不允许选择读取线程优先或写入线程优先策略，在某些读写锁实现中也采用了这种方法，因此，要么AQS的等待队列不能是一个FIFO队列，要么使用两个队列。然而，在实际中很少需要这么严格的排序策略。如果非公平版本的ReentrantReadWriteLock无法提供足够的获取行，那么公平版本的ReentrantReadWriteLock通常会提供令人满意的排序保证，并且能确保读取线程和写入线程不会发生饥饿问题。

**锁升级与插队策略：**

**升级**

ReentrantReadWriteLock是不允许读锁插队操作，在升降级的时候是只允许写锁将为读锁（降级）而不允许读锁升级为写锁（升级）。

**插队策略**

1. 在锁为公平锁的情况下是不允许插队的，在非公平状态下，当存在多个线程获取读锁后处于同时读取状态，当此时一个线程想要写入时，会因为无法获取锁而进入等待队列，如果增加一个新的读线程时，该线程是可以插队获取读锁，当存在多个读线程进入时，这样写锁会处于饥饿状态。

2.在非公平状态下，当存在多个线程获取读锁后处于同时读取状态，当此时一个线程想要写入时，会因为无法获取锁而进入等待队列，如果增加一个新的读线程时，直接将该线程放入线程队列中，等到写线程完成后才让其获取读锁，这样避免写锁处于饥饿状态。

ReentrantReadWriteLock采用了策略二，但是非公平锁写锁是可以随时插队的，可以插队但是不容易插入。读锁仅在等待队列头结点不是想获取写锁的线程的时候可以插队。

**非公平锁**

/**
         * Nonfair version of Sync
         */
        static final class NonfairSync extends Sync {
            private static final long serialVersionUID = -8159625535654395037L;
            //获取写锁时可以直接尝试进行插队操作
            final boolean writerShouldBlock() {
                return false; // writers can always barge
            }
    
            final boolean readerShouldBlock() {
                /* As a heuristic to avoid indefinite writer starvation,
                 * block if the thread that momentarily appears to be head
                 * of queue, if one exists, is a waiting writer.  This is
                 * only a probabilistic effect since a new reader will not
                 * block if there is a waiting writer behind other enabled
                 * readers that have not yet drained from the queue.
                 */
                //获取读锁时先判断当前线程之前是否存在获取写锁线程，如果存在写锁就不能进行插队，否则可以插队
                //从方法名（队列中断第一个线程是否获取排它锁即写锁）
                return apparentlyFirstQueuedIsExclusive();
            }
        }
    
        //AQS中的判断获取读锁的线程前面是否存在获取写锁的线程
        final boolean apparentlyFirstQueuedIsExclusive() {
            Node h, s;
            return (h = head) != null &&
                (s = h.next)  != null &&
                !s.isShared()         &&
                s.thread != null;
        }

**公平锁**

/**
         * Fair version of Sync 公平锁是不存在插队
         */
        static final class FairSync extends Sync {
            private static final long serialVersionUID = -2274990926593161451L;
            final boolean writerShouldBlock() {
                //判断队列前面是否存在排队线程，如果存在排队线程就会使得当前读线程排队
                return hasQueuedPredecessors();
            }
            final boolean readerShouldBlock() {
                //判断队列前面是否存在排队线程，如果存在排队线程就会使得当前写线程排队
                return hasQueuedPredecessors();
            }
        }

**获取读锁**

protected final int tryAcquireShared(int unused) {
                /*
                 * Walkthrough:
                 * 1. If write lock held by another thread, fail.
                 * 2. Otherwise, this thread is eligible for
                 *    lock wrt state, so ask if it should block
                 *    because of queue policy. If not, try
                 *    to grant by CASing state and updating count.
                 *    Note that step does not check for reentrant
                 *    acquires, which is postponed to full version
                 *    to avoid having to check hold count in
                 *    the more typical non-reentrant case.
                 * 3. If step 2 fails either because thread
                 *    apparently not eligible or CAS fails or count
                 *    saturated, chain to version with full retry loop.
                 */
                Thread current = Thread.currentThread();
                int c = getState();
                if (exclusiveCount(c) != 0 &&
                    getExclusiveOwnerThread() != current)
                    return -1;
                int r = sharedCount(c);
                if (!readerShouldBlock() &&
                    r < MAX_COUNT &&
                    compareAndSetState(c, c + SHARED_UNIT)) {
                    if (r == 0) {
                        firstReader = current;
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {
                        firstReaderHoldCount++;
                    } else {
                        HoldCounter rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current))
                            cachedHoldCounter = rh = readHolds.get();
                        else if (rh.count == 0)
                            readHolds.set(rh);
                        rh.count++;
                    }
                    return 1;
                }
                return fullTryAcquireShared(current);
            }

**获取写锁**

protected final boolean tryAcquire(int acquires) {
                /*
                 * Walkthrough:
                 * 1. If read count nonzero or write count nonzero
                 *    and owner is a different thread, fail.
                 * 2. If count would saturate, fail. (This can only
                 *    happen if count is already nonzero.)
                 * 3. Otherwise, this thread is eligible for lock if
                 *    it is either a reentrant acquire or
                 *    queue policy allows it. If so, update state
                 *    and set owner.
                 */
                Thread current = Thread.currentThread();
                int c = getState();
                int w = exclusiveCount(c);
                if (c != 0) {
                    // (Note: if c != 0 and w == 0 then shared count != 0)
                    if (w == 0 || current != getExclusiveOwnerThread())
                        return false;
                    if (w + exclusiveCount(acquires) > MAX_COUNT)
                        throw new Error("Maximum lock count exceeded");
                    // Reentrant acquire
                    setState(c + acquires);
                    return true;
                }
                if (writerShouldBlock() ||
                    !compareAndSetState(c, c + acquires))
                    return false;
                setExclusiveOwnerThread(current);
                return true;
            }

**释放读锁**

protected final boolean tryReleaseShared(int unused) {
                Thread current = Thread.currentThread();
                if (firstReader == current) {
                    // assert firstReaderHoldCount > 0;
                    if (firstReaderHoldCount == 1)
                        firstReader = null;
                    else
                        firstReaderHoldCount--;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        rh = readHolds.get();
                    int count = rh.count;
                    if (count <= 1) {
                        readHolds.remove();
                        if (count <= 0)
                            throw unmatchedUnlockException();
                    }
                    --rh.count;
                }
                for (;;) {
                    int c = getState();
                    int nextc = c - SHARED_UNIT;
                    if (compareAndSetState(c, nextc))
                        // Releasing the read lock has no effect on readers,
                        // but it may allow waiting writers to proceed if
                        // both read and write locks are now free.
                        return nextc == 0;
                }
            }

**释放写锁**

protected final boolean tryRelease(int releases) {
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                int nextc = getState() - releases;
                boolean free = exclusiveCount(nextc) == 0;
                if (free)
                    setExclusiveOwnerThread(null);
                setState(nextc);
                return free;
            }

## 四 AQS总结 ##

学习ReentrantLock，ReentrantReadWriteLock，CountDownLatch和Semaphore之后发现这些类的结构大体相似，同时它们利用了AQS中的state，控制线程抢锁和配合的FIFO队列以及提供的获取/释放方法进行相应的重写来完成线程的并发控制。

ReentrantLock中将state作为某线程获取锁的重入次数，当state不为0时，存在其他线程就会放入阻塞队列中，进入挂起状态，从而使得ReentrantLock具有可重入锁，多条件锁，可中断锁的特性。

Semaphore中将state作为(信号量)许可证的数量，当一个线程申请了许可证之后，许可证的剩余数量小于0时，会将线程放入阻塞队列中，进入挂起状态，当有许可证（信号量）被释放后，阻塞队列中的进程会被唤醒，进入获取许可证的状态。

CountDownLatch中将state表示需要等待多收个线程获取完"共享锁"之后，CountDownLatch上阻塞的线程才可以获取"共享锁"，当state为不大于0的时候CountDownLatch上阻塞的锁就会获取"共享锁"。

ReentrantReadWriteLock则是基于AQS完成共享锁（读锁）和独占锁（写锁）的构造。