亲宝软件园·资讯

展开

ReentrantLock源码探究

纳兰小依 人气:1
`ReentrantLock`是一种可重入锁,可重入是说同一个线程可以多次获取同一个锁,内部会有相应的字段记录重入次数,它同时也是一把互斥锁,意味着同时只有一个线程能获取到可重入锁。 ## 1.构造函数 ``` 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)`即可。 ## 2.常用方法 - `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)](https://www.cnblogs.com/NaLanZiYi-LinEr/p/12391903.html),本文不再赘述。 将当前线程加入等待队列之后,会接着执行`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(); } ``` ## 3.总结 在多线程环境中,`ReentrantLock`的非公平锁要比公平锁拥有更高的性能,因为非公平锁避免了线程挂起产生的上下文切换的开销,但是公平锁能够避免线程饥饿问题,因此各有各的使用场景。从源码来看,`J.U.C`包下的很多类都依赖`AQS`类,因此非常有必要搞懂`AQS`。提到`ReentrantLock`,总免不了与`synchronized`进行对比。`synchronized`也是可重入的,并且在`JDK 1.6`以后,`synchronized`的性能已经得到了很大的提升,因此选择使用`ReentrantLock`一般是考虑使用它的三个优势:可中断、可实现公平锁、可绑定多个条件,这些优势是`synchronized`不具备的。

加载全部内容

相关教程
猜你喜欢
用户评论