java锁机制ReentrantLock源码实例分析
目录
- 一:简述
- 二:ReentrantLock类图
- 三:流程简图
- 四:源码分析
- lock()源码分析:
- 非公平实现:
- 公平锁实现:
- tryAcquire()方法
- 公平锁实现:
- 非公平锁实现:
- addWaiter()
- acquireQueued()
- shouldParkAfterFailedAcquire()
- parkAndCheckInterrupt()
- unlock()方法源码分析:
- tryRelease()
- unparkSuccessor()
- 五:总结
一:简述
ReentrantLock是java.util.concurrent包中提供的一种锁机制,它是一种可重入,互斥的锁,ReentrantLock还同时支持公平和非公平两种实现。本篇文章基于java8,对并发工具ReentrantLock的实现原理进行分析。
二:ReentrantLock类图
三:流程简图
四:源码分析
我们以lock()方法和unlock()方法为入口对ReentrantLock的源码进行分析。
lock()源码分析:
ReentrantLock在构造构造方法创建对象的时候会根据构造函数传递的fair参数 创建不同的Sync对象(默认是非公平锁的实现),reentrantLock.lock()会调用sync.lock()。在Sync类中的lock()方法是一个抽象方法,具体实现分别在FairSync(公平)和NonfairSync(非公平)中。
public ReentrantLock() { //默认是非公平锁 sync = new NonfairSync(); } public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
public void lock() { sync.lock(); }
非公平实现:
final void lock() { // state表示锁重入次数 0代表无锁状态 //尝试抢占锁一次 利用cas替换state标志 替换成功代表抢占锁成功 if (compareAndSetState(0, 1)) //抢占成功将抢占到锁的线程设置为当前线程 setExclusiveOwnerThread(Thread.currentThread()); else acquire(1); }
公平锁实现:
final void lock() { acquire(1); }
可以看出非公平锁在lock()方法开始会尝试cas抢占一次锁,也就是在这里会插队一次。然后都会调用acquire()方法。acquire()方法中调用了三个方法,tryAcquire(),addWaiter(),acquireQueued()三个方法,而这三个方法正是ReentrantLock加锁的核心方法,接下来我们会对这三个方法进行重点的分析。
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) //这里acquireQueued()是一步步将线程是否中断的标志传回来的 如果为true 那么代表线程被中断过 需要设置中断标志 交给客户端处理 // 因为LockSupport.park()之后不会对中断进行响应 所以需要一步一步将中断标记传回来 selfInterrupt(); }
tryAcquire()方法
流程图:
tryAcquire()方法是抢占锁的方法,返回ture表示线程抢占到锁了,如果抢占到锁就什么都不做,直接执行同步代码,如果没有抢占到锁就需要将线程的信息保存起来,并且阻塞线程,也就是调用addWaiter()方法和acquireQueued()方法。
tryAcquire()也有公平和非公平锁两种实现。
公平锁实现:
protected final boolean tryAcquire(int acquires) { //获取当前线程 final Thread current = Thread.currentThread(); //获取state的值 int c = getState(); if (c == 0) { //如果state为0 代表现在是无锁状态 那么可以抢占锁 //公平锁这里多了一个判断 只有在AQS链表中不存在元素 才去尝试抢占锁 否则就去链表中排队 if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { //cas替换成功 那么就代表抢占锁成功了 将获取锁的线程设置为当前线程 setExclusiveOwnerThread(current); return true; } } //判断获取锁的线程是否是当前线程 如果是的话增加重入次数 (即增加state的值) else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) throw new Error("Maximum lock count exceeded"); setState(nextc); return true; } return false; } }
非公平锁实现:
protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); }
final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { //如果state为0 代表现在是无锁状态 非公平锁这里就直接尝试抢占锁 // 公平锁这里多了一个判断 先去看AQS链表中有没有已经在等待的线程 有的话就不会尝试去抢占锁了,非公平锁这里没有这个判断,也就是这里允许插队(不公平) if (compareAndSetState(0, acquires)) { //cas替换成功 那么就代表抢占锁成功了 将获取锁的线程设置为当前线程 setExclusiveOwnerThread(current); return true; } } //判断获取锁的线程是否是当前线程 如果是的话增加重入次数 (即增加state的值) else if (current == getExclusiveOwnerThread()) { int nextc = c + acquires; if (nextc < 0) // overflow throw new Error("Maximum lock count exceeded"); //重新设置state的值 setState(nextc); return true; } return false; }
接下来分析addWaiter()方法
addWaiter()
addWaiter()方法的作用就是将没有获取到锁的线程封装成一个Node对象,然后存储在AbstractQueuedSynchronizer这个队列同步器中(为了偷懒简称AQS)。我们先看一下Node对象的结构。
static final class Node { static final Node SHARED = new Node(); static final Node EXCLUSIVE = null; static final int CANCELLED = 1; static final int SIGNAL = -1; static final int CONDITION = -2; static final int PROPAGATE = -3; volatile int waitStatus; volatile Node prev; volatile Node next; volatile Thread thread; Node nextWaiter; final boolean isShared() { return nextWaiter == SHARED; } final Node predecessor() throws NullPointerException { Node p = prev; if (p == null) throw new NullPointerException(); else return p; } Node() { // Used to establish initial head or SHARED marker } Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; } }
通过Node对象的结构我们可以看出这是一个双向链表的结构,保存了prev和next引用,thread成员变量用来保存阻塞的线程引用,并且node是有状态的,分别是CANCELLED,SIGNAL,CONDITION,PROPAGATE,其中ReentrantLock涉及到的状态就是SIGNAL(等待被唤醒),CANCELLED(取消)(由于相关性不强,其他的状态在后续文章用到的时候在讲吧)。而AQS又保存了链表的头结点head和尾结点tail,所以实际上AQS存储阻塞线程的数据结构是一 个Node双向链表。addWaiter()方法的作用就是将阻塞线程封装成Node并且将其保存在AQS的链表结构中。
流程图:
源码分析:
private Node addWaiter(Node mode) { //将没有获得锁的线程封装成一个node Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; //如果AQS尾结点不为null 代表AQS链表已经初始化 尝试将构建好的节点添加到链表的尾部 if (pred != null) { node.prev = pred; //cas替换AQS的尾结点 if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } //没有初始化调用enq()方法 enq(node); return node; }
private Node enq(final Node node) { //自旋 for (;;) { Node t = tail; //尾结点为空 说明AQS链表还没有初始化 那么进行初始化 if (t == null) { // Must initialize //cas 将AQS的head节点 初始化 成功初始化head之后,将尾结点也初始化 //注意 这里我们可以看到head节点是不存储线程信息的 也就是说head节点相当于是一个虚拟节点 if (compareAndSetHead(new Node())) tail = head; } else { //尾结点不为空 那么直接添加到链表的尾部即可 node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
接下来分析acquireQueued()
acquireQueued()
acquireQueued()方法的作用就是利用Locksupport.park()方法将AQS链表中存储的线程阻塞起来。
流程图:
源码分析:
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; //进入自旋 for (;;) { //获取当前节点的前一个节点 final Node p = node.predecessor(); // 如果前一个节点是head 而且再次尝试获取锁成功,将节点从AQS队列中去除 并替换head 同时返回中断标志 if (p == head && tryAcquire(arg)) { setHead(node); p.next = null; // help GC failed = false; //注意 只有抢占到锁才会跳出这个for循环 return interrupted; } //去除状态为CANCELLED的节点 并且阻塞线程 线程被阻塞在这 //注意 线程被唤醒之后继续执行这个for循环 尝试抢占锁 没有抢占到的话又会阻塞 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { if (failed) //如果失败 将node状态修改为CANCELLED cancelAcquire(node); } }
在acquireQueued()方法中有两个方法比较重要shouldParkAfterFailedAcquire()方法和parkAndCheckInterrupt()方法。
shouldParkAfterFailedAcquire()
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) //如果节点是SIGNAL状态 不需要处理 直接返回 return true; if (ws > 0) { /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ //如果节点状态>0 说明节点是取消状态 这种状态的节点需要被清除 用do while循环顺便清除一下前面的连续的、状态为取消的节点 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. */ //正常的情况下 利用cas将前一个节点的状态替换为 SIGNAL状态 也就是-1 //注意 这样队列中节点的状态 除了最后一个都是-1 包括head节点 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }
parkAndCheckInterrupt()
private final boolean parkAndCheckInterrupt() { //挂起当前线程 并且返回中断标志 LockSupport.park(thread) 会调用UNSAFE.park()方法将线程阻塞起来(是一个native方法) LockSupport.park(this); return Thread.interrupted(); }
到这里lock()方法也就分析完了 接下来我们分析unlock()方法
unlock()方法源码分析:
流程图:
reentrantLock的unlock()方法会调用sync的release()方法。
public void unlock() { //每次调用unlock 将state减1 sync.release(1); }
release()方法有两个方法比较重要,tryRelease()方法和unparkSuccessor(),tryRelease()方法计算state的值,看线程是否已经彻底释放锁(这是因为ReentrantLock是支持重入的),如果已经彻底释放锁那么需要调用unparkSuccessor()方法来唤醒线程,否则不需要唤醒线程。
public final boolean release(int arg) { //只有tryRelease返回true 说明已经释放锁 需要将阻塞的线程唤醒 否则不需要唤醒别的线程 if (tryRelease(arg)) { Node h = head; //如果头结点不为空 而且状态不为0 代表同步队列已经初始化 且存在需要唤醒的node //注意 同步队列的头结点相当于是一个虚拟节点 这一点我们可以在构建节点的代码中很清楚的知道 //并且在shouldParkAfterFailedAcquire方法中 会把head节点的状态修改为-1 //如果head的状态为0 那么代表队列中没有需要被唤醒的元素 直接返回true if (h != null && h.waitStatus != 0) //唤醒头结点的下一个节点 unparkSuccessor(h); return true; } return false; }
tryRelease()
protected final boolean tryRelease(int releases) { //减少重入次数 int c = getState() - releases; //如果获取锁的线程不是当前线程 抛出异常 if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; //如果state为0 说明已经彻底释放了锁 返回true if (c == 0) { free = true; //将获取锁的线程设置为null setExclusiveOwnerThread(null); } //重置state的值 setState(c); //如果释放了锁 就返回true 否则返回false return free; }
unparkSuccessor()
private void unparkSuccessor(Node node) { //获取头结点的状态 将头结点状态设置为0 代表现在正在有线程被唤醒 如果head状态为0 就不会进入这个方法了 int ws = node.waitStatus; if (ws < 0) //将头结点状态设置为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. */ //唤醒头结点的下一个状态不是cancelled的节点 (因为头结点是不存储阻塞线程的) Node s = node.next; //当前节点是null 或者是cancelled状态 if (s == null || s.waitStatus > 0) { s = null; //从aqs链表的尾部开始遍历 找到离头结点最近的 不为空的 状态不是cancelled的节点 赋值给s 这里为什么从尾结点开始遍历而是头结点 应该是因为添加结点的时候是先初始化结点的prev的, 从尾结点开始遍历 不会出现prve没有赋值的情况 for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) //调用LockSupport.unpark()唤醒指定的线程 LockSupport.unpark(s.thread); }
五:总结
最后总结一下,有以下几点需要注意:
1.ReentrantLock有公平锁和非公平锁两种实现,其实这两种实现的差别只有两点,第一是在lock()方法开始的时候非公平锁会尝试cas抢占锁插一次队, 第二是在tryAcquire()方法发现state为0的时候,非公平锁会抢占一次锁,而公平锁会判断AQS链表中是否存在等待的线程,没有等待的线程才会去抢占锁。
2.AQS存储阻塞线程的数据结构是一个双向链表的结构,而且它是遵循先进先出的,因为它是从头结点的下一个节点开始唤醒,而添加新的节点的时候是添加到链表的尾部,所以AQS同时也是一个队列的数据结构。
3.线程被唤醒之后会继续执行acquireQueued()方法,因为它是阻塞在acquireQueued()方法的for循环中的,唤醒后尝试去获取锁,获取成功就会将节点从AQS中删除并跳出for循环,否则又会继续阻塞。(获取锁失败的原因就是因为有人插队啊。。也就是非公平锁导致的)。
以上就是java锁机制ReentrantLock源码实例分析的详细内容,更多关于java锁机制ReentrantLock的资料请关注我们其它相关文章!