【并发编程】ReentrantLock 源码分析

心已赠人 2022-03-20 10:30 285阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 ReentrantLock 概述 #

ReentrantLock 和 Synchronized 都可以实现同步加锁效果，将并行操作变为串行操作，保证共享资源的线程安全，

ReentrantLock 基于 AQS 实现，通过 lock() 方法获得锁，unlock() 方法释放锁。

# 二 ReentrantLock 实例 #

MyService：负责加锁和释放锁。

package com.jpeony.concurrent.locks.reentrantlock.demo1;
    
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * @author yihonglei
     */
    public class MyService {
        private ReentrantLock lock = new ReentrantLock();
    
        public void serviceMethod() {
            try {
                lock.lock();
                System.out.println(Thread.currentThread().getName() + ", begin ");
            } finally {
                System.out.println(Thread.currentThread().getName() + ", end ");
                lock.unlock();
            }
        }
    }

MyThread：负责调用服务方法。

package com.jpeony.concurrent.locks.reentrantlock.demo1;
    
    /**
     * @author yihonglei
     */
    public class MyThread extends Thread {
        private MyService service;
    
        public MyThread(MyService service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.serviceMethod();
        }
    }

RunTest：测试加锁和释放锁。

package com.jpeony.concurrent.locks.reentrantlock.demo1;
    
    /**
     * @author yihonglei
     */
    public class RunTest {
        public static void main(String[] args) {
            MyService service = new MyService();
            MyThread t1 = new MyThread(service);
            MyThread t2 = new MyThread(service);
            MyThread t3 = new MyThread(service);
            MyThread t4 = new MyThread(service);
            MyThread t5 = new MyThread(service);
    
            t1.setName("Thread-A");
            t2.setName("Thread-B");
            t3.setName("Thread-C");
            t4.setName("Thread-D");
            t5.setName("Thread-E");
    
            t1.start();
            t2.start();
            t3.start();
            t4.start();
            t5.start();
    
        }
    }

运行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70][]

从运行结果可以看到，每个线程访问时先调用 lock() 锁定，打印完数据后调用 unlock() 方法释放锁，其他线程才能

获取锁打印信息。打印的数据按线程持有锁分组的，但是线程运行顺序是随机的，也即线程的启动顺序与线程的

运行顺序无关，运行顺序取决于cpu调度。

# 三 ReentrantLock 源码分析 #

主要分析 lock() 加锁、unlock() 释放锁。ReentrantLock 加锁和解锁机制，都是基于 AQS 模板方法实现的，

理解 AQS，更容易理解 ReentrantLock 锁机制的本质，[AQS 源码分析。][AQS]

## 构造器 ##

默认是非公平的加锁方式。

/**
     * 默认是非公平
     */
    public ReentrantLock() {
        // 最顶层是 AQS
        sync = new NonfairSync();
    }
    
    /**
     * 可以选择公平或非公平 fair：true 公平，false 非公平
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

## 1、ReentrantLock\#lock() ##

加锁入口。

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

final void lock() {
        // 非公平时，线程先 CAS 尝试获取锁
        if (compareAndSetState(0, 1))
            // 获取锁成功，设置线程为独占锁线程，用于实现线程的重入锁
            setExclusiveOwnerThread(Thread.currentThread());
        else
            // 线程获取锁入口
            acquire(1);
    }

非公平锁机制，每次调用当前线程都会去先尝试获取锁，不管锁是否被占用，都要先尝试去获取锁，获取不到了再去排队获取锁，

而公平锁是发现有锁，就老老实实去排队获取锁。

**ReentrantLock\#acquire()**

尝试获取锁，失败则加入等待队列并挂起线程，等待唤醒获取锁。

public final void acquire(int arg) {
        // 尝试获取锁，获取失败则进入等待队列
        if (!tryAcquire(arg) &&
            // addWaiter 构建独占模式节点入队，acquireQueued() 尝试获取锁，没有获取到则挂起线程，并进入等待队列
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            // 线程是否已中断判断
            selfInterrupt();
    }

**ReentrantLock\#tryAcquire() **

尝试获取锁，默认走非公平锁的实现。

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

**Sync\#nonfairTryAcquire()**

1）如果 state 是 0，表示锁未被线程持有，尝试去获取锁，通过 CAS 轻量级操作将 state 变为1，

如果成功，则获取锁成功，否则获取锁失败；

2）如果 state 不是 0，判断是否是当前线程来获取锁，是则是重入锁，只要重入的次数未超过 int 最大值，

即可重入获取成功，如果不是当前线程，则获取锁失败；Lock 锁的重入是通过独占线程变量引用判断，

Synchronized 的重入是通过对象头偏向线程 id 判断实现。

final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        // 如果锁未被持有，则尝试去获取锁
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        // 如果是同一个线程再次获取锁，设置锁状态增加1，返回获取锁成功
        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;
    }

**AbstractQueuedSynchronizer\#addWaiter()**

如果获取锁失败，说明锁被占用，没有被释放，基于当前线程构建独占式节点入等待队列，

并挂起线程，等待唤醒重新尝试获取锁。

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;
        // 尝试快速添加到队列，第一次tail是空，会走enq逻辑
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 初始化队列并将当前节点添加到队尾
        enq(node);
        return node;
    }

**AbstractQueuedSynchronizer\#enq()**

初始化队列，当前线程构建的独占节点入队尾。AQS 的队列基于双向链表实现，头结点永远不存放值，

即基于哨兵结点方式简化的带头链表实现队列方式。

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

**AbstractQueuedSynchronizer\#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)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // shouldParkAfterFailedAcquire 第一次循环，将会修改节点状态为待唤醒， 
                // parkAndCheckInterrupt 第二次循环，LockSupport.park()暂停线程，程序计数器会标记在这个地方，当线程被唤醒后，线程将会从这个字节码地址继续执行线程，去尝试获取锁。
        // 在unlock() 释放锁的时候，会唤醒等待队列头结点线程尝试获取锁。
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

##  2、ReentrantLock\#unlock() ##

释放锁。

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

AbstractQueuedSynchronizer\#release()

释放锁逻辑。尝试去释放锁，唤醒头结点线程尝试获取锁。

public final boolean release(int arg) {
        // 尝试释放锁
        if (tryRelease(arg)) {
            // 释放锁成功，唤醒头结点的后续节点线程，头结点上是永远没有阻塞线程的，只有对待唤醒线程的next引用。
            Node h = head;
            // 节点不为空，节点状态不是初始状态
            if (h != null && h.waitStatus != 0)
                // 唤醒线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

**ReentrantLock.Sync\#tryRelease()**

释放状态值。

protected final boolean tryRelease(int releases) {
        // state - 1
        int c = getState() - releases;
        // 独占锁的线程不是当前线程，不允许释放锁，哪个线程加的锁，由哪个线程释放。
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        // 如果c为0，则释放锁成功，设置独占变量exclusiveOwnerThread为null
        boolean free = false;
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        return free;
    }

AbstractQueuedSynchronizer\#unparkSuccessor()

唤醒线程，被唤醒的线程尝试获取锁。

/**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            // 恢复头节点的等待状态为初始值0
            compareAndSetWaitStatus(node, ws, 0);
    
        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        // 如果节点为空或者是已被取消，迭代找到下一个符合唤醒的节点线程
        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);
    }

#  四 ReentrantLock 类与 Condition 类配合实现等待/通知机制 #

关键字 synchronized 与 wait() 和 notify() 或 notifyAll() 结合可以实现等待/通知机制。而 ReentrantLock 和 Condition 配合

也可以实现等待/通知机制。使用 notify() 和 notifyAll() 无法选择性的通知线程。使用 Condition 的优点就是可以在 Lock

对象里面创建多个 Condition (对象监视器)实例，线程对象可以注册在指定的 Condition 中，从而实现有选择性地进行

线程通知，使线程调度更加的灵活。

**Condition类方法:**

await()：使线程等待相当于Object对象中的wait()方法。

await(long time, TimeUnit unit)：相当于Object对象中的wait(long timeout)方法。

signal()：通知单个线程相当于Object对象中的notify()方法。

signalAll()：通知所有线程相当于Object对象中的notifyAll()方法。

**单个Condition实现等待/通知机制实例:**

Service服务类:

package com.jpeony.concurrent.locks.reentrantlock.demo2;
    
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * @author yihonglei
     */
    public class MyService {
        private Lock lock = new ReentrantLock();
        private Condition condition = lock.newCondition();
    
        /**
         * 等待
         */
        public void await() {
            try {
                lock.lock();
                System.out.println("await time:" + System.currentTimeMillis());
                condition.await(); // 使线程处于等待状态
            } catch (InterruptedException ex) {// 线程中断异常
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 通知
         */
        public void signal() {
            try {
                lock.lock();
                System.out.println("signal time:" + System.currentTimeMillis());
                condition.signal();// 通知等待线程进行唤醒
            } catch (IllegalMonitorStateException ex) {// 监视器不合法异常
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

线程:

package com.jpeony.concurrent.locks.reentrantlock.demo2;
    
    /**
     * @author yihonglei
     */
    public class MyThread extends Thread {
        private MyService service;
    
        public MyThread(MyService service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.await();
        }
    }

测试类:

package com.jpeony.concurrent.locks.reentrantlock.demo2;
    
    /**
     * 关键字synchronized与wait()合notify()或notifyAll()结合可以实现等待/通知机制。
     * 而ReentrantLock和Condition配合也可以实现等待/通知机制。
     * <p>
     * 使用notify()和notifyAll()无法选择性的通知线程。
     * 使用Condition的优点就是可以在Lock对象里面创建多个Condition(对象监视器)实例，
     * 线程对象可以注册在指定的Condition中，从而实现有选择性地进行线程通知，使线程调度更加的灵活。
     * <p>
     * Condition类中常用方法:
     * await() 使线程等待相当于Object对象中的wait()方法
     * await(long time, TimeUnit unit)相当于Object对象中的wait(long timeout)方法
     * signal() 通知单个线程相当于Object对象中的notify()方法
     * signalAll() 通知所有线程相当于Object对象中的notifyAll()方法
     *
     * @author yihonglei
     */
    public class RunTest {
        public static void main(String[] args) {
            try {
                MyService service = new MyService();
                MyThread myThread = new MyThread(service);
                /*
                 * 线程启动后执行run方法，调用MyService中的await()方法，
                 * 使得线程处于等待状态
                 */
                myThread.start();
    
                // 当前线程(main线程)休眠5秒后往下执行
                Thread.sleep(5000);
    
                /*
                 * 调用signal方法，通知MyService中Condition(监视器)中注册的线程
                 * 进行唤醒执行
                 */
                service.signal();
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }

测试结果:

![70][]

将运行结果signal time与await time做差，结果为5000毫秒左右，因为调用过程耗了点，

正好为RunTest类中Thread.sleep(5000)的5000毫秒，实现了等待/通知机制。

# 五 多个 Condition 通知部分线程 #

上面实例分析了线程指定单个Condition实现等待/通知机制，但是，当多个线程多个Condition时，如何实现通知部分线程?

创建Service类:

package com.jpeony.concurrent.locks.reentrantlock.demo3;
    
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * @author yihonglei
     */
    public class MyService {
        private Lock lock = new ReentrantLock();
    
        private Condition conditionA = lock.newCondition();
        private Condition conditionB = lock.newCondition();
    
        /**
         * A等待
         */
        public void awaitA() {
            try {
                lock.lock();
                System.out.println("waitA begin time: " + System.currentTimeMillis());
                conditionA.await();// 等待
                System.out.println("waitA end time: " + System.currentTimeMillis());
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * B等待
         */
        public void awaitB() {
            try {
                lock.lock();
                System.out.println("waitB begin time: " + System.currentTimeMillis());
                conditionB.await();// 等待
                System.out.println("waitB end time: " + System.currentTimeMillis());
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * conditionA通知:
         * signal() 通知单个线程
         * signalAll() 通知所有线程
         */
        public void signalAll_A() {
            try {
                lock.lock();
                System.out.println("signalAll_A begin time: " + System.currentTimeMillis()
                        + ", ThreadName = " + Thread.currentThread().getName());
                // 通知conditionA监视器中所有处于等待的线程进行唤醒
                conditionA.signalAll();
            } catch (IllegalMonitorStateException ex) {
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * conditionB通知:
         * signal() 通知单个线程
         * signalAll() 通知所有线程
         */
        public void signalAll_B() {
            try {
                lock.lock();
                System.out.println("signalAll_B begin time: " + System.currentTimeMillis()
                        + ", ThreadName = " + Thread.currentThread().getName());
                // 通知conditionB监视器中所有处于等待的线程进行唤醒
                conditionB.signalAll();
            } catch (IllegalMonitorStateException ex) {
                ex.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    
    }

线程A：

package com.jpeony.concurrent.locks.reentrantlock.demo3;
    
    /**
     * @author yihonglei
     */
    public class ThreadA extends Thread {
        private MyService service;
    
        public ThreadA(MyService service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.awaitA();
        }
    }

线程B：

package com.jpeony.concurrent.locks.reentrantlock.demo3;
    
    /**
     * @author yihonglei
     */
    public class ThreadB extends Thread {
        private MyService service;
    
        public ThreadB(MyService service) {
            super();
            this.service = service;
        }
    
        @Override
        public void run() {
            service.awaitB();
        }
    }

测试类:

package com.jpeony.concurrent.locks.reentrantlock.demo3;
    
    /**
     * @author yihonglei
     */
    public class RunTest {
        public static void main(String[] args) {
            try {
                MyService service = new MyService();
                ThreadA threadA = new ThreadA(service);
                threadA.setName("AAAAA");
                threadA.start();
    
                ThreadB threadB = new ThreadB(service);
                threadB.setName("BBBBB");
                threadB.start();
    
                // main线程休眠5秒，才执行后面的代码
                Thread.sleep(5000);
                /*
                 * 线程休眠5秒后开始执行service.signalAll_A(),
                 * 单纯就是唤醒ThreadA线程，不唤醒ThreadB线程，
                 * 这么测试的目的，就是想验证Condition能否做到只唤醒部分线程
                 */
                service.signalAll_A();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

运行结果:

![70 1][]

从运行结果可以看出，A经历了等待--唤醒--唤醒后执行，而B一直处于等待状态，未被唤醒。

线程注册时指定Condition，唤醒时根据Condition唤醒对应注册的线程，从而达到部分通知的能力。

# 六 总结 #

1）ReentrantLock 能实现 Synchronized 一样的同步效果，通过 lock() 加锁，unlock() 解锁。

2）ReentrantLock 依赖特殊的 cpu 指令，手动加锁和解锁实现同步效果。

3）ReentrantLock 的 Condition 可以实现等待/通知模式（生产/消费者模型），线程注册到 Condition，

通过 await() 方法等待，通过 signal() 或 signalAll() 实现通知。

4）[Synchronized][] 是 JVM 层面的锁机制，依赖 JVM 底层实现加锁和解锁，底层也是类似 AQS 实现方式，

ReentrantLock 相对更灵活些，比如自己可以控制获取锁的超时时间等，还针对读多写少并发场景基于 AQS 实现了

[ReentrantReadWriteLock][]（读写锁），这些特性都是 synchronized 没有的，但是 synchronized 使用更简单，

完全依赖 JVM 内部机制，代码更加简洁。

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70]: /images/20220320/5071be10980c4bfdb4916010b542d89c.png
[AQS]: https://jpeony.blog.csdn.net/article/details/102535590
[70]: /images/20220320/0261e8a959d44ab3a187bc33d3ccce9d.png
[70 1]: /images/20220320/bac3bca93a4e458f9ae1de2eaee47a5a.png
[Synchronized]: https://jpeony.blog.csdn.net/article/details/87008888
[ReentrantReadWriteLock]: https://jpeony.blog.csdn.net/article/details/113250932