您现在的位置是:首页 > 文章详情

Java并发编程之ReentrantLock源码分析

日期:2018-12-22点击:269

ReentrantLock介绍

从JDK1.5之前,我们都是使用synchronized关键字来对代码块加锁,在JDK1.5引入了ReentrantLock锁。在JDK1.6之前synchronized关键字性能比ReentrantLock锁要差,JDK1.6之后性能基本是持平,但ReentrantLock锁功能要比synchronized关键字功能强大。

特点

synchronized关键字和ReentrantLock锁都是重入锁,可重入锁是指当一个线程获取到锁后,此线程还可继续获得这把锁,在此线程释放这把锁前其他线程则不可获得这边锁。相比synchronized关键字,ReentrantLock锁具有锁获取超时和获取锁响应中断的特点。ReentrantLock锁还分公平锁和非公平锁,公平锁模式是按线程调用加锁的先后排队顺序获取锁,非公平锁模式是已经在排队中的线程按顺序获取锁,但是新来的线程会和排队中的线程进行竞争,并不保证先排先获取锁。

ReentrantLock 源码分析

ReentrantLock实现了java.util.concurrent.locks.Lock接口和java.io.Serializable接口,前者是对实现Java锁的一种规范,后者说明ReentrantLock可以序列化。 ReentrantLock定义了一个成员变量

private final Sync sync; 

Sync类型是ReentrantLock的内部类,继承至AbstractQueuedSynchronizer ,AbstractQueuedSynchronizer是一个带空头的双向列表,为ReentrantLock的锁排队提供了基础支持。 ReentrantLock的UML关系图如下

下面我们解析下ReentrantLock中几个常用方法。

lock()方法源码分析

lock()是ReentrantLock中最常用的方法,用来对代码块加锁。lock()先是调用Sync的lock()的方法,Sync#lock()实现分为非公平模式和公平模式,我们对这2个模式分别讲解

非公平模式

Sync#lock()非公平模式代码如下:

 final void lock() { //用CAS方法设置枷锁状态 if (compareAndSetState(0, 1)) setExclusiveOwnerThread(Thread.currentThread()); else //抢锁失败,进入后续逻辑。 acquire(1); } 

新来线程先调用compareAndSetState(0, 1)方法用CAS方法设置加锁状态,这里是非公平模式实现要点,这样做主要是为了新来的线程和排队中的线程竞争,排队中的线程激活后也会用CAS方法设置加锁状态,就是看哪个线程线程抢的快,哪个能拿到锁。如果设置加锁状态成功,则设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果设置加锁状态失败即抢锁失败,则调用acquire(1)进入排队逻辑。

AbstractQueuedSynchronizer#acquire(int arg)实现代码如下:

 public final void acquire(int arg) { //先调用tryAcquire(arg)再试下能不能获取到锁,无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); } 

先调用tryAcquire(arg)再试下能不能获取到锁,获取成功则执行结束,无法获取则调用acquireQueued(addWaiter(Node.EXCLUSIVE), arg)进入排队,此方法返回参数为是否中断当前线程,排队过程中如果线程被中断则会返回ture,此时调用selfInterrupt()中断当前线程。

tryAcquire(arg)直接调用了非公平模式nonfairTryAcquire(acquires)方法我们看下实现:

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

调用getState()方法获取加锁状态,如果为0表示当前未被加锁,尝试CAS设置加锁状态获取锁,如果成功同样设置AbstractQueuedSynchronizer中的全局变量线程为当前当前线程。如果已被加锁,这判断当前线程和加锁线程是否是同一线程,如果是同一线程则将获取锁的状态加1返回获取锁成功,这里就是可重入锁实现的核心,状态的值表示当前线程重入了多少次,之后的释放锁就要释放相同的次数。

接下来我们看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)方法,acquireQueued主要功能是对当前线程阻塞,阻塞到能被上个获取到锁线程释放为止,addWaiter(Node.EXCLUSIVE)则是将当前线程加入到排队队列中。 我们先来看下addWaiter(Node.EXCLUSIVE)实现

 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; //CAS快速添加节点到尾部 if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } //如果尾节点不存在或者添加失败走最大努力添加节点逻辑 enq(node); return node; } private Node enq(final Node node) { for (;;) { Node t = tail; //如果头尾节点为空则创建空节点当头尾节点 if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else { //CAS添加节点到尾部 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } } 

创建已当前线程为基础的节点,先走快速添加到尾部逻辑,获取尾节点如果尾节点存在,将当前节点和尾节点相连,并用CAS方式将当前节点设置为尾节点,这边使用CAS方式考虑了多个线程同时操作尾节点的情况,所以如果尾节点已经变更则快速添加节点操作失败,调用enq(node)方法走最大努力添加节点的逻辑。enq(node)最大努力添加逻辑就是一直添加节点直到添加节点到尾部成功。

下面看下acquireQueued(addWaiter(Node.EXCLUSIVE), arg)的实现

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

acquireQueued里有个循环,这个循环的主要作用就是在线程激活后重试获取锁直到获取锁。node.predecessor()获取当前线程节点的前一个节点,如果是头节点,则当前线程尝试获取锁,获取锁成功设置当前节点为头节点。如果获取失败或者非头节点则调用shouldParkAfterFailedAcquire(p, node)判断是否需要阻塞等待,如果需要阻塞等待则调用parkAndCheckInterrupt()阻塞当前线程并让出cup资源直到被前一个节点激活,继续循环逻辑。

我们先来看下shouldParkAfterFailedAcquire(p, node)的实现

 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) return true; if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } 

先获取前个节点的状态,状态分以下4类

static final int CANCELLED = 1; static final int SIGNAL = -1; static final int CONDITION = -2; static final int PROPAGATE = -3; 

除了CANCELLED关闭状态是非正常,其他状态均正常状态。判断当前状态是否是SIGNAL正常状态,如果是就返回成功,这样当前线程就可以阻塞安心的等待上个节点的激活。如果状态为CANCELLED关闭状态则删除所有当前节点之前状态为CANCELLED的节点,返回失败让当前线程重试获取锁,如果是初始化0状态则CAS方式设置状态为SIGNAL。

接下来看下阻塞方法parkAndCheckInterrupt()

private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); } 

方法很简单调用LockSupport.park(this)阻塞当前线程,这里要讲下方法返回时调用Thread.interrupted()判断当前线程是否被中断,如果被中断的话,当前线程获取到锁后会调用Thread.currentThread().interrupt()中断线程。

公平模式

公平模式和非公平模式大部分代码相同,主要是获取锁的逻辑不同,我们就讲下代码不同的部分 lock()代码如下

final void lock() { acquire(1); } 

非公平模式模式先尝试设置状态来获取锁,而公平模式则直接调用acquire(1)去走排队逻辑。

尝试获取锁的方法tryAcquire(int acquires)也不一样代码如下

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

该方法跟非公平锁基本都一样,只是在获取锁的时候加了hasQueuedPredecessors()判断,这个方法主要判断了当前线程是否在头节点的下个节点,这样保证了获取锁的顺序性。

unlock()方法源码分析

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

先尝试释放锁,如果释放产品这判断当前节点是否为0不为0调用unparkSuccessor(h)方法激活下个节点的线程,否则直接返回。这里会有个疑问为什么h.waitStatus为0不去激活下个节点的线程,如果不激活下个节点的线程是否一直阻塞的,答案是否定的。这样做主要是为了释放锁的效率。waitStatus为0是初始化的值,这个值还没被下个节点线程调用shouldParkAfterFailedAcquire(p, node)方法设置成SIGNAL状态,也就说明下个节点线程还没被阻塞,此时如果下个节点线程调用此方法并设置成SIGNAL状态,势必它会重新获取锁,从而获取到锁避免了上述的问题。

下面来看下tryRelease(arg)方法

 protected final boolean tryRelease(int releases) { //重入次数减1 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; } 

因为锁可重入,因此调用getState()获取状态的值并减去一次重入次数,得到的c就是剩余重入的次数,然后判断当前释放的线程是否是当前占有锁的线程,如果不是抛出异常,否则先判断c是否为0表示当前线程持有的锁是否释放完全,如果是则设置持有锁的线程的变量为空,并设置锁状态为0,否则设置剩余的c到锁的状态。

接下来看下unparkSuccessor(h)的实现

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

获取当前节点状态,设置如果当前节点正常情况则设置成0,然后取当前节点的下个节点,如果下个节点状态非正常即CANCELLED状态,则从队列的尾部开始查找查到最靠近当前的节点且状态正常的节点,然后调用LockSupport.unpark(s.thread)通知此节点停止阻塞。这边会有个疑问如果调用LockSupport.unpark(s.thread)方法后,此节点才调用LockSupport.park(this)去阻塞,这样会不会发生此节点永久阻塞的问题,答案是否定的,LockSupport.unpark(s.thread)方法的实现其实是为线程设置了一个信号量,LockSupport.park(this)就算后调,如果线程相同也会收到此信号从而激活线程,这里的实现原理就不展开讲。

原文链接:https://my.oschina.net/u/945573/blog/2991876
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章