Java并发编程之ReentrantLock源码分析

Posted 2021-02-01 夜里码码

ReentrantLock介绍

从JDK1.5之前，我们都是使用synchronized关键字来对代码块加锁，在JDK1.5引入了ReentrantLock锁。synchronized关键字性能比ReentrantLock锁要差，而且ReentrantLock锁功能要比synchronized关键字功能强大。

特点

synchronized关键字和ReentrantLock锁都是重入锁，可重入锁是指当一个线程获取到锁后，此线程还可继续获得这把锁，在此线程释放这把锁前其他线程则不可获得这边锁。相比synchronized关键字，ReentrantLock锁具有锁获取超时和获取锁响应中断的特点。ReentrantLock锁还分公平锁和非公平锁，公平锁模式是按线程调用加锁的先后排队顺序获取锁，非公平锁模式是已经在排队中的线程按顺序获取锁，但是新来的线程会和排队中的线程进行竞争，并不保证先排先获取锁。

ReentrantLock 源码分析

ReentrantLock实现了java.util.concurrent.locks.Lock接口和java.io.Serializable接口，前者是对实现Java锁的一种规范，后者说明ReentrantLock可以序列化。 ReentrantLock定义了一个成员变量

private final Sync sync;

Sync类型是ReentrantLock的内部类，继承至AbstractQueuedSynchronizer ，AbstractQueuedSynchronizer是一个带空头的双向列表，为ReentrantLock的锁排队提供了基础支持。 ReentrantLock的UML关系图如下

下面我们解析下ReentrantLock中几个常用方法。

lock()方法源码分析

lock()是ReentrantLock中最常用的方法，用来对代码块加锁。lock()先是调用Sync的lock()的方法，Sync#lock()实现分为非公平模式和公平模式，我们对这2个模式分别讲解

非公平模式

Sync#lock()非公平模式代码如下：

        final void lock() {
        //用CAS方法设置枷锁状态
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
        //抢锁失败，进入后续逻辑。
                acquire(1);
        }

新来线程先调用compareAndSetState(0, 1)方法用CAS方法设置加锁状态，这里是非公平模式实现要点，这样做主要是为了新来的线程和排队中的线程竞争，排队中的线程激活后也会用CAS方法设置加锁状态，就是看哪个线程线程抢的快，哪个能拿到锁。如果设置加锁状态成功，则设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果设置加锁状态失败即抢锁失败，则调用acquire(1)进入排队逻辑。

AbstractQueuedSynchronizer#acquire(int arg)实现代码如下：

    public final void acquire(int arg) {
    //先调用tryAcquire(arg)再试下能不能获取到锁，无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

先调用tryAcquire(arg)再试下能不能获取到锁，获取成功则执行结束，无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队，此方法返回参数为是否中断当前线程，排队过程中如果线程被中断则会返回ture，此时调用selfInterrupt()中断当前线程。

tryAcquire(arg)直接调用了非公平模式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;
                }
            }
            //判断是否是相同线程，如果是则表示当前线程的锁重入了
            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;
        }

调用getState()方法获取加锁状态，如果为0表示当前未被加锁，尝试CAS设置加锁状态获取锁，如果成功同样设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果已被加锁，这判断当前线程和加锁线程是否是同一线程，如果是同一线程则将获取锁的状态加1返回获取锁成功，这里就是可重入锁实现的核心，状态的值表示当前线程重入了多少次，之后的释放锁就要释放相同的次数。

接下来我们看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法，acquireQueued主要功能是对当前线程阻塞，阻塞到能被上个获取到锁线程释放为止，addWaiter(Node.EXCLUSIVE)则是将当前线程加入到排队队列中。 我们先来看下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;
        //CAS快速添加节点到尾部
        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 {
            //CAS添加节点到尾部
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    

创建已当前线程为基础的节点，先走快速添加到尾部逻辑，获取尾节点如果尾节点存在，将当前节点和尾节点相连，并用CAS方式将当前节点设置为尾节点，这边使用CAS方式考虑了多个线程同时操作尾节点的情况，所以如果尾节点已经变更则快速添加节点操作失败，调用enq(node)方法走最大努力添加节点的逻辑。enq(node)最大努力添加逻辑就是一直添加节点直到添加节点到尾部成功。

下面看下acquireQueued(addWaiter(Node.EXCLUSIVE), 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)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

acquireQueued里有个循环，这个循环的主要作用就是在线程激活后重试获取锁直到获取锁。node.predecessor()获取当前线程节点的前一个节点，如果是头节点，则当前线程尝试获取锁，获取锁成功设置当前节点为头节点。如果获取失败或者非头节点则调用shouldParkAfterFailedAcquire(p, node)判断是否需要阻塞等待，如果需要阻塞等待则调用parkAndCheckInterrupt()阻塞当前线程并让出cup资源资质被前一个节点激活，继续循环逻辑。

我们先来看下shouldParkAfterFailedAcquire(p, node)的实现

 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            
            return true;
        if (ws > 0) {
             
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
           
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

先获取前个节点的状态，状态分以下4类

static final int CANCELLED =  1;
static final int SIGNAL    = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;

除了CANCELLED关闭状态是非正常，其他状态均正常状态。判断当前状态是否是SIGNAL正常状态，如果是就返回成功，这样当前线程就可以阻塞安心的等待上个节点的激活。如果状态为CANCELLED关闭状态则删除所有当前节点之前状态为CANCELLED的节点，返回失败让当前线程重试获取锁，如果是初始化0状态则CAS方式设置状态为SIGNAL。

接下来看下阻塞方法parkAndCheckInterrupt()

private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

方法很简单调用LockSupport.park(this)阻塞当前线程，这里要讲下方法返回时调用Thread.interrupted()判断当前线程是否被中断，如果被中断的话，当前线程获取到锁后会调用Thread.currentThread().interrupt()中断线程。

公平模式

公平模式和非公平模式大部分代码相同，主要是获取锁的逻辑不同，我们就讲下代码不同的部分 lock()代码如下

final void lock() {
       acquire(1);
}

非公平模式模式先尝试设置状态来获取锁，而公平模式则直接调用acquire(1)去走排队逻辑。

尝试获取锁的方法tryAcquire(int acquires)也不一样代码如下

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;
}

该方法跟非公平锁基本都一样，只是在获取锁的时候加了hasQueuedPredecessors()判断，这个方法主要判断了当前线程是否在头节点的下个节点，这样保证了获取锁的顺序性。

unlock()方法源码分析

unlock()方法比较简单，直接调用sync.release(1)方法。 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;
    }

先尝试释放锁，如果释放产品这判断当前节点是否为0不为0调用unparkSuccessor(h)方法激活下个节点的线程，否则直接返回。这里会有个疑问为什么h.waitStatus为0不去激活下个节点的线程，如果不激活下个节点的线程是否一直阻塞的，答案是否定的。这样做主要是为了释放锁的效率。waitStatus为0是初始化的值，这个值还没被下个节点线程调用shouldParkAfterFailedAcquire(p, node)方法设置成SIGNAL状态，也就说明下个节点线程还没被阻塞，此时如果下个节点线程调用此方法并设置成SIGNAL状态，势必它会重新获取锁，从而获取到锁避免了上述的问题。

下面来看下tryRelease(arg)方法

 protected final boolean tryRelease(int releases) {
        //重入次数减1
            int c = getState() - releases
        //非持有线程抛异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
        //如果释放了所有的重入次则清理持有线程为空
            if (c == 0) {
                free = true;
                setExclusiveOwnerThread(null);
            }
        //设置当前剩余的重入次数
            setState(c);
            return free;
        }

因为锁可重入，因此调用getState()获取状态的值并减去一次重入次数，得到的c就是剩余重入的次数，然后判断当前释放的线程是否是当前占有锁的线程，如果不是抛出异常，否则先判断c是否为0表示当前线程持有的锁是否释放完全，如果是则设置持有锁的线程的变量为空，并设置锁状态为0，否则设置剩余的c到锁的状态。

接下来看下unparkSuccessor(h)的实现

  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)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            //激活线程
            LockSupport.unpark(s.thread);
    }

获取当前节点状态，设置如果当前节点正常情况则设置成0，然后取当前节点的下个节点，如果下个节点状态非正常即CANCELLED状态，则从队列的尾部开始查找查到最靠近当前的节点且状态正常的节点，然后调用LockSupport.unpark(s.thread)通知此节点停止阻塞。这边会有个疑问如果调用LockSupport.unpark(s.thread)方法后，此节点才调用LockSupport.park(this)去阻塞，这样会不会发生此节点永久阻塞的问题，答案是否定的，LockSupport.unpark(s.thread)方法的实现其实是为线程设置了一个信号量，LockSupport.park(this)就算后调，如果线程相同也会收到此信号从而激活线程，这里的实现原理就不展开讲。

原文链接：https://my.oschina.net/u/945573/blog/2991876