JAVA 并发类（六） ReentrantLock 源码分析

àì夳堔傛蜴生んèń 2022-06-08 08:50 191阅读 0赞

ReentrantLock实现了Lock接口，是一种递归无阻塞的同步机制。

首先ReentrantLock有公平锁和非公平锁机制，使用构造方法ReentrantLock(boolean)指定，默认是非公平锁。

公平锁的含义是线程按照发出请求的先后顺序获取锁，而非公平锁的机制下，线程可以插队，后到的线程可以直接跳到队列头部获得锁

================================================

ps:前天看到有同学提问为什么非公平锁的效率比公平锁的效率高？  
原因是在恢复一个被挂起的线程到该线程真正运行之间存在着延迟

举个栗子，假设线程A持有锁，线程B请求该锁，由于锁已经被占用，B被阻塞挂起，知道A释放锁时才会唤醒B，然后B才会重新尝试去获取该锁。但是此时，如果线程C也请求这个锁，那么如果是非公平锁的话，C会在B被完全唤醒之前或者使用以及释放该锁，这样B获得锁的时间没有推迟，而C也更早的获得了锁，并且吞吐量提高了，当然这是在C持有锁时间较短的情况下，如果线程持有锁的情况较长应该使用公平锁，OVER

=================================================

继续：  
ReentrantLock内有三个内部类，分别是Sync,FairSync和NoFairSync，其中FairSync和NoFairSync是Sync的子类。

/** * 该锁同步控制的一个基类.下边有两个子类：非公平机制和公平机制.使用了AbstractQueuedSynchronizer类的 */
    static abstract class Sync extends AbstractQueuedSynchronizer
    
    /** * 非公平锁同步器 */
    final static class NonfairSync extends Sync
    
    /** * 公平锁同步器 */
    final static class FairSync extends Sync

因此使用ReentrantLock的加锁lock操作时，又分为了两种情况，分别是公平锁的lock和非公平锁的lock;

**非公平锁的核心思想**

1.基于CAS尝试把锁数量从0设置为1  
2.如果设置成功，当前线程为独占锁的线程  
3.设置失败，会再次尝试获取锁数量  
4.如果锁数量为0，再基于CAS尝试把锁数量从0设置为1，如果设置成功，当前线程为独占锁的线程  
5.如果锁数量不为0，查看当前线程是不是已经是独占锁的线程了，如果是，则将当前的锁数量+1；如果不是，则将该线程封装在一个Node内，并加入到等待队列中去。等待被其前一个线程节点唤醒。

**非公平锁源码分析**

final void lock() {
       if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

首先通过CAS操作尝试把锁数量从0设置为1，如果成功，当前线程为独占锁的线程，否则执行一个acquire(1)方法

我们下面看acquire(1)的具体实现

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

先调用了tryAcquire方法 看非公平锁中的具体实现

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

再次获取当前的锁数量，如果锁数量为0，表示没有线程持有锁，那么再尝试把当前state设置为1，如果设置成功，那么当前线程就持有该锁，返回true;  
锁数量不为0话，意味着有线程持有该锁，判断是否是当前线程持有\[可重入\]，是的话把当前锁数量+1，否则的话返回false.  
此时tryAcquire(1)方法返回false,会继续执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法，而又先调用了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;
     if (pred != null) {
         node.prev = pred;
         if (compareAndSetTail(pred, node)) {
             pred.next = node;
             return node;
         }
     }
     enq(node);
     return node;
    }

这个方法把当前线程封装成一个Node对象，然后把Node加入等待队列。可以看到代码上的注释写着：先尝试快速入队，快速入队失败就尝试普通入队方法  
根据代码看到，快速入队就是使用CAS操作尝试把尾节点tail设置成node,并把之前的尾节点插入到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;
              }
          }
      }
    }

如果尾节点为空，进行初始化，如果非空，就一直CAS循环到node插入到队尾为止

队列入队成功后返回node节点，我们再回到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;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

获取node的前驱节点p，如果p是头节点，就继续使用tryAcquire(1)方法去尝试请求成功，如果第一次请求就成功，不用中断自己的线程，如果是之后的循环中将线程挂起之后又请求成功了，使用selfInterrupt()中断自己

如果p不是头节点，或者tryAcquire(1)请求不成功，就去执行shouldParkAfterFailedAcquire(Node pred, Node node)来检测当前节点是不是可以安全的被挂起，如果node的前驱节点pred的等待状态是SIGNAL（即可以唤醒下一个节点的线程），则node节点的线程可以安全挂起。如果node的前驱节点pred的等待状态是CANCELLED，则pred的线程被取消了，我们会将pred之前的连续几个被取消的前驱节点从队列中剔除，返回false。

如果node的前驱节点pred的等待状态是除了上述两种的其他状态，则使用CAS尝试将前驱节点的等待状态设为SIGNAL，并返回false（因为CAS可能会失败，这里不管失败与否，都返回false，下一次执行该方法的之后，pred的等待状态就是SIGNAL了）

如果可以安全挂起，就执行parkAndCheckInterrupt()挂起当前线程

最后，直到该节点的前驱节点p之前的所有节点都执行完毕为止，我们的p成为了头节点，并且tryAcquire(1)请求成功，跳出循环，去执行

看下Node节点的具体结构

/** * 同步等待队列（双向链表）中的节点 */
    static final class Node {
        /** 线程被取消了 */
        static final int CANCELLED = 1;
        /** * 如果前驱节点的等待状态是SIGNAL，表示当前节点将来可以被唤醒，那么当前节点就可以安全的挂起了 * 否则，当前节点不能挂起 */
        static final int SIGNAL = -1;
        /**线程正在等待条件*/
        static final int CONDITION = -2;
        /** * waitStatus value to indicate the next acquireShared should * unconditionally propagate */
        static final int PROPAGATE = -3;
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** 一个标记：用于表明该节点正在独占锁模式下进行等待 */
        static final Node EXCLUSIVE = null;
        //值就是前四个int（CANCELLED/SIGNAL/CONDITION/PROPAGATE），再加一个0
    
        volatile int waitStatus;
        /**前驱节点*/
        volatile Node prev;
    
        /**后继节点*/
        volatile Node next;
    
        /**节点中的线程*/
        volatile Thread thread;
    
        /** * Link to next node waiting on condition, or the special value SHARED. * Because condition queues are accessed only when holding in exclusive * mode, we just need a simple linked queue to hold nodes while they are * waiting on conditions. They are then transferred to the queue to * re-acquire. And because conditions can only be exclusive, we save a * field by using special value to indicate shared mode. */
        Node nextWaiter;
    
        /** * Returns true if node is waiting in shared mode */
        final boolean isShared() {
            return nextWaiter == SHARED;
        }
    
        /** * 返回该节点前一个节点 */
        final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
        }
    
        Node() { // Used to establish initial head or SHARED marker
        }
    
        Node(Thread thread, Node mode) { // 用于addWaiter中
            this.nextWaiter = mode;
            this.thread = thread;
        }
    
        Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
        }
    }

**公平锁的核心**

获取一次锁数量

1.如果锁数量为0，如果当前线程是等待队列中的头节点，基于CAS尝试将state（锁数量）从0设置为1一次，如果设置成功，设置当前线程为独占锁的线程；

2.如果锁数量不为0或者当前线程不是等待队列中的头节点或者上边的尝试又失败了，查看当前线程是不是已经是独占锁的线程了，如果是，则将当前的锁数量+1；如果不是，则将该线程封装在一个Node内，并加入到等待队列中去。等待被其前一个线程节点唤醒。

/** * 获取公平锁的方法 * 1）获取锁数量c * 1.1)如果c==0，如果当前线程是等待队列中的头节点，使用CAS将state（锁数量）从0设置为1，如果设置成功，当前线程独占锁-->请求成功 * 1.2)如果c!=0，判断当前的线程是不是就是当下独占锁的线程，如果是，就将当前的锁数量状态值+1（这也就是可重入锁的名称的来源）-->请求成功 * 最后，请求失败后，将当前线程链入队尾并挂起，之后等待被唤醒。 */
            protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
                    if (isFirst(current) && 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;
            }

如果请求失败也是把线程放入队尾挂起等待唤醒，和非公平锁一样

**总结二者的区别**

公平锁的lock加锁少了插队部分(CAS尝试把state从0设置为1从而获得锁的过程)  
公平锁的tryAcquire有判断当前线程是否在等待队列队首的逻辑  
ReentrantLock是基于AbstractQueuedSynchronizer实现的，AbstractQueuedSynchronizer可以实现独占锁也可以实现共享锁，ReentrantLock只是使用了其中的独占锁模式

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

前面介绍了ReentrantLock的公平锁和非公锁的加锁机制，下面再简单说下解锁机制。  
首先解锁的流程：  
1.获取当前的锁数量，然后用这个锁数量减去解锁的数量（1），最后得出结果c  
2.判断当前线程是不是独占锁的线程，如果不是，抛出异常  
3.如果c为0，表示锁被成功释放，把当前独占的线程设置为null,锁数量设置为0，返回true  
4.如果c不为0，说明锁释放失败，锁数量设置为c,返回false  
5.如果锁被释放成功的话，唤醒距离头结点最近的一个非取消的节点

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

调用了tryRelease尝试释放锁

protected final boolean tryRelease(int releases) {
     int c = getState() - releases;//获取现在的锁数量-传入的解锁数量（这里为1）
        if (Thread.currentThread() != getExclusiveOwnerThread())//当前线程不持有锁
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) {
       //锁被释放
            free = true;
            setExclusiveOwnerThread(null);
        }//如果不为0，怎么办，不释放了吗？
        setState(c);
        return free;
    }

尝试释放锁成功就唤醒等待队列中最近的一个节点，释放失败就把锁数量重置

private void unparkSuccessor(Node node) {
    
            int ws = node.waitStatus;
            if (ws < 0)//将ws设为0状态（即什么状态都不是）
                compareAndSetWaitStatus(node, ws, 0);
    
            /* * 获取头节点的下一个等待状态不是cancel的节点 */
            Node s = node.next;//头节点的下一个节点
            if (s == null || s.waitStatus > 0) {
                s = null;
                /* * 注意：从后往前遍历找到离头节点最近的一个非取消的节点，从后往前遍历据说是在入队（enq()）的时候，可能nodeX.next==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);//唤醒离头节点最近的一个非取消的节点
        }

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