Java并发：ReentrantLock源码分析

向右看齐 2022-05-24 08:48 203阅读 0赞

注：JDK1.7  
对照源码效果更佳∠( ᐛ 」∠)\_

> 相关知识：队列同步器AbstractQueuedSynchronizer是用来构建锁或者其他同步组件的基础框架，它使用一个int变量（stats）来表示同步状态，并使用一个内置的FIFO队列来完成线程对资源的获取排序。

首先先来看下ReentrantLock的继承关系。  
![这里写图片描述][70]  
可知ReentrantLock实现了Lock接口，并包含了一个成员变量sycn（ReentrantLock与Sycn为组合关系），Sycn继承了AQS，NonfairSync和FairSync都是Sycn子类。

ReentrantLock的常见用法例子，可以看到，ReentrantLock通过lock()和unlock()来给代码片段实现同步，和`synchronized(x){...}`实现的功能是一样的，但是实现的方式不同。虽然经过JDK1.6后synchronized的性能有了很大的提升，但是一般来讲ReentrantLock比synchronized使用上更加灵活，功能也更加强大。

public class ReentrantLockTest { 
        public static void main(String[] args){
            ReentrantLock reentrantLock = new ReentrantLock();
            Thread thread1 = new Thread(new myRun(reentrantLock),"t1");
            Thread thread2 = new Thread(new myRun(reentrantLock),"t2");
            Thread thread3 = new Thread(new myRun(reentrantLock),"t3");
    
            thread1.start();
            thread2.start();
            thread3.start();
        }
    
        public static class myRun implements Runnable{ 
            private ReentrantLock reentrantLock;
            myRun(ReentrantLock param){
                reentrantLock = param;
            }
            @Override
            public void run() {
                // TODO Auto-generated method stub
                try{
                    reentrantLock.lock();
                    //....................................
                    for(int i=0;i<20;i++){
                        System.out.println(Thread.currentThread().getName()
                                +":"+i);
                    }
                }finally{
                    reentrantLock.unlock();
                }
            }
        }
    }

--------------------

# ReentrantLock.lock() #

首先让我们从lock开始看起。  
因为ReentrantLock分为公平锁和非公平锁，我们这里先看公平锁。

## FairLock ##

**lock()方法**  
AbstractQueuedSynchronizer通过一个int变量(stats)来表示同步状态（0为未被同步，1为被一个线程初次同步），所以这里`acquire(1)`可以看为当前线程尝试去获取锁。

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

`acquire`为AbstractQueuedSynchronizer中的方法。

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

这里一共有`tryAcquire(arg)`，`addWaiter`，`acquireQueued`和`selfInterrupt`4个方法。  
概括来讲：  
1.首先`tryAcquire`会尝试去获取锁，如果获取成功则返回true，此时if语句结束，完成加锁；如果获取失败，返回false，此时执行if语句的第二个条件语句中的`addWaiter`。  
2.顾名思义，`addWaiter`方法会将当前线程包装后存入AQS内置的FIFO队列中，并返回包装后的节点。  
3.`acquireQueued`中有一个死循环`for(;;)`用来继续不停尝试获取锁，获取失败后符合阻塞条件时将会阻塞当前线程，直到其他线程唤醒或者中断唤醒；获取成功时返回当前线程的中断状态，这里获取成功有一个前提，就是当前阻塞的线程被唤醒，而唤醒有两个可能的原因：1因为中断被唤醒，2因为被其他线程的unpark方法唤醒。为1时返回true，2时返回false，让我们返回到if语句中，可以看到，当为1情况时，需要执行`selfInterrupt`，因为1时当前线程被唤醒的原因是中断，所以这里会执行`selfInterrupt`方法中断当前线程。  
4.`selfInterrupt`方法会将当前线程中断。

下面来看看这4个方法的源码  
**tryAcquire方法**  
tryAcquire方法通过`hasQueuedPredecessors`方法来检测当前线程是否为队列中的**第一个等待获取锁的线程**。因为队列为FIFO先进先出，这表明进入队列的时间越早，其节点排序越靠前，该方法保证了锁的获取顺序为线程进入队列的先后顺序，这就使得每个线程都能公平的获取锁。

protected final boolean tryAcquire(int acquires) {
        //保存当前线程
        final Thread current = Thread.currentThread();
        //获取stats状态，AQS方法
        int c = getState();
        //如果锁还未被获取
        if (c == 0) {
            //hasQueuedPredecessors为判断当前等待队列中的第二个线程
            //是否为当前线程(队列中的第一个线程为已获取锁的线程，即之
            //后的线程都是在等待获得锁)，当不是时返回true。
            //compareAndSetState为原子操作，检查stats是否为0，是的
            //话将其修改为acquires，并返回true
            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");
            //原子操作，修改stats值
            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.
        //AQS内置FIFO队列的尾节点
        Node t = tail; // Read fields in reverse initialization order
        //AQS内置FIFO队列的头节点，即为当前获得锁的线程节点
        Node h = head;
        Node s;
        return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
        }

**addWaiter方法**  
Node即为AQS内置FIFO队列的节点，其包装了线程和状态值。  
![这里写图片描述][70 1]  
功能为创建当前线程的包装节点并放入队列的尾部，返回创建的节点。

private Node addWaiter(Node mode) {
        //将当前线程包装为一个Node节点
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        //tail为AQS内置FIFO队列的尾节点
        Node pred = tail;
        //如果尾节点不为空（即当前队列不为空）
        if (pred != null) {
            //将当前线程的包装节点的前置节点置为队列的尾节点（即要将当前线程的包
            //装节点置为队列尾节点）
            node.prev = pred;
            //原子操作，检查当前队列的尾节点是否为之前获取的尾节点，是则将尾节点
            //变量置为当前线程的包装节点，并返回true
            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;
                }
            }
        }
    }

**acquireQueued方法**

final boolean acquireQueued(final Node node, int arg) {
        //是否获取成功，成功为false
        boolean failed = true;
        try {
            //中断状态，中断过为true
            boolean interrupted = false;
            //无限循环来尝试获取锁
            for (;;) {
                //获取当前节点的前置节点
                final Node p = node.predecessor();
                //1.如果前置节点为队列头节点
                //表明只有第一个等待获取锁的节点才能尝试去获取锁（因为队列的
                //头节点为获得锁的线程节点），这里如果当前节点的前置节点为队
                //列的头节点的话就表明当前节点为第一个等待获取锁的节点了。
                //2.尝试获取锁，成功则返回true
                if (p == head && tryAcquire(arg)) {
                    //将当前节点置为队列头结点，表示当前节点的线程获得了锁
                    //也间接表明了队列头节点为获取锁的节点
                    setHead(node);
                    //头节点出队列，方便垃圾处理机制去清理掉它
                    p.next = null; // help GC
                    //成功获取锁
                    failed = false;
                    //返回中断状态
                    return interrupted;
                }
                //如果没有获取锁
                //1.shouldParkAfterFailedAcquire为检查当前线程是否需要阻塞
                //2.如果需要阻塞则执行parkAndCheckInterrupt方法进行阻塞（park）
                // 返回Thread.interrupted()，返回线程中断状态并消除中断状态
                // 说明：如果阻塞被唤醒是因为线程中断则返回true,如果是其他线程
                //执行unpark方法唤醒的，则返回false
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    //中断状态置为true
                    interrupted = true;
            }
        } finally {
            if (failed)
            cancelAcquire(node);
        }
    }
    //获取当前节点的前置节点
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }
    
    //检查当前线程是否需要阻塞
    //1.Node.SIGNAL = -1 ,后继节点的线程处于等待状态，而当前节点的线程如果释放了
    //同步状态或者被取消，将会通知后继节点，使后继节点的线程得以运行。
    //2.Node.CANCELLED = 1 ,在同步队列中等待的线程等待超时或者被中断，需要从同步
    //队列中取消等待，节点进入该状态将不会变化。
    //3.Node.CONDITION = -2 ,节点线程等待在condition上。
    //4.Node.PROPAGATE = -3 ,下一次共享式同步状态将会无条件的被传播下去。
    //5.Node.INITIAL = 0 ,初始状态
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //前置节点的状态
        int ws = pred.waitStatus;
        //参照上面
        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. */
            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. */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    //将当前线程阻塞，并在被唤醒时返回中断状态（中断唤醒时为true,其他线程unpark
    //唤醒时为false）,且interrupted()方法在返回中断状态后会将线程的中断状态消除
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

**cancelAcquire方法**  
在`acquireQueued`中，不管是因为成功获取锁返回还是因为中断退出，最后都会根据锁是否获取来判断是否执行`cancelAcquire`方法。  
当获取失败（中断，异常等等）情况下，会执行`cancelAcquire`。  
因为涉及的`unparkSuccessor`也会出现在`unlock方法`中，所以详细的分析放入下一篇unlock篇中。  
这里只需要知道在`acquireQueued`如果出现异常导致锁获取失败会执行`cancelAcquire`进行节点的删除和状态的清理。

private void cancelAcquire(Node node) {
        // Ignore if node doesn't exist
        if (node == null)
            return;
        //首先将thread置空，同时表明该node已经没用了
        node.thread = null;
    
        // Skip cancelled predecessors
        //跳过waitStats = 1 的节点，即跳过已经没用的节点，找到最近的还有用的前置
        //节点（waitStats < 1）,赋值给pred 
        Node pred = node.prev;
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;
    
        // predNext is the apparent node to unsplice. CASes below will
        // fail if not, in which case, we lost race vs another cancel
        // or signal, so no further action is necessary.
        //获取最近的还有用的前置节点的后继节点
        Node predNext = pred.next;
    
        // Can use unconditional write instead of CAS here.
        // After this atomic step, other Nodes can skip past us.
        // Before, we are free of interference from other threads.
        //将当前节点状态置为 1 ，表明该节点已经取消等待
        node.waitStatus = Node.CANCELLED;
    
        // If we are the tail, remove ourselves.
        //当前节点是否为尾节点，是则将尾节点重置为最近的有用的前置节点
        if (node == tail && compareAndSetTail(node, pred)) {
            //因为此时pred已经被设置成尾节点，所以将next指针置空
            compareAndSetNext(pred, predNext, null);
        } else {
            //如果不为尾节点
            // If successor needs signal, try to set pred's next-link
            // so it will get one. Otherwise wake it up to propagate.
            int ws;
            //1.最近的有用的前置节点不为头结点
            //2.最近的有用的前置节点状态等于-1，赋值给ws
            //3.最近的有用的前置节点状态小于等于0 并且将最近的有用的前置节点状态
            // 置为-1
            //4.最近的有用的前置节点的线程不为空，即该节点获取锁过程中还没有执行
            // 过该方法
            //2，3表示将pred的状态设置为-1
            //确认pred 不为头结点，pred 状态为-1，pred 的线程不为空时，检查当前
            //节点node的后置节点，如果也不为空且状态为有用，则将当前节点出队，完
            //成对因中断，异常导致取消等待的线程节点的出队
            if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
            (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                    compareAndSetNext(pred, predNext, next);
            } else {
                //唤醒等待队列中的线程
                //进入的条件为
                //1. 最近的有用的前置节点 为头节点，我们知道头节点为获取锁的线程
                //节点，这时表明当前节点与头节点之间已经不存在有用的节点了，所以
                //需要执行唤醒操作。
                //2. 最近的有用的前置节点 的线程为空，表明 最近的有用的前置节点 
                //应该也经历过该方法，但是还没有被去除，唤醒下一个线程后会在
                //shouldParkAfterFailedAcquire中重新调整队列。
    
                //此时node的waitStats = Node.CANCELLED = 1
                //unparkSuccessor为唤醒当前节点后最近的有用（waitStats<1）的节
                //点
                unparkSuccessor(node);
            }
            node.next = node; // help GC
        }
    }

**selfInterrupt方法**  
将线程中断  
执行条件为当第一次尝试获取锁失败，之后再次循环尝试获取锁时成功获取锁，但是线程中断状态为true。  
结合上面的`parkAndCheckInterrupt()`方法可知，线程被唤醒的原因为中断，而`interrupted()`方法在返回中断状态后还会将线程的中断状态消除，所以在这里需要重新给线程加上中断状态。

private static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }

下图为公平锁的lock实现流程。（没装制图软件还真是头大）  
![这里写图片描述][70 2]

总结下：  
1.一个新线程请求获得锁。  
2`.tryAcquire(1)`尝试获得锁，如果成功，则返回true，退出if判断，获取锁完毕；如果获取失败，进入if第二个条件判断。  
3.2失败时`addWaiter`，将当前线程包装为一个Node节点，并放入队列尾部。  
4.`sequireQueued`，在`for(;;)`中再次尝试获取锁，成功时获得锁并将原队列头出队，将自己置为头结点，返回中断情况（默认为false）；失败时继续尝试获得锁。过程中当满足条件时阻塞线程（一般而言都是在第二次尝试时满足条件），此时线程等待唤醒，唤醒后返回中断情况，如果是因中断唤醒，则将中断情况置为true，继续尝试获取锁。  
5.获得锁，判断返回的中断情况，true时执行线程中断操作，false时完成锁获取。

--------------------

## NonFairLock ##

事实上ReentrantLock非公平锁在实现上与公平锁十分相似，甚至可以说是一模一样。  
首先来看下`lock`实现  
**lock方法**

final void lock() {
        //绕过队列，直接去尝试获取锁
        //原子操作，当stats为0时置为1并返回true
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

与公平锁的`lock`实现对比下

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

我们可以看到非公平锁会直接去尝试获取锁，这意味着会先绕过等待队列，这对于先进入队列等待获取锁的线程而言无疑是不公平的。因此，公平锁这里就没有前面那一段内容

**acquire方法**  
`acquire`与公平锁中的相同。  
1.tryAcquire尝试获取锁。（**与公平锁不同**）  
2.addWaiter将当前线程包装为node节点放入队列尾部。  
3.acquireQueued将会持续的尝试获取锁，当条件符合时阻塞当前线程。  
4.selfInterrupt中断线程。

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

**tryAcquire方法**  
使用了Sync类中的nonfairTryAcquire方法  
尝试获取锁

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

**nonfairTryAcquire方法**  
可以对比下公平锁中的实现

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

公平锁`tryAcquire`

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

可以发现非公平锁与公平锁的`tryAcquire`实现区别只是是否包含`!hasQueuedPredecessors()`，回顾下可知该方法将会判断当前线程是否为队列中第一个等待获取锁的节点（因为队列头是已经获取锁的线程节点，所以这里指队列中的第二个节点，即第一个等待获取锁的节点）。  
在其他部分与公平锁一样，所以这里也就不再赘述。

--------------------

# ReentrantLock.unlock() #

调用了Sync父类AbstractQueuedSynchronizer类的`release方法`

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

**release**  
1.`tryRelease`尝试进行解锁。  
2.如果解锁成功则进行队列中等待线程的唤醒操作`unparkSuccessor`  
经过上一篇对`lock`的源码分析`shouldParkAfterFailedAcquire`和`cancelAcquire`，我们可以知道Node.waitStats的状态只出现过-1，1，0（初始状态）三种，置于-2（与condition有关），-3（与共享式有关）则暂时还未出现过，那么在这里我们就先排除掉-2，-3来分析。  
头节点代表的是获取锁的节点（虽然thread已经=null了），在`tryRelease`中只是对锁状态进行了更新，却没有对队列进行更新，可知此时获取锁的线程节点（头节点）还是上一个线程节点。因此需要对队列进行更新，并唤醒下一个等待线程。  
`if (h != null && h.waitStatus != 0)`中表明头节点的状态为-1（后继节点处于等待状态）或者1（当前节点已经被取消）时才会执行`unparkSuccessor`。  
`unparkSuccessor`将会唤醒等待线程，且如果当前节点状态<0（有效时）会将状态置为0。（然后被唤醒的线程在成功获取锁后会将自己设置为头节点（虽然会把thread置为null）当然这也是后话了）。

public final boolean release(int arg) {
        //尝试解锁
        if (tryRelease(arg)) {
            Node h = head;
            //头节点不为空且状态不等于初始化值
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

`h != null`可以理解，但是为什么要`h.waitStatus != 0`呢？  
其实可以这么理解，在解锁后只有当队列中含有等待线程时，才会需要进行线程唤醒操作，而head节点为获取锁的节点（虽然Thread = null了），那么我们只需要看head之后的节点是否还存在等待节点即可，怎么判断呢？这时就可以用到当前节点的waitStats了（waitStats不仅代表了当前节点的状态，它还同时记录了队列中其他相关节点的状态），-1为后继节点处于等待中，所以-1时需要唤醒操作；1时为在同步队列中等待的线程等待超时或者被中断，需要从同步队列中取消等待，所以也需要对等待节点进行唤醒。

**tryRelease方法**  
尝试解锁

protected final boolean tryRelease(int releases) {
        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;
    }

**unparkSuccessor方法**  
对队列中的等待线程进行唤醒操作

private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        //s = 当前节点的后置节点
        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);
    }

[70]: /images/20220524/75c30788afd940a395b807760ce635ff.png
[70 1]: /images/20220524/35091f2170164b3bb724c12d5b28a56b.png
[70 2]: /images/20220524/1681e296ea384f89ae5e5cc6230e4832.png