jdk11源码-ReentrantLock源码

红太狼 2022-10-02 11:54 383阅读 0赞

> 更多java源码分析请见：jdk11源码分析系列文章专栏：[Java11源码分析][Java11]

### 文章目录 ###

*  欢迎关注本人公众号
 *  概述
 *  new ReentrantLock
 *  AQS队列
 *   *  waitStatus
 *  非公平锁加锁
 *   *  lock.lock();
     *  tryAcquire
     *  addWaiter和enq方法
     *  AQS队列操作过程中的并发问题
     *  acquireQueued
     *  获取非公平锁过程总结
     *  cancelAcquire
 *  公平锁加锁
 *  release
 *  AQS内部实际状态
 *  几个疑问
 *   *  AQS是FIFO的，那么他是怎么实现非公平锁的？
     *  非公平锁：饥饿
     *  ReentrantLock性能是如何保证的？
     *  ReentrantLock是通过java代码层实现的，那么其线程安全是如何保证 的？
     *  羊群效应

# 欢迎关注本人公众号 #

![在这里插入图片描述][20200515113638671.jpg_pic_center]

# 概述 #

ReentrantLock是java中常用的加锁工具，下面是一个典型的写法：

ReentrantLock lock = null;
    try { 
        System.out.println(System.currentTimeMillis());
        lock = new ReentrantLock();
        lock.lock();
        lock.lock();
    
        TimeUnit.SECONDS.sleep(1);
        System.out.println(System.currentTimeMillis());
    } catch (Exception e) { 
        e.printStackTrace();
    } finally { 
        if (null != lock) { 
            lock.unlock();
            lock.unlock();
        }
    }

ReentrantLock使用非常灵活，支持公平锁和非公平锁，并且可以与condition配合使用来实现复杂的应用场景。接下来就详细分析一下ReentrantLock的源码实现。

**类图**  
所有源码的解读，都要现有一个概览，从全局解读出发，然后逐步细化理解各个模块的含义。我们先看一下整体的类图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]  
其中Sync是ReentrantLock内部的一个抽象类。NonfairSync和FairSync是ReentrantLock公平锁和非公平锁的实现。AbstractQueuedSynchronizer维护了AQS队列，AQS队列是ReentrantLock实现的核心。

# new ReentrantLock #

ReentrantLock构造函数有两个重载，当参数fair=true时，创建的是公平锁（公平与非公平锁后面会讲）。默认创建的是非公平锁。

public ReentrantLock() { 
        sync = new NonfairSync();
    }
    
    public ReentrantLock(boolean fair) { 
        sync = fair ? new FairSync() : new NonfairSync();//需注意这里三元运算符的优先级高哦
    }

# AQS队列 #

这里简单介绍一下AQS队列的结构，后面会详细介绍队列的变更请求。  
AQS的核心实现在AbstractQueuedSynchronizer，该类继承了AbstractOwnableSynchronizer。AbstractOwnableSynchronizer内部实现比较简单，核心代码：

/** * 独占模式下，当前锁的拥有者。 * 指某个线程拥有锁 */
    private transient Thread exclusiveOwnerThread;
    
    protected final void setExclusiveOwnerThread(Thread thread) { 
        exclusiveOwnerThread = thread;
    }

上面提到了独占模式，AQS定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，ReentrantLock使用该模式）和Share（共享，多个线程可同时执行，Semaphore/CountDownLatch使用该模式）。  
在AbstractQueuedSynchronizer的内部类Node中定义了这两种模式：

static final class Node { 
        /** 共享模式 */
        static final Node SHARED = new Node();
        /** 独占模式 */
        static final Node EXCLUSIVE = null;
    }

好了，进入正题，AQS队列。AQS队列是一个等待队列，他是CLH队列的变体。关键字段如下：

private transient volatile Node head;//队列头 
    private transient volatile Node tail; //队列尾
    private volatile int state;  //状态
    
    static final class Node { 
            static final Node SHARED = new Node();//共享模式
            static final Node EXCLUSIVE = null;  //独占模式
    
    //下面这四个值是变量waitStatus的值
            static final int CANCELLED =  1; 
            static final int SIGNAL    = -1;      
            static final int CONDITION = -2;  
            static final int PROPAGATE = -3;
            
            volatile int waitStatus;
            volatile Node prev;//前驱节点
            volatile Node next;//下一个节点
            volatile Thread thread;//当前节点代表的线程
            /** 排他锁模式下表示正在等待条件的下一个节点，因为只有排他锁模式有conditions；所以在共享锁模式下，我们使用’SHARED’这个特殊值来表示该字段。 */
            Node nextWaiter;
    }

state是关键之一，他是volatile的，保证了线程间可见性。ReentrantLock加锁的所标记就记录在state中，state初始值是0，每当线程请求一个锁，state加1，在本文开始的代码中，连续获取两个锁，获取锁成功后，state值是2。ReentrantLock是可重入的，之所以可重入，就是依赖这个volatile 的state保证的。当然这里可重入指的是同一个获取锁的线程可以多次获取锁，然后获取N次锁，必须unlock释放N次，保证state=0才表明当前线程释放了锁。  
在多线程并发请求锁时，采用CAS修改state的值，修改成功则获取锁成功，修改失败则加入到AQS等待队列尾部。

## waitStatus ##

waitStatus是状态属性，AQS队列的核心之一。他有四个值：

*  static final int SIGNAL = -1  
    这个节点的后继节点是阻塞的（LockSupport.park(this)），所以当前节点被释放或者取消时需要唤醒它的后继节点（后继线程）。
 *  static final int CANCELLED = 1;  
    这个节点由于超时或中断被取消了。CANCELLED 状态的节点永远不会再次阻塞。
 *  static final int CONDITION = -2;  
    这个节点当前在一个condition队列中。
 *  static final int PROPAGATE = -3;  
    一个releaseShared操作必须被广播给其他节点。（只有头节点的）该值会在doReleaseShared方法中被设置去确保持续的广播，即便其他操作的介入。
 *  0：不是上面的值的情况。  
    这个值使用数值排列以简化使用。非负的值表示该节点不需要信号（通知）。因此，大部分代码不需要去检查这个特殊的值，只是为了标识。  
    对于常规的节点该字段会被初始化为0，竞争节点该值为CONDITION。

AQS队列典型结构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]

> AQS队列的头结点并不关联任何线程，他是一个默认的Node节点。后面enq方法会具体讲

# 非公平锁加锁 #

## lock.lock(); ##

我们先看非公平锁的实现。由于是非公平锁，所以`lock.lock();`调用的是`NonfairSync`类的lock方法：

static final class NonfairSync extends Sync { 
            final void lock() { 
                if (compareAndSetState(0, 1))      //1
                    setExclusiveOwnerThread(Thread.currentThread());  //2
                else
                    acquire(1);  //3
            }
    
            protected final boolean tryAcquire(int acquires) { 
                return nonfairTryAcquire(acquires);   //4
            }
        }
    
    public final void acquire(int arg) { //5
            if (!tryAcquire(arg) &&  //6
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))   //7
                selfInterrupt();   //8
        }

总体流程如下：  
第一步：CAS原子性的修改`AbstractQueuedSynchronizer#state`的值，由0改为1，成功则说明当前线程加锁成功.  
然后第二步，设置`AbstractOwnableSynchronizer#exclusiveOwnerThread`的值为当前线程，表示当前锁的拥有者是当前线程。  
如果第一步中修改失败，则进入第三步：acquire(1)。申请1个state（共享锁可以申请多个哦），这里可以理解为申请一个信号量，因为ReentrantLock的实现基本都是在java层面使用代码实现的。  
acquire方法中首先尝试获取锁tryAcquire（第6步），如果获取失败，则将当前线程以独占模式`Node.EXCLUSIVE`加入等待队列尾部（addWaiter方法）。  
acquireQueued()：以独占无中断模式获取锁，这个方法会一直无限循环，直到获取到资源或者被中断才返回。如果等待过程中被中断则返回true。`这里有自旋锁的意思，加入队列中的线程，不断的重试检测是否可以执行任务。`

接下来一个一个方法逐个阅读源码：

## tryAcquire ##

tryAcquire的具体实现是在NonfairSync类中，然后调用其父类Sync 中的nonfairTryAcquire（）方法。

static final class NonfairSync extends Sync { 
        protected final boolean tryAcquire(int acquires) { 
            return nonfairTryAcquire(acquires);
        }
    }
        
    final boolean nonfairTryAcquire(int acquires) { 
        final Thread current = Thread.currentThread();
        int c = getState();//:1、获取volatile int state的值
        if (c == 0) { //2：state=0表示当前可以加锁
            if (compareAndSetState(0, acquires)) { //CAS将state设置为acquires的值
                setExclusiveOwnerThread(current);//设置当前拥有锁的线程
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) { //当前锁的拥有者线程是currentThread
            int nextc = c + acquires;//将state累加上acquires
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);//设置state的值。由于这里只有获取锁的线程才能执行，所以不会出现并发，不需要额外的加锁处理
            return true;
        }
        return false;//当前锁的拥有者线程不是currentThread，直接返回false，也就是获取锁失败
    }

注释已经添加好，nonfairTryAcquire的实现还是比较简单的：如果当前没有锁，那么加锁。如果已经有了锁，那么看看当前锁的拥有者线程是不是currentThread，是则累加state的值，不是则返回失败。  
`**特别说明：这也是ReentrantLock为什么是可重入锁的原因，同一个线程加多次锁（lock.lock）也就是给state的值累加而已。**`

## addWaiter和enq方法 ##

按照指定模式（独占还是共享）将节点添加到等待队列。

private Node addWaiter(Node mode) { 
        Node node = new Node(Thread.currentThread(), mode);
        //1、首先尝试以快速方式添加到队列末尾
        Node pred = tail;//pred指向现有tail末尾节点
        if (pred != null) { 
        //新加入节点的前一个节点是现有AQS队列的tail节点
            node.prev = pred;
          	//CAS原子性的修改tail节点
            if (compareAndSetTail(pred, node)) {     
                //修改成功，新节点成功加入AQS队列，pred节点的next节点指向新的节点
                pred.next = node;
                return node;
            }
        }
        //2、pred为空，或者修改tail节点失败，
        //则走enq方法将节点插入队列
        enq(node);
        return node;
    }
    private Node enq(final Node node) { 
        for(;;) { //CAS
            Node t = tail;
            if (t == null) {  
            // 必须初始化。这里是AQS队列为空的情况。
            //通过CAS的方式创建head节点，并且tail和head都指向
            //同一个节点。
                if (compareAndSetHead(new Node()))
                //注意这里初始化head节点，并不关联任何线程！！
                    tail = head;
            } else { 
            //这里变更node节点的prev指针，并且移动tail指针指向node，
            //前一个节点的next指向新插入的node
                node.prev = t;
                if (compareAndSetTail(t, node)) { 
                    t.next = node;
                    return t;
                }
            }
        }
    }

*  上面addWaiter方法的第一行代码`new Node(Thread.currentThread(), mode);`，会创建一个node对象，该对象的重要属性值初始化为：

nextWaiter = Node.EXCLUSIVE; // Node.EXCLUSIVE值为null
    thread = Thread.currentThread();
    waitStatus = 0;// 默认是0

*  addWaiter首先会以快速方式将node添加到队尾，如果失败则走enq方法。失败有两种可能，一个是tail为空，也就是AQS为空的情况下。另一是compareAndSetTail失败，也就是多线程并发添加到队尾，此时会出现CAS失败。
 *  注意enq方法，在`t==null`时，首先创建空的头节点，不关联任何的线程，nextWaiter和thread变量都是null.

## AQS队列操作过程中的并发问题 ##

先看一下没有并发的情况，AQS队列的变化过程：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]

**思考个问题，这里enq方法会存在并发的问题，那么如何保证线程安全的呢？**  
首先使用CAS来操作线程间的共享变量，比如head,tail。由compareAndSetTail和compareAndSetHead方法保证。

//通过objectFieldOffset方法来获取属性的偏移量，用于后面的CAS操作
    headOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
    tailOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
    //CAS：比较交换
    private final boolean compareAndSetTail(Node expect, Node update) { 
        return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
    }

其次，prev，next，t这几个变量如何保证线程安全的呢？  
enq方法中的`tail = head;`这行代码，肯定只有一个线程执行，因为上面有一个CAS，这里只有一个线程会修改成功，进入if方法体。而且这里只存在于AQS队列没有初始化时只执行一次，比较简单。  
另外，**t是方法内部局部变量，是线程私有的，不会有线程安全问题**。当多个线程都想添加到AQS队列尾部，肯定只有一个执行compareAndSetTail成功并且移动tail指针指向最新的末尾node，其他线程则会重新执行for循环。

## acquireQueued ##

上面tryAcquire失败没有获取到锁，addWaiter加入了AQS等待队列，进入acquireQueued方法中，acquireQueued方法以独占无中断模式获取锁，这个方法会一直无限循环，直到获取到资源或者被中断才返回。有点类似于自旋锁的意思。

final boolean acquireQueued(final Node node, int arg) { 
        boolean failed = true;//是否获取到资源
        try { 
            boolean interrupted = false;//是否中断
            for (;;) { 
                //获取前一个节点
                final Node p = node.predecessor();
    
    			//如果当前node节点是第二个节点，紧跟在head后面，
    			//那么tryAcquire尝试获取资源
                if (p == head && tryAcquire(arg)) {                 
                    setHead(node);//获取锁成功，当前节点成为head节点
                    p.next = null; // 目的：辅助GC
                    failed = false;
                    return interrupted;//返回是否中断过
                }
                
                //当shouldParkAfterFailedAcquire返回成功，
                //也就是前驱节点是Node.SIGNAL状态时，
                //进行真正的park将当前线程挂起，并且检查中断标记，
                //如果是已经中断，则设置interrupted =true。
                //如果shouldParkAfterFailedAcquire返回false，
                //则重复上述过程，直到获取到资源或者被park。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);//添加AQS失败，取消任务
        }
    }
    
    //前面讲过，head节点不与任何线程关联，他的thread是null，
    //当然head节点的prev肯定也是null
    private void setHead(Node node) { 
        head = node;
        node.thread = null;
        node.prev = null;
    }
    
    //在Acquire失败后，是否要park中断
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        int ws= pred.waitStatus;//获取到上一个节点的waitStatus
        if (ws == Node.SIGNAL)//前面讲到当一个节点状态时SIGNAL时，
        //他有责任唤醒后面的节点。所以这里判断前驱节点是SIGNAL状态，
        //则可以安心的park中断了。
            return true;
        if (ws > 0) { 
            /* * 过滤掉中间cancel状态的节点 * 前驱节点被取消的情况（线程允许被取消哦）。向前遍历， * 直到找到一个waitStatus大于0的（不是取消状态或初始状态） * 的节点，该节点设置为当前node的前驱节点。 */
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            /* * 修改前驱节点的WaitStatus为Node.SIGNAL。 * 明确前驱节点必须为Node.SIGNAL，当前节点才可以park * 注意，这个CAS也可能会失败，因为前驱节点的WaitStatus状态 * 可能会发生变化 */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    //阻塞当前线程
    //park并且检查是否被中断过
    private final boolean parkAndCheckInterrupt() { 
        LockSupport.park(this);
        return Thread.interrupted();
    }

## 获取非公平锁过程总结 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]

## cancelAcquire ##

上面acquireQueued方法在出现异常时，会执行cancelAcquire方法取消当前node的acquire操作。接下来看看cancelAcquire()都做了哪些事情。

private void cancelAcquire(Node node) { 
        if (node == null)
            return;
    
        node.thread = null;
    
        // 跳过中间CANCELLED状态的节点
        Node pred = node.prev;
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;
    
        Node predNext = pred.next;
    
        // 将node设置为CANCELLED状态
        node.waitStatus = Node.CANCELLED;
    
        // 如果当前节点是tail节点，则直接移除
        if (node == tail && compareAndSetTail(node, pred)) { 
            compareAndSetNext(pred, predNext, null);
        } else { 
            int ws;
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) { //如果pred不是head节点并且是SIGNAL 状态，
                //或者可以设置为SIGNAL 状态，
                //那么将pred的next设置为node.next，也就是移除当前节点
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                    compareAndSetNext(pred, predNext, next);
            } else { 
                unparkSuccessor(node);//唤醒node的后继节点
            }
    
            node.next = node; // help GC
        }
    }
    private void unparkSuccessor(Node node) { 
        //如果waitStatus为负数，则将其设置为0（允许失败）
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
        //唤醒当前节点后面的节点。通常是紧随的next节点，
        //但是当next被取消或者为空，则从tail到node之间的所有节点，
        //往后往前查找直到找到一个waitStatus <=0的节点，将其唤醒unpark
        Node s = node.next;
        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;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

总结一下：  
1、设置thread变量为空，并且设置状态为canceled  
2、跳过中间的已经被取消的节点  
3、如果当前节点是tail节点，则直接移除。否则：  
4、如果其前驱节点不是head节点并且（前驱节点是SIGNAL状态，或者可以被设置为SIGNAL状态），那么将当前节点移除。否则通过LockSupport.unpark()唤醒node的后继节点

# 公平锁加锁 #

公平锁与非公平锁的区别就在于这里，在`tryAcquire`方法中，**首先会检查是否有任何线程等待获取的时间长于当前线程**。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 4]

public final boolean hasQueuedPredecessors() { 
        Node t = tail; 
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

很简单，就是看看AQS队列是否为空，如果不为空，那么head的下一个节点是否为当前请求的线程，如果不是，说明前面有其他线程排队，当前线程应该加入等待队列中。

# release #

public final boolean release(int arg) { 
        if (tryRelease(arg)) { //尝试释放资源 state
            Node h = head;
            if (h != null && h.waitStatus != 0)//如果AQS不为空，并且头节点的waitStatus不是0【重点】
                unparkSuccessor(h);//unpark后继节点
            return true;
        }
        return false;
    }
    
    //这里不需要加锁，因为只有获取锁的线程才会来释放锁，
    //所以这里直接将state减去releases即可
    protected final boolean tryRelease(int releases) { 
        intc = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) { 
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }
    
    private void unparkSuccessor(Node node) { 
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
        //
        Node s = node.next;
        if (s == null || s.waitStatus > 0) { 
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)//注意这里是从AQS队列的尾节点开始查找的，
            //找到最后一个 waitStatus<=0 的那个节点，将其唤醒。
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

这里的重点是，彻底释放完资源（state=0）后，会去唤醒AQS队列中的一个等待节点，该节点查找顺序为**从AQS队列的尾节点开始查找的，找到最后一个 waitStatus<=0 的那个节点**，通过`LockSupport.unpark`将其唤醒。  
这里可能会有疑问，h=head ; h.waitStatus !=0 这个判断何时成立，head初始化时waitStatus=0，waitStatus 是什么时候修改为非0的？其实上面已经列出来了，在`shouldParkAfterFailedAcquire`中，会有这么一行代码`compareAndSetWaitStatus(pred, ws, Node.SIGNAL);`，不仅head节点，所有加入AQS队列的节点的前驱节点都会被设置为SIGNAL，因为他们被park后，需要unpark才可以继续执行。  
综上，AQS队列是一个FIFO队列。

# AQS内部实际状态 #

这里模拟一种实际加锁的情况，看看AQS队列中实际状态是什么样的。  
线程伪代码如下：

new Thread(() -> { 
        try { 
            lock.lock();
            //do something
        }finally { 
            lock.unlock();
        }        
    }).start();

**假设有4个线程，并发竞争互斥锁。那么队列中有几个节点？state的值是多少？**  
测试代码如下：

ReentrantLock lock = new ReentrantLock();
    new Thread(() -> { 
        lock.lock();
        try { TimeUnit.SECONDS.sleep(100);} catch (InterruptedException e) { }
        lock.unlock();
    }, "aaa").start();
    new Thread(() -> { 
        lock.lock();
        try { TimeUnit.SECONDS.sleep(100);} catch (InterruptedException e) { }
        lock.unlock();
    }, "bbb").start();
    new Thread(() -> { 
        lock.lock();
        try { TimeUnit.SECONDS.sleep(100);} catch (InterruptedException e) { }
        lock.unlock();
    }, "ccc").start();
    new Thread(() -> { 
        lock.lock();
        try { TimeUnit.SECONDS.sleep(100);} catch (InterruptedException e) { }
        lock.unlock();
    }, "ddd").start();
    
    new Thread(() -> { 
        for(int i = 0; i < 1000 ; i ++){ 
            try { 
                System.out.println("---> " + lock.getQueueLength() + " " +  lock.getHoldCount());
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    },"eee").start();
    
    TimeUnit.SECONDS.sleep(50000);//主线程等待

运行代码，打印输出;

---> 3  0
    ---> 3  0
    ---> 3  0
    ---> 3  0

此时看一下线程快照，可以发现，bcd三个线程全部是WAITING状态，而且是由于`Unsafe.park`导致的WAITING。aaa线程已经进入了TIMED\_WAITING状态，由Thread.sleep触发的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 6]  
那么分析一下`---> 3 0`这个输出结果。  
`getQueueLength`只是简单统计一下AQS队列中的节点数量。源码如下：

public final int getQueueLength() { 
        intn = 0;
        for (Node p = tail; p != null; p = p.prev) { //从tail尾结点向前累加
            if (p.thread != null)
                ++n;
        }
        return n;
    }

再来看看getHoldCount方法。由于调用getHoldCount方法的线程是eee, 该线程不是加锁的线程，所以会返回0。不过没关系，我们可以加断点，来看一下state的值是多少。

final Thread getOwner() { 
        return getState() == 0 ? null : getExclusiveOwnerThread();
    }
    final int getHoldCount() { 
        return isHeldExclusively() ? getState() : 0;
    }
    final boolean isLocked() { 
        return getState() != 0;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 7]  
由于我们这里测试的是ReentrantLock，是互斥锁，所以state为1。

加断点看一下所有节点的真实状态，验证上面的逻辑分析。bbb,ccc,ddd三个线程追加在head节点的后面，head节点和ccc,ddd线程的waitstatus=-1，这是因为在添加到AQS队列中时会将其前驱节点设置为NODE.SIGNAL状态。  
当然多线程运行时，bbb,ccc,ddd的顺序可能会不同。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 8]  
更直观一点：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 9]

# 几个疑问 #

## AQS是FIFO的，那么他是怎么实现非公平锁的？ ##

AQS队列先进先出不假。但是仔细阅读源码，可以发现，在lock时，会首先尝试获取锁。这里就达到了一定的非公平性。思考这种情况：已经存在了AQS队列，且不为空。当获取锁的线程将state全部释放，还没有来得及唤醒后继线程时，此时新加入的线程，则会通过这里的CAS获取到锁，相当于插队了。这也是为什么公平锁的代码中没有这段代码的原因。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 10]  
另外，在tryAcquire()方法中，公平和非公平方法获取锁的区别也是在于：非公平锁不去校验AQS中是否有不是当前线程的前驱节点（hasQueuedPredecessors()）。

## 非公平锁：饥饿 ##

上面也提到了，即使AQS队列中有等待的线程，新来的线程也可能抢先获取到锁。在竞争激烈的情况下，AQS队列中的线程，可能会长时间获取不到锁，影响了正常任务执行，这就是饥饿。

## ReentrantLock性能是如何保证的？ ##

首先代码中大量使用了CAS来操作，CAS是无锁的，所以性能比较高。  
但是在竞争激烈的情况下，，CAS频繁的进行循环比较，也是非常浪费资源的，比如CPU时钟。因此在jdk代码中，有两处进行了优化：

1.  非公平锁的lock方法中，提前进行一次CAS操作，如果没有这一个其实也是可以的，但是首先进行一次CAS操作，就可以使得一部分线程可以在加入AQS队列前就获取锁，提高了性能。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 11]
2.  在入队时，首先尝试直接加入到队列末尾，因为大部分情况下直接加入队尾是OK的。这样避免了enq方法中的其他的复杂逻辑。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 12]

## ReentrantLock是通过java代码层实现的，那么其线程安全是如何保证 的？ ##

整个实现过程中，对于共享变量，通过CAS来实现。其他的变量都在线程内部，是私有变量，不会发生线程安全问题。

## 羊群效应 ##

简单描述一下羊群效应：假设AQS队列中已经存在了很多等待的线程，那么当获取锁的线程释放锁时，有一种做法是取唤醒所有等待的线程，让他们都去竞争锁，这也符合非公平锁的特征，是比较容易想到的方案。  
但是这中方案会造成巨大的资源浪费，首先是需要通知所有的线程去释放锁，然后所有的线程都去竞争锁，虽然ReentrantLock加锁主要是在java层面，但是也是需要耗费大量的CAS操作及CPU时钟，最终只有一个线程抢到锁，其他的还是乖乖回去等待，固不可取。  
ReentrantLock给出的解决方案是：维护一个 FIFO 队列，也就是AQS队列，队列中每个节点只关心其前一个节点的状态，线程唤醒也只唤醒队头等待线程。如此一来，即保证了非公平锁的特性，又大大降低了多个线程之间的竞争，提升了性能。

这种方式在《从PAXOS到ZOOKEEPER分布式一致性原理与实践》中的分布式锁一章中也有讲解，请自行查阅。

> 更多java源码分析请见：jdk11源码分析系列文章专栏：[Java11源码分析][Java11]

[Java11]: https://it007.blog.csdn.net/column/info/30895
[20200515113638671.jpg_pic_center]: /images/20220112/67abdc2af0084ebe8ea648be65b31323.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: /images/20220112/c2f389299fa243708f24b4fe56489dfe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: /images/20220112/b397343d1a1a48d4944abc6967df92e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]: /images/20220112/dcd16acac5b6482b9596a46ceaddb8a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]: /images/20220112/84e85581609c492f894f3c3c77a67b07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 4]: /images/20220112/25ff15c1e84d477798e7a08d87d9f24a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 5]: /images/20220112/1dc3e05e34b44ddca49f1eba50fa1891.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 6]: /images/20220112/d1a15bdde39b44c6a8f5e6a0a801f942.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 7]: /images/20220112/23f6dde7a5ff4b7c9cbe3ddc8315abb0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 8]: /images/20220112/8b98355a25884c23a4bb2df09de1a10d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 9]: /images/20220112/cef41a0411244c4e95170b4bcbb6510c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 10]: /images/20220112/44faf9d0030a47b4a30171cf56ff1ce2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 11]: /images/20220112/258baddb36fb498eb191964d65c6e0a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9pdDAwNy5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 12]: /images/20220112/c0092a503b9946cfb5c4189c29be6a46.png