可重入的独占锁ReentrantLock概述-公平与非公平锁的实现

Posted 2021-10-24 BLLBL

类图概述

ReentrantLock是可重入的独占锁，R最终还是AQS来实现的，根据参数决定内部是否为公平锁。

其中Sync类继承AQS，重写了AQS提供的钩子方法，比如tryAcquire方法。AQS的state状态值标识线程获取该锁的次数。

非公平的实现

非公平是说先尝试获取锁的线程不一定比后尝试获取锁的线程优先获取锁。（人话：后尝试的线程也可能拿到锁）

实现过程：首先查看state是否为0，为0代表空闲，尝试CAS获取该锁，将state从0改为1，设置当前锁的持有者为自己然后返回。

例子：线程A调用lock方法，执行到nonfairTryAcquire的代码，发现state不为0，发现当前线程不是线程持有者，则返回false，然后当前线程被放入AQS阻塞队列。

这时线程B也调用了lock方法执行到nonfairTryAcquire，发现state为0了，B通过CAS获取了锁，这就是抢夺策略。

公平的实现

hasQueuedPredecessors方法是实现公平性的核心方法。

一句话概括：查询是否有线程等待获取的时间长于当前线程。

实现过程：如果当前线程节点有前驱结点则返回true，否则如果当前AQS队列为空或者当前节点是AQS的第一个节点则返回false。（为什么新来的线程还没入队就有前驱结点？这里应该是在讨论该节点已经入队的情况，该线程入队之后，通过查询线程前面还有没有前驱结点，来判断他是否能公平的获取锁）。

ReentrantLock源码分析

概述

ReentrantLock是一个可重入的互斥锁，也被称为独占锁。它支持公平锁和非公平锁两种模式。

ReentrantLock的使用方法

下面看一个最初级的例子：

public class Test {

    //默认内部采用非公平实现
    ReentrantLock lock=new ReentrantLock();
    
    public void myMethor(){
        lock.lock();

        //需要加锁的一些操作
        
        //一定要确保unlock能被执行到，尤其是在存在异常的情况下
        lock.unlock();
    }
}

在进入方法后，在需要加锁的一些操作执行之前需要调用lock方法，在jdk文档中对lock方法详细解释如下：

获得锁。
如果锁没有被另一个线程占用并且立即返回，则将锁定计数设置为1。 如果当前线程已经保持锁定，则保持计数增加1，该方法立即返回。 如果锁被另一个线程保持，则当前线程将被禁用以进行线程调度，并且在锁定已被获取之前处于休眠状态，此时锁定保持计数被设置为1。

这里也很好的解释了什么是可重入锁，如果一个线程已经持有了锁，它再次请求获取自己已经拿到的锁，是能够获取成功的，这就是可重入锁。

在需要加锁的代码执行完毕之后，就会调用unlock释放掉锁。在jdk文档之中对，unlock的解释如下：

尝试释放此锁。
如果当前线程是该锁的持有者，则保持计数递减。 如果保持计数现在为零，则锁定被释放。 如果当前线程不是该锁的持有者，则抛出IllegalMonitorStateException 。

在这里有一个需要注意的地点，lock和unlock都反复提到了一个计数，这主要是因为ReentrantLock是可重入的。每次获取锁（重入）就将计数器加一，每次释放的时候的计数器减一，直到计数器为0，就将锁释放掉了。

以上就是最基础，最简单的使用方法。其余的一些方法，都是一些拓展的功能，查看jdk文档即可知道如何使用。

源码分析

继承体系

可以看出ReentrantLock继承自AQS并实现了Lock接口。它内部有公平锁和非公平锁两种实现，这两种实现都是继承自Sync。根据ReentrantLock决定到底采用公平锁还是非公平锁实现。

 public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

核心方法源码分析

Lock方法

1. 首先调用具体的Lock实现.sync可能是非公平锁实现也可能是公平锁实现，这取决于你new对象时的参数。

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

我们以非公平锁实现来看下面的下面的代码。

1. 非公平锁的lock方法的具体实现如下

        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

首先进来就是一个判断，其中判断的条件就是compareAndSetState(0, 1).毫无疑问这是一个CAS。它的意思时如果当前的state的值的为0就将1与其交换（可以理解为将1赋值给0）并返回true。其实在这一步如果state的值修改成功了，那么锁就获取成功了。setExclusiveOwnerThread(Thread.currentThread())这行代码就是将当前线程设置为该排他锁的拥有者。

如果CAS失败了，那么就调用acquire(1);

1. 如果初次获得锁失败就调用qcquire(1)
   这个方法的具体实现如下；

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

这个方法进来首先第一步就是调用tryAcquire(arg).
那么该方法是干什么的呢？
非公平锁实际是调用了这个实现：

        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }

它具体的实现是在nonfairTryAcquire(acquires)中。

 final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState(); //获取锁的状态state，这就是前面我们CAS的操作对象
            if (c == 0) {
                //c==0说明没被其它获取
                if (compareAndSetState(0, acquires)) { //CAS修改state
                    //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;
            }
            //锁被其它线程占有，就返回false
            return false;
        }

1. 第二次尝试获取锁失败后，就进行下一步操作
   我们再会过头看void acquire(int arg)首先尝试获取锁，获取成功就直接返回了，获取失败就会执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进行排队。
   这一行代码可以分为两部分看，一部分是addWaiter(Node.EXCLUSIVE)一部分是acquireQueued.我们先看addWaiter(Node.EXCLUSIVE)

 private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
        //队列已经初始化了，就直接入队即可
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node; //返回
            }
        }
        //队列没有初始化，初始化队列并入队
        enq(node);
        return node;
    }

初始化对立对入队的具体实现如下：

    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                //初始化队列
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
            //队列初始化成功，进行入队
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

这里稍微补充一下这个AQS中的这个等待队列。

1. 节点也创建了，等待队列也入了
   现在该看boolean acquireQueued(final Node node, int arg)方法了。
   这个方法的具体实现如下：

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor(); //获取当前节点的先驱节点
                if (p == head && tryAcquire(arg)) {
                //如果当前节点的前一个节点是头节点，就会执行tryAcquire(arg)再次尝试获取锁
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    //根据情况进入park状态
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

unlock方法

1. 和加锁类似，调用具体的实现

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

1. 具体的release实现

    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                //唤醒等待的线程，可以拿锁了
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

该方法首先就调用了tryRelease(arg)方法，这个方法就是实现释放资源的关键。释放的具体操作，也印证了在jdk文档之中的关于unlock和lock的说明。

protected final boolean tryRelease(int releases) {
            int c = getState() - releases; //计算释放后的state的值
            if (Thread.currentThread() != getExclusiveOwnerThread())
                //如果当前线程没有持有锁，就抛异常
                throw new IllegalMonitorStateException();
            boolean free = false; //标记为释放失败
            if (c == 0) {
                //如果state为0了，说没没有线程占有该锁了
                //进行重置所有者
                free = true;
                setExclusiveOwnerThread(null);
            }
            //重置state的值
            setState(c);
            return free;
        }

1. 如果还有线程在等待锁资源，那么就可以唤醒它们了
   回到boolean release(int arg)看

 if (h != null && h.waitStatus != 0)
                //唤醒等待的线程，可以拿锁了
                unparkSuccessor(h);

ReentrantLock的高阶使用方法

我们使用synchronized的时候，可以通过wait和notify来让线程等待，和唤醒线程。在ReentrantLock中，我们也可以使用Condition中的await和signal来使线程等待和唤醒。
以下面这段代码来解释：

public class Test {

    static ReentrantLock lock=new ReentrantLock();
    //获取到condition
    static Condition condition=lock.newCondition();

    public static class TaskA implements Runnable{

        @Override
        public void run() {
            lock.lock();
            System.out.println(Thread.currentThread().getName() + "开始执行");
            try {
                System.out.println(Thread.currentThread().getName() + "准备释放掉锁并等待");
                //在此等待，直到其它线程唤醒
                condition.await();
                System.out.println(Thread.currentThread().getName() + "重新拿到锁并执行");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                lock.unlock();
            }

        }
    }


    public static class TaskB implements Runnable{

        @Override
        public void run() {
            lock.lock();
            System.out.println(Thread.currentThread().getName() + "开始执行");

            System.out.println(Thread.currentThread().getName() + "开始唤醒等待的线程");
            //唤醒等待的线程
            condition.signal();
            try {
                Thread.sleep(2000);
                System.out.println(Thread.currentThread().getName() + "任务执行完毕");
            }catch (InterruptedException e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) {
       
        Thread taskA=new Thread(new TaskA(),"taskA");
        Thread taskB=new Thread(new TaskB(),"taskB");
        taskA.start();
        taskB.start();
    }
    
}

输出结果：

taskA开始执行
taskA准备释放掉锁并等待
taskB开始执行
taskB开始唤醒等待的线程
taskB任务执行完毕
taskA重新拿到锁并执行

现象解释：
首先taskA拿到锁，并执行，到condition.await();释放锁，并进入阻塞。taskB因此拿到刚才taskA释放掉的锁，taskB开始执行。taskB执行到condition.signal();唤醒了taskA，taskB继续执行，taskA因为拿不到锁，因此虽然已经被唤醒了，但是还是要等到taskB执行完毕，释放锁后，才有机会拿到锁，执行自己的代码。

那么这个过程，源码到底是如何实现的呢？

Condition源码分析

await()的源码分析

具体的实现如下：

  public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter(); //添加一个条件节点
            int savedState = fullyRelease(node); //释放掉所有的资源
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
            //如果当前线程不在等待队列中，park阻塞
                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);
        }

基本的步骤如下：

1. 首先判断线程是否被中断，如果中断则抛出InterruptedException()异常
2. 添加当前线程到条件队列中去，然后释放掉所有的资源
3. 如果当前线程不在等待队列中，就直接park阻塞当前线程

signal()方法源码分析

具体的实现代码如下：

        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }

这个方法中最重要的也就是doSignal(first).
它的实现如下：

        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null; //解除等待队列中首节点的链接
            } while (!transferForSignal(first) && //转移入等待队列
                     (first = firstWaiter) != null);
        }

该方法所做的事情就是从等待队列中移除指定节点，并将其加入等待队列中去。
转移节点的方法实现如下：

    final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            //CAS修改状态失败，说明节点被取消了，直接返回false
            return false;

        /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */
        Node p = enq(node); //加入节点到等待队列
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        //如果前节点被取消，说明当前为最后一个等待线程，直接unpark唤醒，
            LockSupport.unpark(node.thread);
        return true;
    }

至此ReentrantLock的源码分析就结束了！