Lock的实现之ReentrantLock详解

摘要

Lock在硬件层面依赖CPU指令，完全由Java代码完成，底层利用LockSupport类和Unsafe类进行操作；虽然锁有很多实现，但是都依赖AbstractQueuedSynchronizer类，我们用ReentrantLock进行讲解；

ReentrantLock调用过程

ReentrantLock类的API调用都委托给一个内部类 Sync ,而该类继承了 AbstractQueuedSynchronizer类；public class ReentrantLock implements Lock, java.io.Serializable {......abstract static class Sync extends AbstractQueuedSynchronizer {......而Sync又分为两个子类：公平锁和非公平锁，默认为非公平锁 /** * Sync object for non-fair locks */ static final class NonfairSync extends Sync { /** * Sync object for fair locks */ static final class FairSync extends Sync { Lock的调用过程如下图（其中涉及到 ReentrantLock类、Sync（抽象类）、AbstractQueuedSynchronizer类，NofairSync类，这些类将 Template方法用的淋漓尽致，相当赞）：先来一张类依赖图：再来一张lock调用图：

Lock API详解

自底而上来看，由被调用一步步向上分析

nofairTryAcquire

/** * Performs non-fair tryLock.tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */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） // overflowthrow new Error（"Maximum lock count exceeded"）;setState（nextc）;return true;}return false;}来看这段代码，首先获取当前状态（初始化为0），当它等于0的时候，代表还没有任何线程获得该锁，然后通过CAS（底层是通过CompareAndSwapInt实现）改变state，并且设置当前线程为持有锁的线程；其他线程会直接返回false;当该线程重入的时候，state已经不等于0，这个时候并不需要CAS，因为该线程已经持有锁，然后会重新通过setState设置state的值，这里就实现了一个偏向锁的功能，即锁偏向该线程；

addWaiter

只有当锁被一个线程持有，另外一个线程请求获得该锁的时候才会进入这个方法/** * Creates and enqueues node for current thread and given mode. * * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared * @return the new node */private Node addWaiter（Node mode） {Node node = new Node（Thread.currentThread（）, mode）;// Try the fast path of enq; backup to full enq on failureNode pred = tail;if （pred ！= null） {node.prev = pred;if （compareAndSetTail（pred, node）） {pred.next = node;return node;}}enq（node）;return node;}首先持有该锁之外的线程进入到该方法，这里涉及到一个CLH（三个人的名字首字母：Craig, Landin, and Hagersten）队列，其实就是一个链表，简单说下CLH队列：CLH队列由node节点组成，mode代表每个Node有两种模式：共享模式和排他模式，并且维护了一个状态：waitStatus，可取值如下：

CANCELLED = 1 由于超时或者被打断，该线程被取消，将不会被block；
SIGNAL = -1 当前线程的后继节点线程通过park正处于或即将处于block状态；
CONDITION = -2 当前线程正处于条件队列，正式因为调用了condition.await造成阻塞；
PROPAGATE = -3 共享锁应该被传播出去

首先，new一个节点，这个时候模式为：mode为 Node.EXCLUSIVE，默认为null即排它锁；然后：如果该队列已经有node即tail！=null，则将新节点的前驱节点置为tail，再通过CAS将tail指向当前节点，前驱节点的后继节点指向当前节点，然后返回当前节点；如果队列为空或者CAS失败，则通过enq入队：/** * Inserts node into queue, initializing if necessary. See picture above. * @param node the node to insert * @return node"s predecessor */private Node enq（final Node node） {for （;;） {Node t = tail;if （t == null） { // Must initializeif （compareAndSetHead（new Node（）））tail = head;} else {node.prev = t;if （compareAndSetTail（t, node）） {t.next = node;return t;}}}}进队的时候，要么是第一个入队并且设置head节点并且循环设置tail，要么是add tail，如果CAS不成功，则会无限循环，直到设置成功，即使高并发的场景，也最终能够保证设置成功，然后返回包装好的node节点；

acquireQueued

/** * Acquires in exclusive uninterruptible mode for thread already in * queue. Used by condition wait methods as well as acquire. * * @param node the node * @param arg the acquire argument * @return {@code true} if interrupted while waiting */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 GCfailed = false;return interrupted;}if （shouldParkAfterFailedAcquire（p, node） &&parkAndCheckInterrupt（））interrupted = true;}} finally {if （failed）cancelAcquire（node）;}}该方法的主要作用就是将已经进入虚拟队列的节点进行阻塞，我们看到，如果当前节点的前驱节点是head并且尝试获取锁的时候成功了，则直接返回，不需要阻塞；如果前驱节点不是头节点或者获取锁的时候失败了，则进行判定是否需要阻塞：/** * Checks and updates status for a node that failed to acquire. * Returns true if thread should block. This is the main signal * control in all acquire loops.Requires that pred == node.prev. * * @param pred node"s predecessor holding status * @param node the node * @return {@code true} if thread should block */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;}这段代码对该节点的前驱节点的状态进行判断，如果前驱节点已经处于signal状态，则返回true，表明当前节点可以进入阻塞状态；否则，将前驱节点状态CAS置为signal状态，然后通过上层的for循环进入parkAndCheckInterrupt代码块park：/** * Convenience method to park and then check if interrupted * * @return {@code true} if interrupted */private final boolean parkAndCheckInterrupt（） {LockSupport.park（this）;return Thread.interrupted（）;}这个时候将该线程交给操作系统内核进行阻塞；总体来讲，acquireQueued就是依靠前驱节点的状态来决定当前线程是否应该处于阻塞状态，如果前驱节点处于cancel状态，则丢弃这些节点，重新构建队列；

Unlock API详解

流程类似lock api相关类的流程，这里讲主要的代码，unlock相对的比较简单首先 ReentrantLock 调用 Sync的release接口也就是AbstractQueuedSynchronizer的release接口/** * Releases in exclusive mode.Implemented by unblocking one or * more threads if {@link #tryRelease} returns true. * This method can be used to implement method {@link Lock#unlock}. * * @param arg the release argument.This value is conveyed to *{@link #tryRelease} but is otherwise uninterpreted and *can represent anything you like. * @return the value returned from {@link #tryRelease} */public final boolean release（int arg） {if （tryRelease（arg）） {Node h = head;if （h ！= null && h.waitStatus ！= 0）unparkSuccessor（h）;return true;}return false;}这个时候会先调用Sync的tryRelease，如果返回true，则释放锁成功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;}这个接口的作用很简单，如果不是获得锁的线程调用直接抛出异常，否则，如果当前state-releases==0也就是lock已经完全释放，返回true，清除资源；这个返回free之后，release拿到head节点，进入以下代码：/** * 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）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）;}这个作用即：当头结点的状态小于0，则将头结点的状态CAS为0，然后通过链表获取下一个节点，如果下一个节点为null或者不符合要求的状态，则从队尾遍历整个链表，直到遍历到离head节点最近的一个节点并且等待状态符合预期，则将头结点的后继节点置为该节点；对刚刚筛出来的符合要求的节点唤醒，也就是该节点获得争夺锁的权利；这就是非公平锁的特点：在队列一直等待的线程不一定比后来的线程先获得锁，至此，unlock 已经解释完成；凌渡冰目前就职于美团 Do what you like to impact the world.本文永久更新链接地址：http://www.linuxidc.com/Linux/2016-11/137285.htm