ReentrantLock源码学习

ReentrantLock,我们称之为可重入锁。其中依赖了AbstractQueuedSynchronizer类来实现线程的同步。
ReentrantLock中定义了一个Sync的同步类,源码如下:
abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = -5179523762034025860L; //抽象方法 abstract void lock(); //非公平,尝试获取资源 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; }//独占方式,尝试释放资源,成功返回true,失败返回false 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; }protected final boolean isHeldExclusively() { // While we must in general read state before owner, // we don't need to do so to check if current thread is owner return getExclusiveOwnerThread() == Thread.currentThread(); }final ConditionObject newCondition() { return new ConditionObject(); }// Methods relayed from outer classfinal Thread getOwner() { return getState() == 0 ? null : getExclusiveOwnerThread(); }final int getHoldCount() { return isHeldExclusively() ? getState() : 0; }final boolean isLocked() { return getState() != 0; }/** * Reconstitutes the instance from a stream (that is, deserializes it). */ private void readObject(java.io.ObjectInputStream s) throws java.io.IOException, ClassNotFoundException { s.defaultReadObject(); setState(0); // reset to unlocked state } }

Sync类继承了AbstractQueuedSynchronizer,简称AQS。 AQS中提供了两种锁:
  • 独占锁,同一时刻只允许一个线程获得锁
    /** * 独占方式,尝试获取资源,成功返回true,失败返回false * @param arg * @return */ @Override protected boolean tryAcquire(int arg) { return super.tryAcquire(arg); }/** * 独占方式,尝试释放资源,成功返回true,失败返回false * @param arg * @return */ @Override protected boolean tryRelease(int arg) { return super.tryRelease(arg); }

  • 共享锁,同一时刻允许多个线程同时获得锁
/** * 共享方式,尝试释放资源,如果释放后允许唤醒后续等待节点则返回true,否则返回false * @param arg * @return */ @Override protected boolean tryReleaseShared(int arg) { return super.tryReleaseShared(arg); }/** * 共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功, 且有剩余可用资源。 * @param arg * @return */ @Override protected int tryAcquireShared(int arg) { return super.tryAcquireShared(arg); }

核心变量
/** * 同步状态变量 */ private volatile int state;

  • state > 0:表示有线程已经抢占到资源,但是并未释放,在重入的情况下state的值可能大于1
  • state = 0:表示当前锁资源处于空闲状态
//保证多线程竞争下state的原子性 protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }

ReentrantLock源码
  • 当调用ReentrantLock.lock()方法实际上是调用抽象静态内部类sync.lock()方法。
public void lock() { sync.lock(); }

syanc有两个具体的实现:
公平锁,必须按照FIFO的规则来访问锁资源
  • static final class FairSync extends Sync { private static final long serialVersionUID = -3000897897090466540L; final void lock() { acquire(1); } 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; } }

    非公平锁,可以不按照FIFO的规则,直接尝试获取锁资源,默认使用非公平锁
  • static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L; final void lock() { //不管当前线程是否排队,直接通过CAS抢占锁资源,如果成功则表示获取锁, //否则这调用 acquire(1)执行锁竞争的逻辑; if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } }

    acquire(int i)方法源码
/** 通过tryAcquire()方法尝试获取独占锁,如果成功则返回true,否则返回false。 如果tryAcquire()方法返回false,则说明当前锁被占用,只能通过addWaiter()方法将当前线程封装成Node并添加到AQS的同步队列中 acquireQueued()方法将Node作为参数,通过自旋去尝试获取锁 */ public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }

CAS实现原理
protected final boolean compareAndSetState(int expect, int update) { // 通过CAS乐观锁的方式来做比较并替换,如果当前内存中state的值和预期值expect相等,则更新为update。如果更新成功则返回true,否则返回false。 //这个操作是原子性的,不涉及state属性,也不会出现线程安全问题 return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }

state属性 state是AQS中的一个属性,它在不同的实现中所表达的含义是不一样的。对重入锁的实现来说,state表示同步状态,它有如下两个含义。
  • 当state=0时,表示无锁状态。
  • 当state>0时,表示已经有线程获得了锁,也就是说state=1,但是因为ReentrantLock允许重入,所以当同一个线程多次获得同步锁的时候,state会递增,比如重入5次,那么state=5。而在释放锁的时候,同样需要释放5次,直到state=0其他线程才有资格获得锁。
nonfairTryAcquire()方法源码
final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); //获取当前线程 int c = getState(); //获取state值 if (c == 0) {//等于0表示无锁 if (compareAndSetState(0, acquires)) {//CAS比较并替换state的值,成功则表示获取锁 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; }

nonfairTryAcquire()方法的实现逻辑如下。
判断当前锁的状态,c==0表示无锁,在无锁状态下通过compareAndSetState()方法修改state抢占锁资源。
○ 如果抢占成功,则返回true。
○ 如果抢占失败,则返回false。
current == getExclusiveOwnerThread(),该判断说明抢占到锁的线程和当前线程是同一个线程,表示线程重入,因此直接增加重入次数并保存到state字段中
AbstractQueuedSynchronizer.addWaiter(Node mode) 当tryAcquire()方法获取锁失败以后,会先调用addWaiter()方法把当前线程封装成Node加入同步队列中;源码如下
private Node addWaiter(Node mode) {//入参mode表示当前节点的状态,传递的参数是Node.EXCLUSIVE,表示独占状态。 Node node = new Node(Thread.currentThread(), mode); //把获取锁失败的线程封装成Node Node pred = tail; //tail在AQS中表示队列对尾的,默认为null if (pred != null) {//在tail不为null的情况下,队列中表示有节点 node.prev = pred; //把当前线程的Node的prev指向tail if (compareAndSetTail(pred, node)) {//通过CAS把node加入AQS队列中,也就是设置为tail pred.next = node; //把原来tail节点的next指向当前node return node; } } enq(node); //当tail=null时,把node添加到同步队列 return node; }

将当前线程封装成Node并进行存储,后续可以直接从节点中得到线程,再通过unpark(thread)方法来唤醒。
通过pred!=null判断当前链表是否已经完成初始化,如果已经完成初始化,则通过compareAndSetTail操作把当前线程的Node设置为tail节点,并建立双向关联。
如果链表还没初始化或者CAS添加失败(存在线程竞争),则调用enq()方法来完成添加操作。
enq()方法
private Node enq(final Node node) { for (; ; ) { Node t = tail; if (t == null) { // 如果为null则调用CAS初始化。直到成功初始化 if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }

该方法采用了自旋锁来完成同步队列的初始化,并把当前节点添加到了同步队列中。AQS的整体结构如图:
ReentrantLock源码学习
文章图片

ReentrantLock释放锁源码分析
public void unlock() { sync.release(1); }public final boolean release(int arg) { if (tryRelease(arg)) {//释放成功 Node h = head; //获取到AQS中的head节点 if (h != null && h.waitStatus != 0) //如果head不为null且状态不等于0,则调用 unparkSuccessor(h)方法唤醒后续节点 unparkSuccessor(h); return true; } return false; }

tryRelease(int releases) 通过修改state值来释放锁
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; }

独占锁在加锁时状态会加1,在释放锁时状态回减1,同一个锁可重入后,可能会递增,出现2,3,4,5这些值,只有调用unlock()方法的次数与调用lock()方法的次数相等,才会将ExclusiveOwnerThread线程设置为空,表示锁释放完毕
unparkSuccessor(Node node)唤醒同步队列中的线程
private void unparkSuccessor(Node node) {int ws = node.waitStatus; //获取head节点的状态 if (ws < 0) compareAndSetWaitStatus(node, ws, 0); //设置节点状态为0 Node s = node.next; //得到head节点的下一个节点 if (s == null || s.waitStatus > 0) { //如果下一个节点为null或者status>0,则cancelled状态 //通过从尾部节点开始扫描,找到距离head最近的一个waitStatus<=0的节点 s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null)//如果next节点不为空,则直接唤醒这个线程 LockSupport.unpark(s.thread); }

unparkSuccessor()方法主要有两个逻辑。
  • 判断当前节点的状态,如果节点状态已失效,则从tail节点开始扫描,找到离head最近且状态为SIGNAL的节点。
  • 通过LockSupport.unpark()方法唤醒该节点。
    为什么要从tail开始往前扫描?
    这和enq()方法有关系,在enq()方法的逻辑中,把一个新节点添加到链表中的逻辑如下。
    将新节点的prev指向tail。
    通过CAS将tail设置为新节点,因为CAS是原子操作,所以能够保证线程的安全性。
    t.next=node,目的是设置原tail的next节点指向新节点。
    如果在CAS操作之后、t.next=node操作之前,存在其他线程调用unlock()方法从head开始往后遍历,由于t.next=node还没执行,所以链表的关系还没有建立完整,就会导致遍历到t节点的时候被中断。而如果从tail往前遍历,就一定不会出现这个问题。
释放锁的线程继续执行 【ReentrantLock源码学习】回到AQS中的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); } }

    推荐阅读