摘要:当前线程已经获取过这个锁,则此时是重入,改变的计数即可,返回表示加锁成功。的核心在于使用更新锁的状态,并利用一个同步队列将获取锁失败的线程进行排队,当前驱节点解锁后再唤醒后继节点,是一个几乎纯实现的加锁与解锁。
简介
Java 并发编程离不开锁, Synchronized 是常用的一种实现加锁的方式,使用比较简单快捷。在 Java 中还有另一种锁,即 Lock 锁。 Lock 是一个接口,提供了超时阻塞、可响应中断以及公平非公平锁等特性,相比于 Synchronized,Lock 功能更强大,可以实现更灵活的加锁方式。
Lock 的主要实现类是 ReentrantLock,而 ReetrantLock 中具体的实现方式是利用另外一个类 AbstractQueuedSynchronizer,所有的操作都是委托给这个类完成。AbstractQueuedSynchronizer 是 Lock 锁的重要组件,本文从 AbstractQueuedSynchronizer 来分析 ReetrantLock 的实现原理。
基本用法先看一下 Lock 的基本用法:
Lock lock = ...; lock.lock(); try{ //处理任务 }catch(Exception ex){ }finally{ lock.unlock(); //释放锁 }
lock.lock() 即是加锁, lock.unolck() 是释放锁,为了保证所能够释放,unlock() 应该放到 finally 中。
下面分别从 lock() 和 unlock() 方法来分析加锁和解锁到底做了什么。
lock下面是 lock() 的代码:
public void lock() { sync.lock(); }
可以看到,只是简单调用了 sync 对应的 lock() 方法。那么这个 sync 是什么呢?其实这个就是 AbstractQueuedSynchronizer 的实现类。可以看一下 ReentrantLock 的构造方法:
/** * Creates an instance of {@code ReentrantLock}. * This is equivalent to using {@code ReentrantLock(false)}. */ public ReentrantLock() { sync = new NonfairSync(); } /** * Creates an instance of {@code ReentrantLock} with the * given fairness policy. * * @param fair {@code true} if this lock should use a fair ordering policy */ public ReentrantLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); }
ReentrantLock 有两个方法,主要目的是选择是公平锁还是非公平锁。公平锁指的是先来后到,先争抢锁的线程先获得锁,而非公平锁则不一定。ReentrantLock 默认使用的是非公平锁,也可以通过构造参数选择公平锁。选择哪个锁其实是生成了一个对象并赋值给变量 sync,下面是涉及到的代码:
/** * Base of synchronization control for this lock. Subclassed * into fair and nonfair versions below. Uses AQS state to * represent the number of holds on the lock. */ abstract static class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = -5179523762034025860L; /** * Performs {@link Lock#lock}. The main reason for subclassing * is to allow fast path for nonfair version. */ abstract void lock(); /** * Performs non-fair tryLock. tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */ 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; } 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 class final 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 } } // 非公平锁 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); } } /** * Sync object for fair locks * 公平锁 */ 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; } }
在 ReentrantLock 中有一个抽象的内部类 Sync,继承于 AbstractQueuedSynchronizer 并实现了一些方法。另有两个类 FairSync 和 NoFairSync 继承了 Sync,它们自然就是公平锁以及非公平锁的实现。下面分析将从公平锁出发,非公平锁与公平锁差别并不是很多。
公平锁 FairSync 加锁的代码如下:
final void lock() { acquire(1); }
只有一行,调用了 acquire,这是 AbstractQueuedSynchronizer 中的一个方法,代码如下:
public final void acquire(int arg) { if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
acquire 的实现也很短,不多在其中却包含了加锁的具体实现,关键就在内部调用的几个方法中。
为了理解加锁和解锁的过程,下面具体介绍一下 AbstractQueuedSynchronizer(以下简称 AQS)。
AbstractQueuedSynchronizerAQS 中使用一个同步队列来实现线程同步状态的管理,当一个线程获取锁失败的时候, AQS将此线程构造成一个节点(Node)并加入同步队列并且阻塞线程。当锁释放时,会从同步队列中将第一个节点唤醒并使其再次获取锁。
同步队列中的节点用来保存获取锁失败的线程的相关信息,包含如下属性:
static final class Node { /** Marker to indicate a node is waiting in shared mode */ // 标识共享模式 static final Node SHARED = new Node(); /** Marker to indicate a node is waiting in exclusive mode */ // 标识独占模式 static final Node EXCLUSIVE = null; /** waitStatus value to indicate thread has cancelled. */ // 线程取消 static final int CANCELLED = 1; /** waitStatus value to indicate successor"s thread needs unparking. */ // 需要唤醒后继节点 static final int SIGNAL = -1; /** waitStatus value to indicate thread is waiting on condition. */ static final int CONDITION = -2; /** * waitStatus value to indicate the next acquireShared should * unconditionally propagate. */ static final int PROPAGATE = -3; // 节点状态,为上面的几个状态之一 volatile int waitStatus; // 前置节点 volatile Node prev; // 后继节点 volatile Node next; // 节点所表示的线程 volatile Thread thread; Node nextWaiter; ... }
Node 是 AQS 的内部类,其中包含一些属性标识一个阻塞线程的节点,包括是独占模式还是共享模式、节点的状态、前驱结点、后继结点以及节点所代表的线程。
同步队列是一个双向列表,在 AQS 中有这样几个属性:
/** * Head of the wait queue, lazily initialized. Except for * initialization, it is modified only via method setHead. Note: * If head exists, its waitStatus is guaranteed not to be * CANCELLED. */ // 头结点 private transient volatile Node head; /** * Tail of the wait queue, lazily initialized. Modified only via * method enq to add new wait node. */ // 尾节点 private transient volatile Node tail; /** * The synchronization state. */ // 锁的状态 private volatile int state;
其中,head 和 tail 分别指向同步队列的头结点和尾节点,state 标识锁当前的状态,为 0 时表示当前锁未被占用,大于 1 表示被占用,之所以是大于 1 是因为锁可以重入,每重入一次增加 1。同步队列的结构大致如下图:
了解了同步队列后,下面具体看看加锁和解锁的过程。
加锁final void lock() { acquire(1); } public final void acquire(int arg) { // 加锁的主要代码 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }
主要逻辑其实就是一行代码:if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)),tryAcquire 是尝试获取一下锁,为什么说是尝试呢?看代码:
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; }
tryAcquire 可以分为三条分支:
当前锁未被占用(getState() == 0),则判断是否有前驱结点,没有的话就用 CAS 加锁(compareAndSetState(0, acquires)),加锁成功则调用 setExclusiveOwnerThread(current) 标示一下并返回 true。
当前线程已经获取过这个锁,则此时是重入,改变 state 的计数即可,返回 true 表示加锁成功。
如果不是上面两种情况,那么说明锁被占用或者 CAS 没有抢过其它线程,则需要进入同步队列,返回 false 表示尝试加锁失败。
回到 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 这一行,如果 tryAcquire(arg) 返回 false 将会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)。先看一下 addWaiter(Node.EXCLUSIVE) ,这个方法的代码如下所示:
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; // 尾节点为空 if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }
其中主要逻辑是将当前线程包装为一个 Node 节点并加入同步队列。如果尾节点为空,则用 CAS 设置尾节点,如果入队失败则调用 enq(node),这个方法内部是一个循环,利用自旋 CAS 把节点加入同步队列,具体代码就不分析了。
在节点加入队列之后,执行的是 acquireQueued 方法,代码如下:
final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; // 这是一个无限循环 for (;;) { final Node p = node.predecessor(); // 如果前驱节点是 head,则尝试获取锁 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 实现了线程的睡眠与唤醒。在内部是一个无限循环,每次获取前驱节点,如果前驱结点是 HEAD,那么尝试去获取锁,获取成功则将此节点变为新的头结点并将原先的头结点出队。如果前驱节点不是头结点或者获取锁失败,那么就会进入 shouldParkAfterFailedAcquire 方法,判断是否进入睡眠,如果这个方法返回 true,则调用 parkAndCheckInterrupt 让线程进入睡眠状态。下面是 parkAndCheckInterrupt 的代码:
private final boolean parkAndCheckInterrupt() { LockSupport.park(this); // 线程在这一步进入阻塞状态 return Thread.interrupted(); }
对于 shouldParkAfterFailedAcquire 来说,如果前驱节点正常,那么会返回 true,表示当前线程应该挂起,如果前驱结点取消了排队,那么当前线程有机会抢锁,此时返回 false,并继续 acquireQueued 中的循环。
解锁相比于加锁,解锁稍微简单一点,看一下 unlock 的代码:
public void unlock() { sync.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; }
首先调用 tryRelease 解锁,如果解锁成功则唤醒后继结点,返回值表示是否成功释放锁。那为什么会解锁不成功,其实是因为重入,看一下 tryRelease 的代码:
protected final boolean tryRelease(int releases) { // 更新 state 计数值 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; }
将 state 减去对应的值,如果 state == 0,那么说明锁已经完全释放。
在 release 中,如果锁已经完全释放,那么将调用 unparkSuccessor 唤醒后继节点,唤醒的节点所代表的线程阻塞在 parkAndCheckInterrupt 中:
private final boolean parkAndCheckInterrupt() { LockSupport.park(this); // 线程在这一步进入阻塞状态 return Thread.interrupted(); }
线程被唤醒后,将继续 acquireQueued 中的循环,尝试获取锁。
总结本文简要分析了 Lock 锁的原理,主要是利用 AbstractQueuedSynchronizer这个关键的类。AQS 的核心在于使用 CAS 更新锁的状态,并利用一个同步队列将获取锁失败的线程进行排队,当前驱节点解锁后再唤醒后继节点,是一个几乎纯 Java 实现的加锁与解锁。
如果我的文章对您有帮助,不妨点个赞支持一下(^_^)
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/76574.html
摘要:有了这个基础,才能发挥作用,使得在节点取消和异常时能够保证队列在多线程下的完整性。 Doug Lea是JDK中concurrent工具包的作者,这位大神是谁可以自行google。 本文浅析ReentrantLock(可重入锁)的原理 Lock接口 showImg(https://segmentfault.com/img/bV2671?w=276&h=176); Lock接口定义了这几个...
摘要:有了这个基础,才能发挥作用,使得在节点取消和异常时能够保证队列在多线程下的完整性。 Doug Lea是JDK中concurrent工具包的作者,这位大神是谁可以自行google。 本文浅析ReentrantLock(可重入锁)的原理 Lock接口 showImg(https://segmentfault.com/img/bV2671?w=276&h=176); Lock接口定义了这几个...
摘要:有了这个基础,才能发挥作用,使得在节点取消和异常时能够保证队列在多线程下的完整性。 Doug Lea是JDK中concurrent工具包的作者,这位大神是谁可以自行google。 本文浅析ReentrantLock(可重入锁)的原理 Lock接口 showImg(https://segmentfault.com/img/bV2671?w=276&h=176); Lock接口定义了这几个...
摘要:的主要功能和关键字一致,均是用于多线程的同步。而仅支持通过查询当前线程是否持有锁。由于和使用的是同一把可重入锁,所以线程可以进入方法,并再次获得锁,而不会被阻塞住。公平与非公平公平与非公平指的是线程获取锁的方式。 1.简介 可重入锁ReentrantLock自 JDK 1.5 被引入,功能上与synchronized关键字类似。所谓的可重入是指,线程可对同一把锁进行重复加锁,而不会被阻...
摘要:开始获取锁终于轮到出场了,的调用过程和完全一样,同样拿不到锁,然后加入到等待队列队尾然后,在阻塞前需要把前驱结点的状态置为,以确保将来可以被唤醒至此,的执行也暂告一段落了安心得在等待队列中睡觉。 showImg(https://segmentfault.com/img/remote/1460000016012467); 本文首发于一世流云的专栏:https://segmentfault...
阅读 1206·2021-11-24 09:39
阅读 2129·2021-11-22 13:54
阅读 2110·2021-09-08 10:45
阅读 1443·2021-08-09 13:43
阅读 2985·2019-08-30 15:52
阅读 3083·2019-08-29 15:38
阅读 2847·2019-08-26 13:44
阅读 3054·2019-08-26 13:30