ReentrantLock详解

starlin 1,052 2018-07-09

ReentrantLock介绍

ReentrantLock可重入锁,是一种递归无阻塞的同步机制。它可以等同于synchronized的使用,但是ReentrantLock提供了比synchronized更强大,更灵活的锁机制,可以减少死锁发生的几率。API介绍如下:

一个可重入的互斥锁定 Lock,它具有与使用 synchronized 方法和语句所访问的隐式监视器锁定相同的一些基本行为和语义,但功能更强大。ReentrantLock 将由最近成功获得锁定,并且还没有释放该锁定的线程所拥有。当锁定没有被另一个线程所拥有时,调用 lock 的线程将成功获取该锁定并返回。如果当前线程已经拥有该锁定,此方法将立即返回。可以使用 isHeldByCurrentThread() 和 getHoldCount() 方法来检查此情况是否发生。

ReentrantLock还提供了公平锁和非公平锁,构造方法接受一个可选的公平参数(默认为非公平锁),当设置为true时,表示公平锁。
公平锁和非公平锁的在于获取锁是否有顺序,公平锁获取锁的顺序是有序的,但是公平锁的效率往往没有非公平锁的效率高,在高并发的情况下,公平锁表现出较低的吞吐量
ReentrantLock

从上图我们可以做如下总结:

  • ReentrantLock实现了Lock,Serializable接口
  • ReentrantLock.Sync(内部类)继承了AQS
  • ReentrantLock.NonfairSync和ReentrantLock.FairSync继承了ReentrantLock.Sync
  • ReentrantLock持有ReentrantLock.Sync对象(实现锁功能)

ReentrantLock 的内部实现

先总体描述下 ReentrantLock 的大致实现,有一个成员属性 sync,所有的方法都是调用该属性的方法。Sync 继承 AbstractQueuedSynchronizer(简称 AQS),AQS 封装了锁和线程等待队列的基本实现。Sync 有两个子类 NonfairSync 和 FairSync,分别对应非公平锁和公平锁。AQS 内部使用volatile int state表示同步状态,在 ReentrantLock 中 state 表示占有线程对锁的持有数量,为 0 表示锁未被持有,为 1 表示锁被某个线程持有,> 1 表示锁被某个线程持有多次(即重入)。

非公平锁

下面我们看非公平锁的lock()方法:

    final void lock() {
    //尝试获取锁
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
        //获取失败,调用AQS的acquire(int arg)方法
            acquire(1);
    }

首先会第一次尝试快速获取锁,如果获取失败,则调用acquire(int arg)方法,该方法定义在AQS中,如下:

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

这个方法首先调用tryAcquire(int arg)方法,在AQS中讲述过,tryAcquire(int arg)需要自定义同步组件提供实现,非公平锁实现如下:

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

    final boolean nonfairTryAcquire(int acquires) {//这里和公平锁唯一的区别是少了一个判断(判断队列中是否为空)
            //当前线程
            final Thread current = Thread.currentThread();
            //获取同步状态
            int c = getState();
            if (c == 0) {
                //如果锁未被持有,则直接获取
                //这里可能为其他线程刚刚释放锁,还有其他线程在等待,但这时直接获取,所以是不公平的
                //获取锁成功,设置为当前线程所有
                if (compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
           
            //判断锁持有的线程是否为当前线程
            //若锁被当前线程持有,属于重入,state ++
            else if (current == getExclusiveOwnerThread()) {
                int nextc = c + acquires;
                //如果 state > 2 ^ 31 - 1, 则抛出异常,这也是最大重入次数
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
    }

所以非公平锁的 lock() 的大致逻辑为:如果锁未被持有,不管等待队列中的线程直接获取;如果锁被自己(当前线程)持有,则把 state 加 1;否则将当前线程加入到等待队列中,并阻塞该线程直到获取成功。

公平锁

我们来看看公平锁的lock方法

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

公平锁的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() &&//1判断同步队列中是否有节点在等待
                    compareAndSetState(0, acquires)) {//2如果1成立,修改state值(表名当前锁被占用)
                    setExclusiveOwnerThread(current);//如果2成立,修改当前占用锁的线程为当前线程
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) {// // 锁被当前线程持有,属于重入,state ++
                int nextc = c + acquires;
                if (nextc < 0)
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;//获取线程锁失败
        }

和上面的非公平锁比较,发现公平锁多了一个hasQueuedPredecessors限制条件,其源码如下:

    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;
        //头节点 != 尾节点
        //同步队列第一个节点不为null
        //当前线程是同步队列第一个节点
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

它用来查看队列中是否有比它等待时间更久的线程,如果没有,就尝试一下是否能获取到锁,如果获取成功,则标记为已经被占用。如果获取锁失败,则调用 addWaiter 方法把线程包装成 Node 对象,同时放入到队列中,但 addWaiter 方法并不会尝试获取锁,acquireQueued 方法才会尝试获取锁,如果获取失败,则此节点会被挂起,源码如下:

    /**
     * 队列中的线程尝试获取锁,失败则会被挂起
     */
    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)) {
                    // 获取成功,将当前节点设置为 head 节点
                    setHead(node);
                    //原 head 节点出队,等待被 GC
                    p.next = null; // help GC
                    // 获取成功
                    failed = false;
                    return interrupted;
                }
                // 判断获取锁失败后是否可以挂起
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    // 线程若被中断,返回 true
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

该方法会使用 for(;;) 无限循环的方式来尝试获取锁,若获取失败,则调用 shouldParkAfterFailedAcquire 方法,尝试挂起当前线程,源码如下:

    /**
     * 判断线程是否可以被挂起
     */
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        // 获得前驱节点的状态
        int ws = pred.waitStatus;
        // 前驱节点的状态为 SIGNAL,当前线程可以被挂起(阻塞)
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            // 若前驱节点状态为 CANCELLED,那就一直往前找,直到找到一个正常等待的状态为止
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            // 并将当前节点排在它后边
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            // 把前驱节点的状态修改为 SIGNAL
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

线程入列被挂起的前提条件是,前驱节点的状态为 SIGNAL,SIGNAL 状态的含义是后继节点处于等待状态,当前节点释放锁后将会唤醒后继节点。所以在上面这段代码中,会先判断前驱节点的状态,如果为 SIGNAL,则当前线程可以被挂起并返回 true;如果前驱节点的状态 >0,则表示前驱节点取消了,这时候需要一直往前找,直到找到最近一个正常等待的前驱节点,然后把它作为自己的前驱节点;如果前驱节点正常(未取消),则修改前驱节点状态为 SIGNAL。

公平锁的大致逻辑为:如果锁未被持有,并且当前线程在等待队列的头部或者等待队列为空,则获取锁;如果锁被自己(当前线程)持有,则把 state 加 1;否则将当前线程加入到等待队列中,并阻塞该线程。

加锁过程示意图

整个加锁的流程就已经走完了,最后的情况是,没有拿到锁的线程会在队列中被挂起,直到拥有锁的线程释放锁之后,才会去唤醒其他的线程去获取锁资源,整个运行流程如下图所示:
![ReentrantLock加锁示意图](https://raw.githubusercontent.com/smartlin/pic/main/_posts/java%E5%B9%B6%E5%8F%91/java%E5%B9%B6%E5%8F%91%E4%B9%8Breentrantlock.md/ReentrantLock%E5%8A%A0%E9%94%81%E7%A4%BA%E6%84%8F%E5%9B%BE.png =866x)

![非公平锁加锁流程](https://raw.githubusercontent.com/smartlin/pic/main/_posts/java%E5%B9%B6%E5%8F%91/java%E5%B9%B6%E5%8F%91%E4%B9%8Breentrantlock.md/%E9%9D%9E%E5%85%AC%E5%B9%B3%E9%94%81%E5%8A%A0%E9%94%81%E6%B5%81%E7%A8%8B.png =996x)

![非公平锁加锁流程1](https://raw.githubusercontent.com/smartlin/pic/main/_posts/java%E5%B9%B6%E5%8F%91/java%E5%B9%B6%E5%8F%91%E4%B9%8Breentrantlock.md/%E9%9D%9E%E5%85%AC%E5%B9%B3%E9%94%81%E5%8A%A0%E9%94%81%E6%B5%81%E7%A8%8B1.png =996x)

![非公平加锁流程2](https://raw.githubusercontent.com/smartlin/pic/main/_posts/java%E5%B9%B6%E5%8F%91/java%E5%B9%B6%E5%8F%91%E4%B9%8Breentrantlock.md/%E9%9D%9E%E5%85%AC%E5%B9%B3%E5%8A%A0%E9%94%81%E6%B5%81%E7%A8%8B2.png =996x)

公平锁和非公平锁区别

区别点lock过程tryAcquire过程
FairSync直接acquire()当前若无线程持有锁,如果同步队列为空,获取锁
NonFairSync先尝试获取锁,再acquire()当前若无线程持有锁,获取锁

释放锁

![非公平锁释放锁](https://raw.githubusercontent.com/smartlin/pic/main/_posts/java%E5%B9%B6%E5%8F%91/java%E5%B9%B6%E5%8F%91%E4%B9%8Breentrantlock.md/%E9%9D%9E%E5%85%AC%E5%B9%B3%E9%94%81%E9%87%8A%E6%94%BE%E9%94%81.png =996x)

ReentrantLock提供了释放锁的方法unlock,锁的释放流程为,先调用 tryRelease 方法尝试释放锁,如果释放成功,则查看头结点的状态是否为 SIGNAL,如果是,则唤醒头结点的下个节点关联的线程;如果释放锁失败,则返回 false。

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

unlock内部使用了sync的release方法释放锁,release方法是在AQS中定义的

   public final boolean release(int arg) { 
   // 尝试释放锁       
   if (tryRelease(arg)) {
            // 释放成功
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

与获取同步状态的acquire(int arg)方法相似,释放同步状态的tryRelease(int arg)同样是需要自定义同步组件自己实现

    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;// 释放锁后的状态,0 表示释放锁成功
        //如果释放的不是持有锁线程,抛出异常
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
         //state == 0 表示已经释放完全了,其他线程可以获取同步状态了
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);// 清空独占线程
        }
        // 更新 state 值,0 表示为释放锁成功
        setState(c);
        return free;
    }

在 tryRelease 方法中,会先判断当前的线程是不是占用锁的线程,如果不是的话,则会抛出异常;如果是的话,则先计算锁的状态值 getState() - releases 是否为 0,如果为 0,则表示可以正常的释放锁,然后清空独占的线程,最后会更新锁的状态并返回执行结果。

ReentrantLock与synchronized的区别

  • ReentrantLock还提供了条件Condition,对线程的等待、唤醒操作更加详细和灵活
  • ReentrantLock提供了可轮询的锁请求。它会尝试着去获取锁,如果成功则继续,否则可以等到下次运行时处理,而synchronized则一旦进入锁请求要么成功要么阻塞,所以相比synchronized而言,ReentrantLock会不容易产生死锁些
  • ReentrantLock支持更加灵活的同步代码块,但是使用synchronized时,只能在同一个synchronized块结构中获取和释放。注:ReentrantLock的锁释放一定要在finally中处理,否则可能会产生严重的后果
  • ReentrantLock支持中断处理,且性能较synchronized会好些

Demo

public class ReentrantLockTest extends Thread{
    public static ReentrantLock reentrantLock = new ReentrantLock();
    public static int i = 0;

    @Override
    public void run() {
        for (int j = 0; j < 10000; j++) {
            reentrantLock.lock();
            try {
                i++;
            } catch (Exception e) {

            }finally {
                reentrantLock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReentrantLockTest reentrantLock1 = new ReentrantLockTest();
        ReentrantLockTest reentrantLock2 = new ReentrantLockTest();

        reentrantLock1.start();
        reentrantLock2.start();
        reentrantLock1.join();
        reentrantLock2.join();
        System.out.println(i);
    }
}

上面这段代码输出位20000,如果去掉锁,则随机输出一个小于20000的数字


# java并发