ReentrantLock重入锁同步线程源码学习

Posted 2020-11-25 yxlaisj

1. 基本概念

　　程序运行过程中，两个或多个线程(thread)并发执行并共享某个资源时，可能对共享资源不同步地修改，造成数据错误（所谓错误，就是修改后的数据不符合预期），为了避免数据错误，普遍采用了线程同步技术，所谓同步，就是避免多个线程毫无规则地征用资源，而是使这些线程看起来像是步调一致、有序地使用共享资源。比如，有线程A,B和C, 有一个可共享的资源R，为了避免多线程并发造成对R资源的冲突，必须同步线程，即当A正在对R进行操作的时候，B和C就在这一块操作代码之外等待A完成操作。

　　Java中，同步的范围主要有两种，一种是同步方法，对于同一个对象，用synchronized修饰的所有方法，同一时刻至多只能有一个线程进入其中之一。第二种是同步代码块，可用synchronize包围起来，也可以用锁Lock对象，而后者，最常用的就是ReentrantLock，即重入锁。

      synchronized关键字的是操作系统底层的互斥锁(mutex lock)来实现的，跟操作系统有关。而ReentrantLock的实现原理是CAS+自旋算法，CAS(Compare and save)就是实现更改数据原子性的算法，底层是JNI调用C代码实现的， 自旋算法就是在一个不设循环次数的循环中不断地循环，直到满足一定条件为止。持有ReentrantLock的线程执行被锁住的代码块，新来的线程尝试去获取ReentrantLock，如果获取到了，就进入代码块执行，如果没有获取到，就会阻塞在代码块之外。

     使用Lock的优点在于可以更加精细地控制线程执行锁操作和释放锁，比如可以使用tryLock()的重载方法指定尝试执行锁操作的时间，如果在指定时间内没有锁成功，就可以根据业务需要做其它处理。Lock还可以指定线程是否“公平地”争取锁，是先来后到还是后来的也能挣先。

     本篇随笔，只学习和分析ReentrantLock的源码和实现原理。

 

2. ReentrantLock 的锁操作

　　ReentrantLock是最常用的锁，每个线程就像一个人，一个人（线程）执行任务（程序）走到了一座房子（需要同步执行的代码块）的门前（代码块入口），推门并从内上锁这一过程，就是ReentrantLock.lock()操作，当然，如果推门推不动，说明屋子里有别人呢，得等他完成任务后把锁打开（unlock）。

2.1  demo

　　学习任何源码，最好是写一个简单的demo，作为一个入口而又能对整体执行有个了解才好入手。

　　现有两个账户，一个from，一个to，初始额度都是100元， from向to中转账，应该两个账户总金额恒定为200元才对。试用多线程，模拟多个人同时使用该账户转账。Account是账户对象，ReentrantLockDemo负责模拟多线程转账操作。首先，不用锁，开启5个线程去跑转账的方法，打印出每个线程转账完成后的结果。

 1 public class ReentrantLockDemo {
 2     // 转出账户
 3     private Account from;
 4     // 转入账户
 5     private Account to;
 6     // 持有重入锁
 7     private Lock lock = new ReentrantLock();
 8 
 9     public ReentrantLockDemo(Account from, Account to) {
10         this.from = from;
11         this.to = to;
12     }
13 
14     /**
15      * 转账操作，从from账户转money额度到to账户
16      */
17     public void transform(int money) {
18 //        lock.lock();
19         try {
20             from.minus(money);
21             Thread.yield();     // 为了更大几率让其它线程进入
22             to.add(money);
23             System.out.println(Thread.currentThread() + " from.money = " + from.money + "; to.money = " + to.money);
24         } finally {
25 //          lock.unlock();
26         }
27     }
28 
29     public static void multiThreadTransform(ReentrantLockDemo demo) {
30         ExecutorService exec = Executors.newFixedThreadPool(5);
31         for (int i = 0; i < 5; i++) {
32             exec.execute(() -> {
33                 demo.transform(10);
34             });
35         }
36         exec.shutdown();
37     }
38 
39     public static void main(String[] args) {
40         Account from = new Account(),
41                 to = new Account();
42         ReentrantLockDemo demo = new ReentrantLockDemo(from, to);
43         multiThreadTransform(demo);
44 
45     }
46 }

 1 class Account {
 2     // 账号初始额度为100元
 3     int money = 100;
 4 
 5     void add(int m) {
 6         money += m;
 7     }
 8     void minus(int m) {
 9         money -= m;
10     }
11 }

结果：

Thread[pool-1-thread-2,5,main] from.money = 80; to.money = 130
Thread[pool-1-thread-4,5,main] from.money = 60; to.money = 140
Thread[pool-1-thread-1,5,main] from.money = 80; to.money = 130
Thread[pool-1-thread-3,5,main] from.money = 70; to.money = 130
Thread[pool-1-thread-5,5,main] from.money = 50; to.money = 150

　　可以从上面结果看到，线程2转账完成后，from账户为80元，to账户为130元，相加为210元，在这个线程之中的人，读出了错误的数据。而且90,110的组合也不一定出现。

　　加上锁之后，结果就符合预期了。

 1 /**
 2      * 转账操作，从from账户转money额度到to账户
 3      */
 4     public void transform(int money) {
 5         lock.lock();　　// 锁操作，实现线程同步
 6         try {
 7             from.minus(money);
 8             Thread.yield();     // 为了更大几率让其它线程进入
 9             to.add(money);
10             System.out.println(Thread.currentThread() + " from.money = " + from.money + "; to.money = " + to.money);
11         } finally {
12           lock.unlock(); 　　// 释放锁，以便于接下来的线程进入代码块
13         }
14     }

Thread[pool-1-thread-1,5,main] from.money = 90; to.money = 110
Thread[pool-1-thread-5,5,main] from.money = 80; to.money = 120
Thread[pool-1-thread-2,5,main] from.money = 70; to.money = 130
Thread[pool-1-thread-3,5,main] from.money = 60; to.money = 140
Thread[pool-1-thread-4,5,main] from.money = 50; to.money = 150

 

2.2 抽象队列同步器AQS

　　进入ReentrantLock.lock()方法后，看到源码是这样的：

1 public void lock() {
2     sync.lock();
3 }

　　方法中，调用的是sync.lock()，Sync是ReentrantLock的内嵌类，它继承了AbstractQueuedSynchronizer，是抽象队列同步器，所以ReentrantLock底层的线程同步实际是由队列同步器实现的。ReentrantLock, Sync的依赖关系如下图所示:

　　1)    ReentrantLock依赖于队列同步器Sync对象;

　　2)   Sync对象有两个子类，分别是FairSync和NonfairSync，从名称来看应该是公平锁和非公平锁的实现；

　　3）ReentrantLock同步由Sync.lock()实现;

　　4)  使用ReentrantLock默认构造器，创建的是非公平队列同步器NonfairSync。

　　那么Sync对象是如何实现线程同步的，接着往下看。

 

2.3 NonfairSync.lock()

 1         /**
 2          * Performs lock.  Try immediate barge, backing up to normal
 3          * acquire on failure.
 4          */
 5         final void lock() {
 6             if (compareAndSetState(0, 1))
 7                 setExclusiveOwnerThread(Thread.currentThread());
 8             else
 9                 acquire(1);
10         }

1 protected final boolean compareAndSetState(int expect, int update) {       // See below for intrinsics setup to support this
2         return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
3     }

　　进入到NonfairSync.lock()方法，通过注释内容可知，该方法执行锁操作，一个新来的线程，不分先来后到立即先尝试去加锁，如果失败，就执行acquire(1)方法。 compareAndSetState(0, 1)就是CAS操作，即对比-更新，这一步是原子操作，第一个参数是期望值，第二个参数是更新值，意思是：如果现在的值是0，那么就更新为1， 其原子性是底层C代码实现的。这里要更新的值是RenentrantLock对象所持有的抽象队列同步器的状态，代码如下。 当stateOffset的值是0时，更新为1。假设有两个线程同时运行到了下面代码的第3行，由于是原子操作，同一时刻，只可能有一个线程能成功改变stateOffset的值。 假设当前线程成功了，则在AQS中，把当前排他模式的线程持有者设置为当前线程setExclusiveOwnerThread()，如果没有上锁成功，那么就执行acquire()方法.

2.4 AbstractQueuedSynchronizer.acquire()

　　

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

 

　　该方法，以独占模式获取锁，忽略中断。这个方法的原理就是一个线程不断地去尝试锁操作，直到成功，否则就一直阻塞。没有立即成功执行锁操作的线程，会放入一个FIFO队列，队列中的线程会根据相应策略对竞争锁。这个方法做了这么几件事：

　　1).  先尝试去执行锁才注意tryAquire(arg)，如果成功了，就退出方法，线程继续执行下去；

　　2). 创建一个Node节点，封装当前线程并入AQS队列，即执行addWaiter()方法；

　　3). 入队后，就会循环地去尝试锁，如果锁成功了就继续执行，没有获取到就阻塞，然后继续尝试锁操作，即执行acquireQueued()方法；

      4). 第3）步中，如果返回了true，则表示阻塞的线程被中断了，那么selfInterrupt()的作用是标记当前线程被终端，为什么在这里标记，这是因为线程被中断后，标记会被清除，所以这里重新标记一下。

      5). 为什么第3)步中，线程会被中断，这个问题留在之后进入acquireQueued()方法解答

 

 

2.5 NonfairSync.nonfairTryAcquire()

　　

 1 final boolean nonfairTryAcquire(int acquires) {
 2        final Thread current = Thread.currentThread();　　　　// 获取当前线程
 3        int c = getState();　// 获得当前AQS同步器状态  
 4        if (c == 0) {　　// 0表示当前锁是释放状态，可以上锁
 5            if (compareAndSetState(0, acquires)) {　　// 使用CAS原子操作更新AQS的state，使AQS对象的状态变成上锁状态
 6                setExclusiveOwnerThread(current);　　　// 把当前线程设置为独占线程
 7                return true;
 8             }
 9         }
10         else if (current == getExclusiveOwnerThread()) {　　// AQS是上锁状态，且当前线程就是AQS中的独占线程
11            int nextc = c + acquires;　　
12             if (nextc < 0) // overflow
13                 throw new Error("Maximum lock count exceeded");
14             setState(nextc);　　// 更新AQS状态, 无需用CAS操作，因为只有当前独占线程能进入到这里
15             return true;
16         }
17         return false;
18  }

 

　　调用方法tryAcquire()，会调用NonfairSync.nonfairTryAcquire()方法， 返回true的话，表示执行锁操作成功了，返回false，表示执行锁操作失败了。上面的方法每个步骤的作用已经注释了，流程非常清晰，上锁的过程，实际上就是更新AQS的state的过程，另外，注意这里默认是独占锁，所以CAS更新操作成功之后，还会设置当前线程为独占线程，而state，独占线程的引用变量等参数，都是AQS对象持有的，可以看到RenetrantLock的核心就是AQS，而NonfairSync是AQS的子类。

 

2.6 AbstractQueuedSynchronizer.addWaiter()

　　上面执行锁操作，如果执行失败了，那么就会执行addWaiter()方法，这个方法是讲当前线程封装成一个Node，然后将这个Node存入一个FIFO队列：

 1 private Node addWaiter(Node mode) {
 2         Node node = new Node(Thread.currentThread(), mode);　　// 封装当前线程为Node对象
 3         // Try the fast path of enq; backup to full enq on failure　　
 4         Node pred = tail;　　// tail是当前队列的队尾节点
 5         if (pred != null) {
 6             node.prev = pred;　　// 如果tail节点不是null，就把当前Node放到tail节点后面
 7             if (compareAndSetTail(pred, node)) {　　// 原子操作，更新当前队列的队尾节点为新封装的Node节点
 8                 pred.next = node;
 9                 return node;
10             }
11         }
12         enq(node);　　// 如果tail节点是null，就会调用enq方法，进行必要的队列初始化，然后再把node节点入队操作
13         return node;
14     }

　　然后看enq(node)方法:

 1 /**
 2      * Inserts node into queue, initializing if necessary. See picture above.
 3      * @param node the node to insert
 4      * @return node‘s predecessor
 5      */
 6     private Node enq(final Node node) {
 7         for (;;) {　　// 自旋
 8             Node t = tail;　　
 9             if (t == null) { // Must initialize　　
10                 if (compareAndSetHead(new Node()))　　// 如果队列尾节点是null，就初始化队列：1.创建头节点并，此时只有一个节点，头节点也是尾节点，注意，由于这是在一个自旋循环中的操作，所以这一步防止死循环
11                     tail = head;
12             } else {　　// 当自旋操作确保了tail不为null以后，就会在这一步，把封装有当前线程的node节点入队
13                 node.prev = t;　　
14                 if (compareAndSetTail(t, node)) {
15                     t.next = node;
16                     return t;
17                 }
18             }
19         }
20     }

　　以上方法通过自旋+CAS，将封装有当前线程对象的Node节点对象存入AQS对象维护的队列中，必须再次强调，CAS操作是JNI调用底层C代码保证更新操作的原子性，以保证同一时刻最多只能有一个线程成功更新目标值。

上述方法中，compareAndSetHead(new Node())采用原子操作创建的head节点，是一个没有封装任何thread的占位节点，因为这个时候还没有封装有线程的节点占有锁。

 

2.7 AbstractQueuedSynchronizer.acquireQueued()

　　此前的种种操作，总结起来其实就是尝试几次上锁，没上锁成功的话，就把当前线程排队。那么线程是怎么实现阻塞等待过程的呢，就是这个方法实现的，具体来说，这里的实现原理是线程每隔一段时间就会去尝试上锁操作，如果上锁成功的话就会继续执行被锁住的代码块，如果上锁失败会继续  阻塞--尝试上锁--阻塞--尝试上锁  这一过程，显然，从这里的描述来看，肯定又用到了自旋+CAS，所以说，ReentrantLock的最终要的算法就是自旋+CAS。

 1 /**
 2      * Acquires in exclusive uninterruptible mode for thread already in
 3      * queue. Used by condition wait methods as well as acquire.
 4      *
 5      * @param node the node
 6      * @param arg the acquire argument
 7      * @return {@code true} if interrupted while waiting
 8      */
 9     final boolean acquireQueued(final Node node, int arg) {
10         boolean failed = true;
11         try {
12             boolean interrupted = false;
13             for (;;) {
14                 final Node p = node.predecessor();　　// 当前节点的前置节点
15                 if (p == head && tryAcquire(arg)) {　　// 只有前置节点是head节点的线程，才有资格去竞争锁操作
16                     setHead(node);　　　　　// 当前节点封装的线程成功加锁，就设置为head节点
17                     p.next = null; // help GC
18                     failed = false;
19                     return interrupted;
20                 }
21                 if (shouldParkAfterFailedAcquire(p, node) &&
22                     parkAndCheckInterrupt())　　// 上一步没有获取锁，就阻塞一会。
23                     interrupted = true;
24             }
25         } finally {
26             if (failed)
27                 cancelAcquire(node);
28         }
29     }

　　这是lock()方法的核心，注意，只有封装了当前线程的节点的前置节点是头节点， 才能去争取锁，即tryAcquire()，这个方法在上面已经学习过了，其内部自旋+CAS的实现，保证只能有一个线程成功争取到锁操作。假设争取到了锁，就会把当前线程的节点设置为head节点。

　　每尝试一次锁操作，如果不成功，该线程便会阻塞一会，这个阻塞功能是在第二个if中的方法实现的shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt() ， 用自然语言描述就是:

if(要不要阻塞 && 阻塞一会) {
       设置中断标志  
}

　　1) 要不要阻塞shouldParkAfterFailedAcquire()

 1 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 2         int ws = pred.waitStatus;　　// Node对象的属性，表示当前节点代表的线程的等待状态
 3         if (ws == Node.SIGNAL)　　// 前置节点状态为SIGNAL，表示它获取了同步状态，那么节点就应该阻塞
 4             /*
 5              * This node has already set status asking a release
 6              * to signal it, so it can safely park.
 7              */
 8             return true;
 9         if (ws > 0) {　　　　　　// 前置节点被cancel了
10             /*
11              * Predecessor was cancelled. Skip over predecessors and
12              * indicate retry.
13              */
14             do {
15                 node.prev = pred = pred.prev;　　// 连续赋值操作，将pred.prev分别赋值给pred和node.prev，这个操作循环进行，直到找到不是取消状态的前置节点
16             } while (pred.waitStatus > 0);
17             pred.next = node;　　// 剔除掉取消状态的节点
18         } else {
19             /*
20              * waitStatus must be 0 or PROPAGATE.  Indicate that we
21              * need a signal, but don‘t park yet.  Caller will need to
22              * retry to make sure it cannot acquire before parking.
23              */
24             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);　　// 设置前任节点为SIGNAL，设置成功，在下一次循环就会阻塞当前节点
25         }
26         return false;
27     }

　　以上方法主要是判断当前节点是否应该阻塞，如果返回了true，那么就会执行下面的方法实现阻塞

1  /**
2      * Convenience method to park and then check if interrupted
3      *
4      * @return {@code true} if interrupted
5      */
6     private final boolean parkAndCheckInterrupt() {
7         LockSupport.park(this);　　// 将线程挂起从而实现当前线程阻塞
8         return Thread.interrupted();　　// 返回线程中断状态
9     }

　　　这个方法调用了底层代码实现了线程挂起，具体细节就不再深究了。值得注意的是返回线程状态，在2.4节中我提了一个问题，就是在acquire()方法中，当成功执行了锁操作后，为什么要执行selfInterrupt()操作，这里就可以解答了，LockSupport.part()将线程挂起，支持两种结束方式，一种是前置节点释放锁之后，当前节点正常调用LockSupport.unPark()来取消阻塞，这里不涉及到中断，return Thread.interrupted()就为false；

第二种是LockSupport.park()支持其它线程中断当前线程的阻塞状态，这样返回Thread.interrupted()为true，但是Thread.interrupted()方法，是会清除线程的中断状态标记的，为什么这么做？这样做是为了使这个被中断的线程在acquireQueued方法中，下次阻塞时，忽略线程中断状态。 如果当前线程的前置节点是head节点，那么它成功执行了锁操作之后，返回结果就应该是true ，然后调用selfInterrupt（）方法中断当前线程。

 1 static void selfInterrupt() { 2 Thread.currentThread().interrupt(); 3 } 　

 

3. ReentrantLock 的unLock操作

　　ReentrantLock释放锁与上锁是相反的过程，而这一过程的底层，都是调用了AQS的方法。

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

3.1 sync.release

　　实际释放锁调用的是AQS的release方法,这个方法如下：

1 public final boolean release(int arg) {
2         if (tryRelease(arg)) {　　　　// 尝试释放锁
3             Node h = head;
4             if (h != null && h.waitStatus != 0)　　
5                 unparkSuccessor(h);　　// 锁释放成功，会将head节点的watistatus设置为0，这样才可以使head节点的后继节点有资格去竞争锁，回忆上面的shouldParkAfterFailedAcquire()方法
6             return true;
7         }
8         return false;
9     }

　　释放锁过程很简单，就是：

　　1） 先尝试释放锁，如果成功了，

　　2)  取到head节点，如果head节点的waitStatus不是0(SIGNAL)，

      3)  在unparkSuccessor方法中，就是将head节点的waitStatus状态设置为0，以同步AQS状态，使后继节点可以争取锁，并且在这步操作中结束后置节点的阻塞状态

 

3.2 sync.tryRelease()

 1 protected final boolean tryRelease(int releases) {
 2             int c = getState() - releases;　　// getState()获取到AQS的状态，是一个int值，代表一个线程上锁的次数，减去release代表要释放的锁的个数，这里假设为1
 3             if (Thread.currentThread() != getExclusiveOwnerThread())　　　// 当前线程如果不是独占的，就抛出异常　
 4                 throw new IllegalMonitorStateException();
 5             boolean free = false;
 6             if (c == 0) {　　// c等于0表示AQS同步器没有线程独占，其它线程可以准备竞争上锁了
 7                 free = true;
 8                 setExclusiveOwnerThread(null);
 9             }
10             setState(c);
11             return free;
12         }

　　这步操作也比较明确，就是去释放锁，并且更新AQS的状态，如果AQS的state==0，就返回true，以便于通知其它线程准备抢锁。

 

3.3 sync.unparkSuccessor()

　　

 1 private void unparkSuccessor(Node node) {
 2         /*
 3          * If status is negative (i.e., possibly needing signal) try
 4          * to clear in anticipation of signalling.  It is OK if this
 5          * fails or if status is changed by waiting thread.
 6          */
 7         int ws = node.waitStatus;
 8         if (ws < 0)
 9             compareAndSetWaitStatus(node, ws, 0);
10 
11         /*
12          * Thread to unpark is held in successor, which is normally
13          * just the next node.  But if cancelled or apparently null,
14          * traverse backwards from tail to find the actual
15          * non-cancelled successor.
16          */
17         Node s = node.next;
18         if (s == null || s.waitStatus > 0) {
19             s = null;
20             for (Node t = tail; t != null && t != node; t = t.prev)
21                 if (t.waitStatus <= 0)
22                     s = t;
23         }
24         if (s != null)
25             LockSupport.unpark(s.thread);
26     }

　　这个方法中传入的node是head节点，进行以下操作：

　　1） 将head节点状态设置为0，但是如果这个节点的线程是1(CANCELED)，就不去设置

　　2） 这个head节点所持有的线程既然释放了锁，那么它的后继节点线程就应该解除阻塞状态去获取锁了

　　3） 该方法中使用for循环找到一个距离当前head节点最近的一个符合条件（不为null，且状态不是CANCELED）的节点

　　4） 调用LockSupport.unpark()方法，解除该节点线程的阻塞状态

　　在这里，LockSupport.unpark(s.thread)方法，使用JNI调用了底层代码实现的，不做解释。以上全部内容就结束了，鄙人才疏学浅，是鄙人学习Java过程中记录自己点点滴滴的理解过程，如果有任何错误和疏漏，希望能得到各位的赐教。