深入理解AQS框架

左手的ㄟ右手 2022-12-18 06:56 224阅读 0赞

### 深入理解AQS框架 ###

*   *  如何自己手动实现一把锁？
     *  ReentrantLock的内部结构以及源码分析
     *  非公平锁
     *  可打断和不可打断的区别
     *  公平锁
     *  解锁过程
     *  await队列
     *  读写锁

## 如何自己手动实现一把锁？ ##

在讲解ReentrantLock之前，我会先演示如何用AQS的框架手动实现一把锁。  
首先，我们定义一个自己的Sync类去继承AQS的模板类AbstractQueuedSynchronizer并且提供这个类里面抽象方法的实现，代码如下：

/**
         * 实现独占锁，锁状态0代表未获取锁，1代表已经获取到锁
         */
        private static class Sync extends AbstractQueuedSynchronizer {
        
            /**
             * 
             * @param arg
             * @return
             */
            @Override
            protected boolean tryAcquire(int arg) {
        
            	//可能会有多个线程争夺这个锁资源，因此使用CAS能够保证加锁操作的原子性
                if (compareAndSetState(0, 1)) {
        
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            /**
             * 独占锁只能有一个线程能获取到锁，所以释放锁无需CAS保证线程的安全性
             *
             * @param arg
             * @return
             */
            @Override
            protected boolean tryRelease(int arg) {
        
                if (getState() == 0) {
        
                    throw new UnsupportedOperationException();
                }
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            /**
             * 返回锁的状态
             *
             * @return
             */
            @Override
            protected boolean isHeldExclusively() {
        
                return getState() == 1;
            }
    
            /**
             * 创建一个新的condition
             *
             * @return
             */
            protected Condition newCondition() {
        
                return new ConditionObject();
            }
        }

Sync这个类主要是提供加锁和释放锁的逻辑，如果可以成功把state改为1则表示加锁成功，把state改为0则表示释放锁。AQS设计的性能比较高的地方在于它用的是CAS保证原子操作而不是让线程park睡眠，加锁成功则调用setExclusiveOwnerThread(Thread.currentThread())方法让当前线程占有锁。  
Sync 的代码写完了以后就可以写一个自己的锁，代码也很简单：

public class SelfLock implements Lock {
        
        private Sync sync = new Sync();
        
        /**
         * 请求锁
         */
        @Override
        public void lock() {
        
            sync.acquire(1);
        }
    
    	/**
    	* 先不支持打断
    	*/
        @Override
        public void lockInterruptibly() throws InterruptedException {
        
    
        }
    
        /**
         * 尝试获取锁
         *
         * @return
         */
        @Override
        public boolean tryLock() {
        
            return sync.tryAcquire(1);
        }
    
        /**
         * 在time时间内尝试获取锁
         *
         * @param time
         * @param unit
         * @return
         * @throws InterruptedException
         */
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        
            return sync.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        /**
         * 释放锁
         */
        @Override
        public void unlock() {
        
            sync.release(1);
        }
    
        @Override
        public Condition newCondition() {
        
            return sync.newCondition();
        }
    }

这样我们的锁类就写完了，接着写个测试类：

public class Test {
        
        static Lock lock = new SelfLock();
    
        public static void main(String[] args) {
        
            for (int i = 0; i < 10; i++) {
        
                new Thread(new Work()).start();
            }
        }
    
        private static class Work implements Runnable {
        
            @Override
            public void run() {
        
                lock.lock();
                try {
        
                    Thread.sleep(1000);
                    System.out.println(Thread.currentThread().getId() + " is work");
                    for (int i = 0; i < 5; i++) {
        
                        System.out.print(i);
                        System.out.print(" ");
                    }
                    System.out.println();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                } finally {
        
                    lock.unlock();
                }
            }
        }
    }

在测试类里面就起10个线程去输出0-4，如果我们的锁是有效那么每个线程就会排队输出而不会出现插队现象。OK，现在我们来瞧瞧运行结果，为了证明我写的锁是有效的我分别截出加锁的结果和未加锁的结果。  
加锁：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center]

未加锁：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 1]

## ReentrantLock的内部结构以及源码分析 ##

上面演示了怎么通过AQS来实现一把锁，是不是觉得简单到不敢想象。不过人家Doug Lea能够被称为大师，他写的代码肯定没有那么简单，下面我们来分析一下ReentrantLock的源码（敲黑板，重点来了）。

首先，我们来看看ReentrantLock的结构（先不着急看代码）：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 2]  
借助IDEA可以看到ReentrantLock有三个内部类，Sync、FairSync和NonfairSync。其中Sync是一个抽象类，FairSync和NonfairSync都是Sync的实现类。而Sync又继承了AbstractQueuedSynchronizer而AbstractQueuedSynchronizer又继承了AbstractOwnableSynchronizer这个类。下面对他们的作用做一个简单介绍：  
1、顶层父类AbstractOwnableSynchronizer的作用很简单，这个类主要是定义了一个存储当前持有锁的线程变量exclusiveOwnerThread，并且提供exclusiveOwnerThread变量的get和set方法。  
2、AbstractOwnableSynchronizer的子类AbstractQueuedSynchronizer是一个抽象类，主要定义了一个双向链表来存储一个Node节点，Node是AbstractQueuedSynchronizer的一个内部类，用于封装需要进入队列的线程，并且有prev和next变量指向它的前节点和后节点。为什么需要这样的一个链表呢？因为锁只能被一个线程持有，其他没有抢到锁的线程需要一个队列来存储需要睡眠排队的线程。**（注意，并非进入队列就代表排队，这个非常重要，重要到你能不能正确理解AQS的源码）**

3、FairSync和NonfairSync顾名思义就是公平锁和非公平锁，这里也有个很重要的点需要注意，**公平和非公平是体现在线程加锁的时候的，一旦线程入队并且睡眠等待就只能排队等待，也就没有公平和非公平可言。**（总结为：一朝排队永远排队）

上面都只是理论，下面我就用源码来证明这个些理论。  
首先我们先看ReentrantLock的构造方法，默认的构造器返回的是非公平锁，有参构造器根据fair来创建公平锁和非公平锁，代码如下：

/**
         * Creates an instance of {@code ReentrantLock}.
         * This is equivalent to using {@code ReentrantLock(false)}.
         */
        public ReentrantLock() {
        
            sync = new NonfairSync();
        }
    
        /**
         * Creates an instance of {@code ReentrantLock} with the
         * given fairness policy.
         *
         * @param fair {@code true} if this lock should use a fair ordering policy
         */
        public ReentrantLock(boolean fair) {
        
            sync = fair ? new FairSync() : new NonfairSync();
        }

ReentrantLock有两个构造方法，一个是有参构造方法，一个是无参构造方法。无参构造方法创建的是一把非公平锁，有参构造方法根据传入的布尔值判断创建的是公平锁还是非公平锁。所以默认使用的是非公平锁，下面先分析非公平锁然后再来对比公平锁的区别。

## 非公平锁 ##

首先我们来看ReentrantLock的lock方法：

public void lock() {
        
            sync.lock();
        }

ReentrantLock的lock方法很简单，就是调用sync的lock方法请求加锁。而sync有两个实现类，我们先来看非公平锁的实现类：

/**
         * 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() {
        
            	//非公平锁不管三七二十一，一进来就抢锁
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());//抢锁成功把当前线程设置为占有锁的线程
                else//抢锁失败再申请加锁
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
        
                return nonfairTryAcquire(acquires);
            }
        }

compareAndSetState是父类AbstractQueuedSynchronizer的方法，主要通过unsafe进行CAS操作修改stateoffset的值为1，CAS直接用的是CUP指令，效率相对来说会比较高。可以看的非公平锁之所以性能会比公平锁高就是因为线程进来的时候什么都不管，直接CAS抢锁，如果抢锁成功了就不用睡眠。  
下面重点分析抢锁失败后进入的acquire方法：

//acquire方法，就是一个模板方法，方法里的tryAcquire由子类实现
         public final void acquire(int arg) {
        
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

可以看的acquire方法在一个if判断里面调了三个方法，首先是tryAcquire方法尝试加锁，如果加锁失败则先进入addWaiter方法把当前线程打包成一个node并且入队，然后再进入acquireQueued方法自旋加锁，如果加锁失败则睡眠。  
首先分析tryAcquire代码：

protected final boolean tryAcquire(int acquires) {
        
                return nonfairTryAcquire(acquires);
            }
            
            /**
             * 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)) {
        //通过CAS尝试获取锁
                        //获取成功则设置持有锁的线程为当前线程并且返回
                        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;
            }

tryAcquire方法调用了nonfairTryAcquire方法，nonfairTryAcquire方法进来会再看看锁有没有被释放掉，如果释放掉会再抢一次锁，抢不到就判断这是不是锁重入，如果是则把锁标志+1返回true，否则返回false。假设我们的线程抢锁失败返回false，就会执行&&后面acquireQueued(addWaiter(Node.EXCLUSIVE), arg)的代码。  
先看addWaiter方法：

/**
         * Creates and enqueues node for current thread and given mode.
         *
         * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
         * @return the new node
         */
        private Node addWaiter(Node mode) {
        
            Node node = new Node(Thread.currentThread(), mode);//把当前线程封装成Node对象
            // Try the fast path of enq; backup to full enq on failure
            Node pred = tail;//获取双向链表的尾节点 pred 
            if (pred != null) {
        //如果pred 不为空，说明队列已经初始化，直接入队
                node.prev = pred;//把当前线程node的prev指向pred
                if (compareAndSetTail(pred, node)) {
        //通过CAS操作把tail改为当前线程node
                    pred.next = node;//把pred的next指向node，完成入队
                    return node;
                }
            }
            //如果入队失败失败或者队列没有初始化，此时则调用enq方法初始化队列并入队
            enq(node);
            return node;//返回的是当前线程的node
        }

addWaiter主要做的事情就是把当前线程打包成一个Node，如果队列已经初始化则直接入队，如果入队失败或者队列没有初始化则进入enq方法：

/**
         * Inserts node into queue, initializing if necessary. See picture above.
         * @param node the node to insert
         * @return node's predecessor
         */
        private Node enq(final Node node) {
        
            for (;;) {
        
                Node t = tail;//获取双向链表的最后一个node t
                if (t == null) {
         // t为空则说明队列没有初始化，此时就要进行初始化操作
                    if (compareAndSetHead(new Node()))//通过CAS操作把双向链表的head设置为一个新的Node
                    //思考个问题，为什么要new一个Node作为头节点和尾节点？？
                    //之所以要创建一个new的Node作为队首，是因为第一个线程假设是t1拿到锁的时候不会形成队列，
                    //此时第二个线程假设是t2进来但实际上t1还在执行，所以用一个线程指向null的Node作为队首代表的就是正在执行的t1线程
                    //把tail也指向这个new 的 Node，t2进来的时候第一次循环创建队列，第二次循环才进else把Node 入队
                    //之所以用一个死循环是因为这个CAS操作很可能会失败，因为t2线程在初始化队列的时候可能会有t3已经把队列初始化了，下次循环判断队列已经初始化不用初始化直接入队
                        tail = head;
                } else {
        
                    //队列已经初始化则把Node入队，此处操作和addWaiter方法入队一样
                    node.prev = t;//t不为空则把node的prev指向t
                    if (compareAndSetTail(t, node)) {
        //通过CAS操作把tail改为当前线程node
                        t.next = node;//把t的next指向node，完成入队
                        return t;
                    }
                }
            }
        }

当addWaiter方法执行完后，当前线程的node就成功入队了，而且还把这个node返回作为acquireQueued方法的第一个参数。我们来看看这个acquireQueued的方法：

/**
         * Acquires in exclusive uninterruptible mode for thread already in
         * queue. Used by condition wait methods as well as acquire.
         *
         * @param node the node
         * @param arg the acquire argument
         * @return {@code true} if interrupted while waiting
         */
        final boolean acquireQueued(final Node node, int arg) {
        
            boolean failed = true;
            try {
        
                boolean interrupted = false;//打断标志
                for (;;) {
        
                    final Node p = node.predecessor();//返回node的prev节点
                    //如果p是头节点，也就是node已经是第一个节点了，那么它会继续调tryAcquire尝试获取锁
                    //为什么要自旋一次是因为拿到锁的线程有可能已经执行完了并且把锁给释放，这样做能尽可能地避免线程睡眠
                    if (p == head && tryAcquire(arg)) {
        
                        //加锁成功，head的节点出队，当前线程的Node成为新的head，
                        //把node设置为head并且把node的thread和prev设置为空，因为tryAcquire加锁成功了，所以当前线程可以拿出来继续往下执行
                        setHead(node);
                        //加锁成功后node就成为了队首，所以p就需要出队了
                        //此时再把p的next值向空，就完成p节点的出队，到下次gc的时候就会回收该对象
                        p.next = null; // help GC
                        failed = false;
                        //if (!tryAcquire(arg) &&
                        //acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                        //selfInterrupt();
                        //返回打断标志，如果线程曾经被打断了都返回true，
                        //返回true会进入selfInterrupt方法再调一次Thread.currentThread().interrupt();进行打断（注意interrupt和interruptd的区别）
                        //打断后的逻辑需要在线程里面处理，如果没处理还是不会打断线程
                        //补充一点个人的理解，这里的不支持打断并不是真的不能打断，而是其实是需要抢到锁了才能调线程的interrupt去打断
                        return interrupted;
                    }
                    //node不是第一个节点或者尝试加锁失败后进入这里
                    //入队后还得判断我这个线程是否应该睡眠，如果需要睡眠才去睡眠
                    //shouldParkAfterFailedAcquire和parkAndCheckInterrupt分析看下文
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        //只有被打断才会把interrupted 改为true
                        //被打断后修改打断标志，下一次循环还是会去获取锁，
                        //不管下次获取锁是否成功，这个打断标志都为true
                        interrupted = true;
                }
            } finally {
        
                //一般来说failed是不会为true，除非你的线程发生了异常
                //如果为true则取消当前节点
                if (failed)
                    cancelAcquire(node);
            }
        }

acquireQueued的方法比较复杂，首先是一个for循环（自旋），for循环里面判断我是不是第一个排队的线程，如果是就尝试加锁，否则就判断我是不是应该睡眠，如果要睡眠才睡眠。此时的线程已经进入队列了，但还会尝试获取锁而没有进入睡眠，验证了上面的结论：**并非进入队列就代表排队**。  
下面再来分析shouldParkAfterFailedAcquire方法和parkAndCheckInterrupt方法，看看什么时候会让线程睡眠：

/**
         *
         *waitStatus：0代表无状态，-1代码睡眠，>0代表被取消
         *
         *
         * 这里为什么要通过当前线程改上一个节点的waitStatus？
         * 首先你不能自己改自己的waitStatus状态，因为你自己都还没睡眠如果你吧waitStatus状态改了就相当于欺骗别人，所以需要真的睡眠了才能改这个状态
         *但你真的睡眠了以后又没办法自己修改这个waitStatus，所以需要后面的线程帮你去改这个状态
         *
         *
         * @param pred node's predecessor holding status
         * @param node the node
         * @return {@code true} if thread should block
         */
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        
            //获取前一个节点的waitStatus
            //waitStatus表示当前节点是否需要唤醒下一个节点的线程
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)
                /*
                 * Node属性waitStatus值 = Node.SIGNAL（值是-1）时
                 * 前节点p已经是睡眠状态了，所以当前线程可以睡眠排队
                 */
                return true;
            if (ws > 0) {
        
                /*
                 * Node属性waitStatus值 > 0时，说明前节点p线程已经被取消，
                 * 让队列的最后一个waitStatus的值 不大于0的作为前节点.
                 * 找到前节点后不能肯定前节点的waitStatus是-1（睡眠的），所以执行到最后还会返回false
                 *返回false后不会进入睡眠而是再循环一次再次进入到当前方法修改前节点的waitStatus
                 */
                do {
        
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
        
                /*
                 * Node属性waitStatus值 = 0时
                 * 将前节点p的waitStatus的值改为-1
                 * 注意：这里改完后是返回false，所以还未睡眠，
                 * 还会在上面的循环里再自旋一次加锁，再进入这个方法改一次waitStatus
                 * 这样设计估计是因为我这个操作是CAS操作，是有可能失败的，（如果节点被取消了呢，虽然概率很少但不能保证不发生）
                 * 如果万一失败了则再进入这个方法根据前节点p的waitStatus再操作一遍（个人理解）
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

shouldParkAfterFailedAcquire方法进来就判断前一个node节点的waitStatus状态，如果是0则改为-1。改成功之后还会再循环一次加锁，如果加锁失败才进入parkAndCheckInterrupt方法睡眠：

/**
         * Convenience method to park and then check if interrupted
         *
         * @return {@code true} if interrupted
         */
        private final boolean parkAndCheckInterrupt() {
        
            LockSupport.park(this);//让线程睡眠
            //线程醒来有两种可能，一种是线程被唤醒，另一种是被打断
            //Thread.interrupted方法会把打断标志清空，但是返回值依然是清空前的值
            //由于lock方法设计成不支持打断，所以如果线程在睡眠的时候被打断了就用interrupted方法清空打断标志
            //但是否支持打断不是这里决定的，是由lock的doAcquireInterruptibly方法和lockInterruptibly和acquireQueued方法决定的
            //所以这里返回打断标志作为是否打断的依据（说白了就是我把标志返回给你，你要不要打断自己实现）
            return Thread.interrupted();
        }

## 可打断和不可打断的区别 ##

现在我们来对比一下支持打断的lockInterruptibly方法调用的doAcquireInterruptibly和不支持打断的lock方法调用的acquireQueued：

/**
         * Acquires in exclusive interruptible mode.
         * @param arg the acquire argument
         */
        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())
                        //对比lock的doAcquireInterruptibly方法可以看到支不支持打断主要的区别在于这里，
                        //lock的acquireQueued方法打断后还会进入循环再次尝试获取锁，
                        //如果获取不到则再次进入睡眠状态，让用户感觉不到线程被打断
                        //而这里不需要抢锁，直接抛异常，终止掉循环，
                        //所以打断后的逻辑需要写在catch块中对InterruptedException的异常处理
                        throw new InterruptedException();
                }
            } finally {
        
                if (failed)
                    cancelAcquire(node);
            }
        }

可以看的支持打断的lockInterruptibly调用的doAcquireInterruptibly方法是直接抛出异常，抛出异常后就会退出死循环不再抢锁。

## 公平锁 ##

看完非公平锁再来理解公平锁就简单多了，公平锁的加锁主要不同在于tryAcquire方法，下面看看公平锁的加锁：

/**
             * 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() &&
                        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 final boolean hasQueuedPredecessors() {
        
            // The correctness of this depends on head being initialized
            // before tail and on head.next being accurate if the current
            // thread is first in queue.
            Node t = tail; // Read fields in reverse initialization order
            Node h = head;
            Node s;
            //这个返回的代码看起来简单，实际上并不简单
            //判断h和t是否相等，这里可以分两种种情况分析：
            //1、队列没初始化，t和h都为null。
            //这种情况非常直观，队列都没形成说明肯定没人排队，直接就返回false。返回false后会通过CAS直接抢锁，即便抢锁失败了后面还是会有机会进来这里
            //2、队列已经初始化了，此时h和t不相等。
            //通过上面分析的非公平锁可知，队首h是指向的是一个线程为null的Node，所以当线程进来之后h和t不等的话肯定有线程在队列里
            //那么我们就拿第一个排队的Node s，head不是第一个排队，head指向的node其实是虚拟出来代表正在指向的线程
            //然后判断这个s是不是空，其实就是判断这个队列是不是只有一个线程排队（再强调一次，head节点正在执行并没有排队）
            //显然，只要s不为空则说明肯定这个队列排队的线程>=1，虽然这个队列初始化了并且还有线程排队但还得看当前线程和s的线程是否相同
            //1）如果不相等，那就去排队。这个很正常，别人都在排队,你又不是排在第一个的人所以你只能去乖乖排队
            //2）如果相等，说明你是第一个排队的人。排队的第一个人居然是自己为什么会这样呢？
            //以下只是我个人理解，不一定对。但你们可以参考一下
            //假设现在有个t1的线程持有锁了，然后t2线程进来判断是否有队列，此时肯定返回为false，因为队列根本没有初始化
            //返回false则!hasQueuedPredecessors()就是true那么就会执行CAS加锁，此时肯定加锁失败，因为t1已经持有锁了
            //那么tryAcquire就会返回false，则!tryAcquire(arg)为true，然后进入acquireQueued方法，
            //在执行acquireQueued方法钱会调用一个addWaiter把队列初始化并且把t2打包成一个node入队。虽然入队了，但还未睡眠
            //入队后就进入到acquireQueued方法，acquireQueued方法首先判断t2是不是第一个排队的，结果发现是，
            //那我就再调tryAcquire再次尝试获取锁，此时进来发现t2就排在了第一位，那就再给t2多一次加锁的机会，所以这里如果当前线程t2和s节点的线程相等就返回false让t2再尝试CAS获取锁 
            //如果说此时 CAS加锁失败了，就是进入到 acquireQueued方法调用 shouldParkAfterFailedAcquire逻辑（shouldParkAfterFailedAcquire的逻辑和前面非公平锁一致） 
            return h != t &&
                ((s = h.next) == null || s.thread != Thread.currentThread());
        }
    
    }

好了，公平锁最难的就是这里了。如果这个看懂了以后其他的和非公平锁差不多，都是一朝排队永远排队。

## 解锁过程 ##

看完ReentrantLock的加锁过程现在来看看解锁过程的源码（代码比较短，所以我放一起了）：

public void unlock() {
        
            sync.release(1);//调用release释放锁
        }
        
        //释放锁的逻辑
        public final boolean release(int arg) {
        
            //使用tryRelease尝试释放锁
            if (tryRelease(arg)) {
        
                Node h = head;
                //不等于0才会去unpark别
                if (h != null && h.waitStatus != 0)
                //锁的状态和持有锁的线程都被释放了，则调用unparkSuccessor唤醒下一个线程
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
        
        //尝试释放锁
        protected final boolean tryRelease(int releases) {
        
                int c = getState() - releases;//支持重入锁，所以释放一次就减1
                //为了代码的严谨性做的判断，一般来说释放锁的线程都是持有锁的线程
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                //如果c为0，说明这把锁可以释放了
                if (c == 0) {
        
                    free = true;
                    setExclusiveOwnerThread(null);//释放锁的时候把持有锁的线程清空
                }
                setState(c);//把锁的状态改为计算得到c值
                return free;
       }

重点来看unparkSuccessor的代码：

private void unparkSuccessor(Node node) {
        
            int ws = node.waitStatus;
            //如果head的ws小于0立马改成0，因为如果是-1还代表head有义务叫醒别人
            //改成0后就不需要唤醒别人
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            Node s = node.next;
            //极端情况，只有s被别人取消了才进
            if (s == null || s.waitStatus > 0) {
        
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            //s不为空则唤醒s，s就是队列中第一个排队的人
            //此时唤醒的s会在parkAndCheckInterrupt地方醒来，
            //然后再次进入到循环尝试加锁，加锁成功（一般来说都会加锁成功）则把head指向的node出队，然后再把head重新指向s
            if (s != null)
                LockSupport.unpark(s.thread);
        }

## await队列 ##

把公平锁和非公平锁讲清楚了之后我们再来看ReentrantLock的await队列，ReentrantLock的newCondition可以创建一个Condition。每个Condition都有一个对应的双向链表负责存储等待相同条件的线程Node，让线程睡眠等待的方法是await，让线程醒来的则是signal和signalAll方法，现在主要来看的是Condition的await、signal和signalAll方法。（synchronized的wait和await是相类似，不过synchronized的wait线程都放在一个队列里面，不像ReentrantLock的await那样可以根据Condition区分）

await：

/**
     *  await就是把当前线程打包成node放到Condition队列中
     *  释放锁并且唤醒ReentrantLock队列的第一个node，然后让当前线程睡眠
     **/
     public final void await() throws InterruptedException {
        
                if (Thread.interrupted())
                    throw new InterruptedException();
                //如果Condition的队列没有初始化则初始化队列并且把当前线程打包成node入队
                //此时的node是的waitStatus是-2，被signal转移到ReentrantLock队列后才改为0（看下面的transferForSignal方法）
                //返回队尾
                Node node = addConditionWaiter();
                //释放掉锁并且唤醒ReentrantLock队列的第一个node
                int savedState = fullyRelease(node);
                int interruptMode = 0;
                while (!isOnSyncQueue(node)) {
        
                    //让当前线程睡眠
                    LockSupport.park(this);
                    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                        break;
                }
                if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                    interruptMode = REINTERRUPT;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
      }

addConditionWaiter初始化Condition队列并把当前线程打包成waitStatus为-2的node进行入队，返回这个node：

private Node addConditionWaiter() {
        
                Node t = lastWaiter;
                // If lastWaiter is cancelled, clean out.
                if (t != null && t.waitStatus != Node.CONDITION) {
        
                    unlinkCancelledWaiters();
                    t = lastWaiter;
                }
                //创建出来的node的waitStatus的Node.CONDITION，Node.CONDITION的值是-2
                Node node = new Node(Thread.currentThread(), Node.CONDITION);
                if (t == null)
                    firstWaiter = node;
                else
                    t.nextWaiter = node;
                lastWaiter = node;
                return node;
            }

signal和signalAll的区别在于signal调用的是doSingal,而signalAll调用的是doSignalAll方法（把代码放一起好比较）：

public final void signal() {
        
                //判断持有锁的是不是当前线程
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                //拿到Condition的第一个node
                Node first = firstWaiter;
                if (first != null)
                    doSignal(first);//把first放到ReentrantLock队列中排队
            }
            
            
            public final void signalAll() {
        
                //判断持有锁的是不是当前线程
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                //拿到Condition的第一个node
                Node first = firstWaiter;
                if (first != null)
                    doSignalAll(first);//把first放到ReentrantLock队列中排队
            }

signal和signalAll的区别：doSignal只把一个等待的node转移到ReentrantLock的队尾进行排队，而doSignalAll则把所有的node都转移到ReentrantLock的队尾进行排队

private void doSignal(Node first) {
        
            //将第一个在Condition的waiter队列排队的node出队
            //然后调用transferForSignal将node放到ReentrantLock的队列里面
                do {
        
                    if ( (firstWaiter = first.nextWaiter) == null)
                        lastWaiter = null;
                    first.nextWaiter = null;
                } while (!transferForSignal(first) &&
                         (first = firstWaiter) != null);
            }
            
            private void doSignalAll(Node first) {
        
                //将lastWaiter 和 firstWaiter 清空，
                //因为signalAll会将所有在waiter排队的node都转移到ReentrantLock的队列里面
                lastWaiter = firstWaiter = null;
                //逐个转移，直到所有的node都转移完成为止
                do {
        
                    Node next = first.nextWaiter;
                    first.nextWaiter = null;
                    transferForSignal(first);
                    first = next;
                } while (first != null);
            }

transferForSignal：

final boolean transferForSignal(Node node) {
        
            //把Condition的waitStatus改为0
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
                
             //enq方法前面讲过，就是入队（入的是ReentrantLock的队列）
             //入队后会返回node前一个节点p
            Node p = enq(node);
            int ws = p.waitStatus;
            //CAS操作把前一个节点p的waitStatus改为-1
            //如果前节点p被取消了那就直接唤醒node的线程让他执行
            //CAS操作一般来说都能改成功，如果改不成功或许是p的节点突然被取消了
            //此时认为这个waitStatus是暂时的、无害的错误，可以唤醒node继续执行
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
        }

## 读写锁 ##

ReentrantLock的源码分析完了以后我们再来分析读写锁的源码，读写锁的源码也写的很绕，比较难，我也尽可能地讲细点。  
首先，我们来看看读写锁的构造方法：

public ReentrantReadWriteLock() {
        
            this(false);
        }
    
        public ReentrantReadWriteLock(boolean fair) {
        
            sync = fair ? new FairSync() : new NonfairSync();
            readerLock = new ReadLock(this);//实例化读锁
            writerLock = new WriteLock(this);//实例化写锁
        }
    	
    	//读锁的构造方法
        protected ReadLock(ReentrantReadWriteLock lock) {
        
            sync = lock.sync;
        }
    
    	//写锁的构造方法
        protected WriteLock(ReentrantReadWriteLock lock) {
        
            sync = lock.sync;
        }

可以看到读写锁的构造方法默认syn也是创建一把非公平锁（注意：这里的公平锁和非公平锁是ReentrantReadWriteLock里面的内部类，要和ReentrantLock的区分开来），然后再用sync去实例化读写锁，可以看的读锁和写锁的sync都是同一个对象。  
接下来我们来看看获取读写锁的代码：

ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();//获取写锁
        ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();//获取读锁

writeLock和readLock的方法直接返回ReentrantReadWriteLock的readerLock和writerLock：

/** 内部的读锁 */
        private final ReentrantReadWriteLock.ReadLock readerLock;
        /** 内部的写锁 */
        private final ReentrantReadWriteLock.WriteLock writerLock;
        
        public ReentrantReadWriteLock.WriteLock writeLock() {
         return writerLock; }
        public ReentrantReadWriteLock.ReadLock  readLock()  {
         return readerLock; }

然后我们先来看看写锁加锁方法lock的代码：

public void lock() {
        
                sync.acquire(1);//调用sync的acquire方法
            }
            
            /**
             * acquire方法前面已经介绍过了，就是一个模板方法，
             * 前面已经分析读锁和写锁的sync对象都是同一个，
             * 而这里的tryAcquire就是sync的tryAcquire方法，
             * 如果tryAcquire方法加锁失败后的逻辑和ReentrantLock是一样的，
             * 都是调用addWaiter和acquireQueued进入队列并且准备睡眠
             * addWaiter和acquireQueued是父类AbstractQueuedSynchronizer实现的，所以逻辑和ReentrantLock一样，这里就不再重复讲解
             */
            public final void acquire(int arg) {
        
                if (!tryAcquire(arg) &&
                    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                    selfInterrupt();
            }
    
    		/**
    		 * 这是ReentrantReadWriteLock的内部类Sync的tryAcquire方法
    		 */
            protected final boolean tryAcquire(int acquires) {
        
                //拿到当前线程
                Thread current = Thread.currentThread();
                //拿到当前锁状态
                int c = getState();
    			//这里通过运算得到写锁的状态,后面作一个比较详细的分析
                int w = exclusiveCount(c);
                //如果已经上锁了，有可能是读锁也有可能是写锁，也有可能是降级锁（写读锁），读写锁不支持升级所以就没有读写锁
                //c的前16位代表的是读锁状态，后16位代表的是写锁状态（后面证明）
                if (c != 0) {
        
                    // 1、上的是读锁（w==0为true代表不是写锁，所以只能是读锁）直接返回false，
                    //    假如现在的锁状态是读锁，而这个方法请求的是写锁并且返回了false，说明不支持锁的升级
                    // 2、不是写锁重入锁也返回false，
                    //    就算上的是写锁，如果上锁的线程不是当前线程也返回了false，说明写写互斥
                    if (w == 0 || current != getExclusiveOwnerThread())
                        return false;
                    // 第一个if没返回false代表这个锁要么是写锁，要么是写读锁，这时支持写锁的重入
                    // 由于是写锁重入，所以只需判断写锁的重入次数是否大于MAX_COUNT，这个判断一般来说无用，因为没人会重入这么多次
                    if (w + exclusiveCount(acquires) > MAX_COUNT)
                        throw new Error("Maximum lock count exceeded");
                    // 重入成功直接修改state，由于后16位代表写锁状态，所以直接相加即可
                    setState(c + acquires);
                    return true;
                }
                //如果没有上锁的话，这个流程和ReentrantLock相似，
                //公平锁对writerShouldBlock的实现是直接调用AQS模板的hasQueuedPredecessors方法（前面已经讲解），
                //所以这里的公平锁也是先判断有无队列，为什么要
                //如果有队列则返回true，说明没有队列可以执行compareAndSetState代码加锁
                //有队列则返回fale，此时不能直接加锁，需要入队
                //而非公平锁对writerShouldBlock的实现是直接返回false，所以非公平锁则直接抢锁
                if (writerShouldBlock() ||
                    !compareAndSetState(c, c + acquires))
                    //加锁失败则返回false
                    return false;
                //加锁成功则把当前线程设置为占有锁线程
                setExclusiveOwnerThread(current);
                return true;
            }

看到这里可以得出一个结论：如果一个线程请求的是写锁，而这个锁的状态已经是上锁了的话，那么就只能是重入。总结为：**读写互斥，写写互斥**。  
接下来再看看exclusiveCount方法是如何运算得到写锁的状态的：

static int exclusiveCount(int c) {
         
    		return c & EXCLUSIVE_MASK; 
    	}

这个exclusiveCount方法就只有一行代码，首先我们来看EXCLUSIVE\_MASK的值是多少：

static final int SHARED_SHIFT   = 16;
    	static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

可以看到EXCLUSIVE\_MASK 的值就是1向左移动16位 - 1，所以就是2^16 - 1 = 65,536 - 1 = 65535(不信可以自己debug一下)，转换为二进制就是1111111111111111。然后c是一个int类型，它有32位，如果让c和EXCLUSIVE\_MASK做一个与运算，那么得到的就是c的低16位的值（也就是说c的后16位代表写锁状态）。  
接下来我们分析一下读锁的加锁代码（**读锁是很复杂的**）：

//读锁的lock代码
        public void lock() {
        
            sync.acquireShared(1);
        }
    
    	//AbstractQueuedSynchronizer的acquireShared代码，这是一个共享锁
        public final void acquireShared(int arg) {
        
            //tryAcquireShared 加锁成功则返回一个正数（读锁可以重入），加锁失败则返回-1
            if (tryAcquireShared(arg) < 0)
                doAcquireShared(arg);
        }

下面再来看看tryAcquireShared的代码：

//返回-1表示加锁失败
        protected final int tryAcquireShared(int unused) {
        
        		//获取当前线程
                Thread current = Thread.currentThread();
                //获取锁的state
                int c = getState();
                //首先exclusiveCount(c) != 0判断是否上了写锁，上了写锁还得判断持有锁的线程是不是当前线程
                //如果不是当前线程说明不是锁的降级则直接返回-1，表示上锁失败，因为读写互斥
                //如果没上锁、上读锁或者锁的降级则不return
                if (exclusiveCount(c) != 0 &&
                    getExclusiveOwnerThread() != current)
                    return -1;
                    
                //得到读锁上锁次数r
                int r = sharedCount(c);
                
                /**
                 * 1、不需要排队
                 * 2、r少于最大的重入次数MAX_COUNT 
                 * 3、加锁是否成功（由于加锁是对c的前16位操作，所以SHARED_UNIT的值是1向右移动16位）
                 */
                if (!readerShouldBlock() &&
                    r < MAX_COUNT &&
                    compareAndSetState(c, c + SHARED_UNIT)) {
        
                    //1、2、3满足则进到这里，为什么还有进这里？因为加锁成功后还得区分是不是锁的降级
    
    				/**
    				 * r代表的是加锁成功前的加锁次数
    				 * 如果r==0为true，则说明之前没有线程给它加过锁或者持有锁的所有线程都释放了这把锁
    				 * 如果持有锁的所有线程都把锁释放了则会对firstReader 和 firstReaderHoldCount 重新赋值
    				 * 如果没有线程加过锁，则把当前线程赋值给一个局部变量firstReader ，并且记录加锁次数firstReaderHoldCount 
    				 * 此时读锁加锁完成，返回1
    				 */
                    if (r == 0) {
        
                        firstReader = current;
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {
        
                    	//如果firstReader和当前线程相同，说明这是读锁的重入，读锁重入则把firstReaderHoldCount++，记录重入次数
                        firstReaderHoldCount++;
                    } else {
        
                       //否则的话就是读锁的重入，这里的代码很绕很复杂，Doug Lea为了将代码的性能提高到极限做了一个很绕的操作
                       //因为读锁支持并发，所以需要有个线程安全的数据结构来存储每个线程的重入次数，Doug Lea选择了使用ThreadLocal，并对ThreadLocal进行了扩展
                       //虽然ThreadLocal能够存储重入的次数，但是Doug Lea认为线程每次去ThreadLocal拿这个数肯定比在局部变量里拿要慢，
                       //所以把第一个加锁的线程记录到firstReader变量里，把最后一个加锁的线程记录到cachedHoldCounter中，以后这两个线程再来重入就不用到ThreadLocal拿
                       //这里可以分为两种情况：
                       //1、cachedHoldCounter为空或者cachedHoldCounter记录的线程不是当前线程，
                       //   则从ThreadLocal的子类对象readHolds的get方法去取一个HoldCounter并复制给cachedHoldCounter ，
                       //   readHolds和get方法下面分析，它会把HoldCounter缓存到readHolds中，此处就是将最后一个重入的线程缓存到readHolds
                       //2、cachedHoldCounter记录的是当前线程，则直接修改cachedHoldCounter计数器加+1
                        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);
            }

读锁最复杂的地方在于最后一个else了，这也提现了Doug Lea的代码的厉害之处（哪怕只是0.01%的性能也尽量去提高）。接下来分析一下Doug Lea扩展的ThreadLocal，从上面的代码分析可以指定ThreadLocal存储的是一个HoldCounter对象，我们先来看看HoldCounter的代码，可以看到HoldCounter有一个计数器count和一个线程id的属性：

static final class HoldCounter {
        
                int count = 0;
                // Use id, not reference, to avoid garbage retention
                final long tid = getThreadId(Thread.currentThread());
            }

然后下面直接看ThreadLocalHoldCounter的代码，可以看到ThreadLocalHoldCounter重写了initalValue的方法：

static final class ThreadLocalHoldCounter
                extends ThreadLocal<HoldCounter> {
        
                //重写initialValue方法，返回一个new 的HoldCounter
                public HoldCounter initialValue() {
        
                    return new HoldCounter();
                }
            }

我们再来看看ThreadLocal的get方法和setInitialValue方法：

public T get() {
        
    		//拿到当前线程t
            Thread t = Thread.currentThread();
            //使用t去获取一个map，如果第一次进来肯定获取不到，所以是个null
            ThreadLocalMap map = getMap(t);
            if (map != null) {
        
                ThreadLocalMap.Entry e = map.getEntry(this);
                if (e != null) {
        
                    @SuppressWarnings("unchecked")
                    T result = (T)e.value;
                    return result;
                }
            }
            //没获取到则调用setInitialValue并返回结果
            return setInitialValue();
        }
        
        //setInitialValue方法，初始化这个map，并且返回对应的value值
        private T setInitialValue() {
        
            T value = initialValue();//value就是一个new HoldCounter
            Thread t = Thread.currentThread();//拿到当前线程
            ThreadLocalMap map = getMap(t);//拿到t线程里面的属性threadLocals返回给map（默认是空）
            if (map != null)
                map.set(this, value);//如果不为空则把t和value作为key和value设置到map里面
            else
                createMap(t, value);//如果是空的则用t和value创建一个map
            return value;
        }

看完读写锁的加锁过程，现在来看看解锁过程，先来看看写锁的解锁过程：

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

上面是AQS的模板代码，下面主要看写锁的tryRelease的实现：

protected final boolean tryRelease(int releases) {
        
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                int nextc = getState() - releases;//释放锁
                boolean free = exclusiveCount(nextc) == 0;//是否把锁都释放完了，exclusiveCount上面说了是计算写锁的status
                if (free)
                    setExclusiveOwnerThread(null);//如果已经释放完了则把占有锁的线程设置为null，标识没有线程占有
                setState(nextc);//修改锁的状态
                return free;
            }

写锁的释放锁的过程相对来说比较简单，下面看读锁的释放锁的过程：

public void unlock() {
        
            sync.releaseShared(1);
        }
    
        public final boolean releaseShared(int arg) {
        
            if (tryReleaseShared(arg)) {
        
                doReleaseShared();
                return true;
            }
            return false;
        }

可以看到读锁的释放的是一个共享锁，我们来看看读锁的tryReleaseShared的实现：

protected final boolean tryReleaseShared(int unused) {
        
                Thread current = Thread.currentThread();//拿到当前线程
                //判断当前线程是不是缓存的firstReader，如果是则直接操作firstReader释放锁
                if (firstReader == current) {
        
                    // assert firstReaderHoldCount > 0;
                    if (firstReaderHoldCount == 1)
                        firstReader = null;
                    else
                        firstReaderHoldCount--;
                } else {
        
                	//如果不是firstReader的话再来看是不是缓存的cachedHoldCounter，如果是则直接操作cachedHoldCounter，
                	//否则的话从readHolds里拿出对应的HoldCounter来操作
                    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;
                }
                //由于可能有多个线程同时释放锁（读锁是一把共享锁），所以用CAS保证操作的原子性
                for (;;) {
        
                    int c = getState();
                    int nextc = c - SHARED_UNIT;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }

把锁的状态改成功了以后还会进入到doReleaseShared，下面重点来看doReleaseShared的代码：

private void doReleaseShared() {
        
            //死循环，因为是共享锁，所以可以唤醒队列中的多个读锁
            for (;;) {
        
                Node h = head;
                /**
                 * 第一个if判断头结点不为空并且头结点不等于尾节点，也就是队列不为空的时候才进
                 * 进来后首先判断我这个头结点要不要去唤醒别人，如果要唤醒别人先把头结点的状态改0，改成功后再调unparkSuccessor去唤醒线程
                 * 如果是不需要唤醒别人则尝试把头结点的状态改为-3
                 */
                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;
            }
        }

一旦唤醒了下一个线程，那么排队的线程就会从doAcquireShared方法中醒来，醒来后并且加锁成功则唤醒下一个结点。这里确实很绕，不容易想到释放锁会进到这里来。

private void doAcquireShared(int arg) {
        
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
        
                boolean interrupted = false;
                for (;;) {
        
                    final Node p = node.predecessor();
                    if (p == head) {
        //线程醒来后，p是等于head
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
        //加锁成功后才进
                            //把node设置为head，判断是否唤醒下个结点，如果要唤醒则唤醒，如果不要唤醒则直接返回
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            if (interrupted)
                                selfInterrupt();
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())//线程从这里醒来，进入下一次循环
                        interrupted = true;
                }
            } finally {
        
                if (failed)
                    cancelAcquire(node);
            }
        }

上面我写注释说是setHeadAndPropagate方法唤醒下一个线程，我们来看看代码证明一下是不是真的这样：

private void setHeadAndPropagate(Node node, int propagate) {
        
            Node h = head; 
            setHead(node);
    
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
        
                //拿到下一个结点s
                Node s = node.next;
                //只要s是为空或者s的nextWaiter等于SHARED，也就是共享的，就doReleaseShared释放共享锁再唤醒下一个线程
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }

看完上面的代码后，好像可以得出一个结论，如果排队的线程如下图：

head 读锁

t1 读锁 SHARED

t2 读锁 SHARED

t3 写锁

t4 读锁 SHARED tail

假设此时head把读锁释放了的话，那么应该只有t1和t2可以成功获取到锁，t3和t4则获取不到锁，因为t3的nextWaiter不等于SHARED，因此t3往后的线程都不会唤醒。  
那么这个结论对不对呢？我们可以通过下面的代码验证一下：

public static void main(String[] args) {
        
            ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
            ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
            ReentrantReadWriteLock.ReadLock readLock = readWriteLock.readLock();
    
            Thread head = new Thread(() -> {
        
                readLock.lock();
                try {
        
                    sleep(10000);
                } finally {
        
                    readLock.unlock();
                }
            });
            head.setName("head");
            head.start();
            sleep(1000);
    
            Thread t1 = new Thread(() -> {
        
                readLock.lock();
                try {
        
                    sleep(1000);
                    System.out.println("t1抢到锁了");
                } finally {
        
                    readLock.unlock();
                }
            });
            t1.setName("t1");
            t1.start();
            sleep(1000);
    
            Thread t2 = new Thread(() -> {
        
                readLock.lock();
                try {
        
                    System.out.println("t2抢到锁，进入死循环了");
                    while (true){
        
    
                    }
                } finally {
        
                    readLock.unlock();
                }
            });
            t2.setName("t2");
            t2.start();
            sleep(1000);
    
            Thread t3 = new Thread(() -> {
        
                writeLock.lock();
                try {
        
                    System.out.println("t3抢到锁了");
                } finally {
        
                    writeLock.unlock();
                }
            });
            t3.setName("t3");
            t3.start();
            sleep(1000);
    
            Thread t4 = new Thread(() -> {
        
                readLock.lock();
                try {
        
                    System.out.println("t4抢到锁了");
                } finally {
        
                    readLock.unlock();
                }
            });
            t4.setName("t4");
            t4.start();
        }
    
        public static void sleep(long time){
        
            try {
        
                Thread.sleep(time);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 3]  
依次创建5个线程head（读锁）、t1（读锁）、t2（读锁）、t3（写锁）、t4（写锁），让head线程加读锁然后睡眠10秒钟，然后再释放锁，t2抢到锁后进入一个死循环不释放锁。运行结果发现，当head线程释放了锁之后，t1、t2都被唤醒抢到读锁，而t4没有被唤醒，证实了上面的那个结论。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221122/e23b6b3233ad421b8fd6894f77e12e5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221122/7666e7a65753475da2248877ad3fd7ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221122/976dff61c2be4ee59d81f8db364ded4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0phbmtlRGVuZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221122/ff160690d1f94ae2afb54144addb9ff8.png