ReentrantLock
是一种可重入锁,可重入是说同一个线程可以多次获取同一个锁,内部会有相应的字段记录重入次数,它同时也是一把互斥锁,意味着同时只有一个线程能获取到可重入锁。
public ReentrantLock() { sync = new NonfairSync(); } public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
ReentrantLock
提供了两个构造函数,构造函数只是用来初始化 sync
字段,可以看到,默认情况下 ReentrantLock
使用的是非公平锁,当然,也可以使用带有布尔参数的构造函数来选择使用公平锁。公平锁和非公平锁的实现依赖于两个内部类: FairSync
和 NonfairSync
,接下来认识一下这两个类:
//非公平锁 static final class NonfairSync extends Sync { private static final long serialVersionUID = 7316153563782823691L; /** * Performs lock. Try immediate barge, backing up to normal * acquire on failure. */ final void lock() { if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } } static final class FairSync extends Sync { private static final long serialVersionUID = -3000897897090466540L; final void lock() { acquire(1); } /** * Fair version of tryAcquire. Don't grant access unless * recursive call or no waiters or is first. */ 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; } }
这两个内部类的代码都很短,并且都继承了另一个内部类 Sync
。这里先不急着介绍 Sync
类,因为这个类本身也并不复杂,后续在需要用到其中的方法时顺带讲解,目前只需要知道这个类继承了 AbstractQueuedSynchronizer(AQS)
即可。
lock()
public void lock() { sync.lock(); }
lock
方法提供了加锁的功能,公平锁和非公平锁的加锁操作是不一样的,先来看看非公平锁的细节,接着再讲解公平锁。
final void lock() { //使用CAS操作,尝试将state字段从0修改为1,如果成功修改该字段,则表示获取了互斥锁 //如果获取互斥锁失败,转入acquier()方法逻辑 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); } //设置获得了互斥锁的线程 protected final void setExclusiveOwnerThread(Thread thread) { exclusiveOwnerThread = thread; } public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
对于非公平锁来讲,使用 lock()
方法一上来就尝试获取互斥锁,获取成功就将 exclusiveOwnerThread
指向自己,代表当前是自己持有锁,否则就执行 acquire()
方法的逻辑,下面对 acquire()
方法的逻辑进行逐个分析。
首先是 tryAcquire()
方法,非公平锁重写了该方法,并在内部调用 Sync
类的 nonfairTryAcquire()
。
//从上面的逻辑来看,这里的acquires=1 protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); //c==0,说明当前处于未加锁状态,锁没有被其他线程获取 if (c == 0) { //在锁没有被其他线程占有的情况下,非公平锁再次尝试获取锁,获取成功则将exclusiveOwnerThread指向自己 if (compareAndSetState(0, acquires)) { setExclusiveOwnerThread(current); return true; } } //执行到这里说明锁已经被占有,如果是被自己占有,将state字段加1,记录重入次数 else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; //当nextc达到int类型最大值时会溢出,因此可重入次数的最大值就是int类型的最大值 if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } //执行到这里说明:1)锁未被占有的情况下,抢锁失败,说明当前有其他线程抢到了锁;2)锁已经被其他线程占有 //即只要当前线程没有获取到锁,就返回false return false; } //获取state字段,该字段定义在AQS中 protected final int getState() { return state; } //设置state字段 protected final void setState(int newState) { state = newState; }
当前线程没有在 tryAcquire()
方法中获取到锁时,会先执行 addWaiter(Node.EXCLUSIVE)
方法,其中参数 Node.EXCLUSIVE
是一个常量,其定义是 static final Node EXCLUSIVE = null
,作用是标记锁的属性是互斥锁。 addWaiter()
方法的作用是将当前线程包装成一个 Node
节点,放入等待队列的队尾,该方法在介绍 CountDownLatch
类时详细讲解过,有兴趣的朋友可以参考 ConcurrentHashMap源码探究(JDK 1.8)
,本文不再赘述。
将当前线程加入等待队列之后,会接着执行 acquireQueued()
方法,其源码如下:
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; //自旋 for (;;) { //获取当前节点的前一个节点 final Node p = node.predecessor(); //如果前一个节点是头节点,说明当前节点排在队首,非公平锁会则再次通过tryAcquire方法获取锁 if (p == head && tryAcquire(arg)) { //将自己设置为头节点 setHead(node); //前一个头结点没用了,会被垃圾回收掉 p.next = null; // help GC failed = false; //正常结束,返回false,注意该字段可能会在下面的条件语句中被改变 return interrupted; } //如果前一个节点不是头节点,或者当前线程获取锁失败,会执行到这里 //shouldParkAfterFailedAcquire()方法只有在p的状态是SIGNAL时才返回false,此时parkAndCheckInterrupt()方法才有机会执行 //注意外层的自旋,for循环体会一直重试,因此只要执行到这里,总会有机会将p设置成SIGNAL状态从而将当前线程挂起 //另外,如果parkAndCheckInterrupt()返回true,说明当前线程设置了中断状态,会将interrupted设置为true,代码接着自旋,会在上一个条件语句中返回true if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { //如果在自旋中线程被中断或者发送异常,failed字段的值将会为true,这里会处理这种情况,放弃让当前线程获取锁,并抛出中断异常 if (failed) cancelAcquire(node); } } //方法逻辑是:只有在前置节点的状态是SIGNAL时才返回true,其他情况都返回false private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) return true; //删除当前节点之前连续状态是CANCELLED的节点 if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } //线程在这里阻塞,并在被唤醒后检查中断状态 private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); } // private void cancelAcquire(Node node) { // Ignore if node doesn't exist if (node == null) return; node.thread = null; // Skip cancelled predecessors Node pred = node.prev; while (pred.waitStatus > 0) node.prev = pred = pred.prev; Node predNext = pred.next; node.waitStatus = Node.CANCELLED; // If we are the tail, remove ourselves. if (node == tail && compareAndSetTail(node, pred)) { compareAndSetNext(pred, predNext, null); } else { // If successor needs signal, try to set pred's next-link // so it will get one. Otherwise wake it up to propagate. int ws; if (pred != head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) { Node next = node.next; if (next != null && next.waitStatus <= 0) compareAndSetNext(pred, predNext, next); } else { //唤醒后一个节点 unparkSuccessor(node); } node.next = node; // help GC } }
注意 acquireQueued()
要么会抛出中断异常,要么正常结束返回 false
,只有在线程被唤醒后设置了中断状态才会返回 true
。对比可以发现, acquireQueued()
方法的逻辑与 CountDownLatch
中的 doAcquireSharedInterruptibly()
十分类似,许多方法在 CountDownLatch
这篇博客中讲到过,本文不再对这些方法进行赘述。
介绍完了 acquire()
方法,回过头来看看方法逻辑:
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
如果在 tryAcquire()
方法中没有获取锁,那么将当前线程加入到等待队列队尾,查看节点的前一个节点是否是头结点,是的话当前线程可以继续向下执行,否则就会阻塞挂起。当 acquireQueued
返回true时,说明线程设置了中断状态,就调用 selfInterrupt()
中断该线程,其他情况 selfInterrupt()
方法没机会执行。
到这里非公平锁的加锁流程已经介绍完了,由于代码逻辑比较长,且看源码的过程中会在好几个类中来回切换,思路很容易断,阅读代码的时候要注意。(有必要补个流程图)。
公平锁加锁逻辑
接下来看看公平锁的加锁逻辑:
final void lock() { acquire(1); }
与非公平锁相比,公平锁没有一上来就抢锁的逻辑,这也是公平性的体现。两种锁的 acquire()
方法的框架相同,但是实现细节不同,来看看公平锁的 tryAcquire()
方法:
protected final boolean tryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); //c=0表示当前没有其他线程持有锁 if (c == 0) { //下面的代码与非公平锁相比,多了hasQueuedPredecessors()方法的处理逻辑,公平锁只有在前面没有其他线程排队的情况下才会尝试获取锁 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; } //只要当前线程没有获取到锁,就返回false return false; } public final boolean hasQueuedPredecessors() { // The correctness of this depends on head being initialized // before tail and on head.next being accurate if the current // thread is first in queue. Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; //h != t表示等待队列中有其他节点 //h.next == null可能会有点费解,按理说h!=t之后,h后面肯定会有节点才对,这种情况其实已经见过,在上文介绍acquireQueued()方法时说过, //被唤醒的第一个等待节点会将自己设置为头结点,如果这个节点是队列中的唯一节点的话,它的下一个节点就是null //至于s.thread != Thread.currentThread()这个条件暂时可以忽略,因为公平锁执行到hasQueuedPredecessors方法时根本还没有入队, //这也意味着,只要队列中有其他节点在等候,公平锁就要求其他线程排队等待 return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); }
lockInterruptibly
lockInterruptibly
可以响应中断,来看看该方法的实现: public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); } public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); //先尝试获取锁,获取失败才执行后面的逻辑 if (!tryAcquire(arg)) doAcquireInterruptibly(arg); } private void doAcquireInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.EXCLUSIVE); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; return; } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
lockInterruptibly()
方法几乎与 acquire()
方法完全一样,唯一的区别是 acquire()
方法中, parkAndCheckInterrupt
因为线程设置了中断状态而返回 true
时,只是简单设置了一下 interrupted
字段的值,而 lockInterruptibly()
则是直接抛出异常。
unlock
方法
介绍完加锁的逻辑,接下来看看解锁的逻辑:
public void unlock() { sync.release(1); } public final boolean release(int arg) { //如果成功释放了锁,则执行下面的代码块 if (tryRelease(arg)) { Node h = head; //如果头节点不为null,请求节点状态不是初始状态,就释放头结点后第一个有效节点 //问题:这里为什么需要判断头结点的状态呢??? if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; } // protected final boolean tryRelease(int releases) { int c = getState() - releases; //线程没有持有锁的情况下,不允许释放锁,否则会抛异常 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; //可重入性的判断,如果释放了一次锁,使得c=0,就指针释放锁,做法是将记录锁的字段exclusiveOwnerThread重新指向null //注意,只有最后一次释放可重入锁,才会返回true if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; } //唤醒node节点的下一个有效节点,这里的有效指的是状态不是CANCELLED状态的节点 private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); 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); }
newCondition()
ReentrantLock
可以实现绑定多个等待条件,这个功能是在 newCondition()
方法中实现的,每次调用 newCondition()
方法时,都会产生一个新的 ConditionObject
对象,这是 AQS
中的一个内部类,代码很长,这里就不详细讨论了。来简单看看该方法的源码: public Condition newCondition() { return sync.newCondition(); } final ConditionObject newCondition() { return new ConditionObject(); }
在多线程环境中, ReentrantLock
的非公平锁要比公平锁拥有更高的性能,因为非公平锁避免了线程挂起产生的上下文切换的开销,但是公平锁能够避免线程饥饿问题,因此各有各的使用场景。从源码来看, J.U.C
包下的很多类都依赖 AQS
类,因此非常有必要搞懂 AQS
。提到 ReentrantLock
,总免不了与 synchronized
进行对比。 synchronized
也是可重入的,并且在 JDK 1.6
以后, synchronized
的性能已经得到了很大的提升,因此选择使用 ReentrantLock
一般是考虑使用它的三个优势:可中断、可实现公平锁、可绑定多个条件,这些优势是 synchronized
不具备的。