Java并发之互斥二：ReentrantLock（基于公平锁）
发布时间：2024年01月12日
1.构造方法
不传参数默认是非公平锁
根据传的fair是否公平来创建锁
- 基于公平锁来分析一下
``
所以最终我们又需要看到AQS这个类中，AQS是一个抽象的队列同步器
- AQS：
AQS内部类（对应同步队列的每一个节点）：
static final class Node {
// 枚举：共享模式
static final Node SHARED = new Node();
// 枚举：独占模式
static final Node EXCLUSIVE = null;
// 表示当前节点处于取消状态
static final int CANCELLED = 1;
// 表示当前节点需要它的后续节点（SIGNAL 表示其实是 后续节点的状态，需要当前节点去喊它）
static final int SIGNAL = -1;
// 表示当前节点处于等待状态
static final int CONDITION = -2;
// 共享模式下的节点状态
static final int PROPAGATE = -3;
// node状态（0，SIGNAL(-1)，CANCELLED(1)，CONDITION(-2)，PROPAGATE(-3)）
// waitStatus == 0 是默认状态
// waitStatus >0 取消状态
// waitStatus == -1 表示当前节点如何是head节点时，释放锁之后，需要唤醒它的后续节点
volatile int waitStatus;
// 因为node需要构建成fifo 队列，索引prev指向前继节点
volatile Node prev;
// 因为node需要构建成fifo 队列，索引prev指向后续节点
volatile Node next;
// 线程本尊
volatile Thread thread;
// reentrantLock 没有用到
Node nextWaiter;
/**
* Returns true if node is waiting in shared mode.
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* Returns previous node, or throws NullPointerException if null.
* Use when predecessor cannot be null. The null check could
* be elided, but is present to help the VM.
*
* @return the predecessor of this node
*/
final Node predecessor() {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
/** Establishes initial head or SHARED marker. */
Node() {}
/** Constructor used by addWaiter. */
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
THREAD.set(this, Thread.currentThread());
}
/** Constructor used by addConditionWaiter. */
Node(int waitStatus) {
WAITSTATUS.set(this, waitStatus);
THREAD.set(this, Thread.currentThread());
}
/** CASes waitStatus field. */
final boolean compareAndSetWaitStatus(int expect, int update) {
return WAITSTATUS.compareAndSet(this, expect, update);
}
/** CASes next field. */
final boolean compareAndSetNext(Node expect, Node update) {
return NEXT.compareAndSet(this, expect, update);
}
final void setPrevRelaxed(Node p) {
PREV.set(this, p);
}
// VarHandle mechanics
private static final VarHandle NEXT;
private static final VarHandle PREV;
private static final VarHandle THREAD;
private static final VarHandle WAITSTATUS;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
NEXT = l.findVarHandle(Node.class, "next", Node.class);
PREV = l.findVarHandle(Node.class, "prev", Node.class);
THREAD = l.findVarHandle(Node.class, "thread", Thread.class);
WAITSTATUS = l.findVarHandle(Node.class, "waitStatus", int.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
}
}
AQS属性：
// 头节点：任何时刻 头节点对应的线程都是当前持锁线程
private transient volatile Node head;
// 阻塞队列的尾节点 （同步队列不包含head节点，head.next ---> tail认为是阻塞队列）
private transient volatile Node tail;
// 表示资源
// 独占模式：0 表示未加锁状态，>0 表示已经加锁状态
private volatile int state;
继承父类AbstractOwnableSynchronizer的属性：
// 独占模式：表示持有锁的线程
private transient Thread exclusiveOwnerThread;
2.加锁 lock方法
- acquire方法：
// aqs定义的获取锁的方法
public final void acquire(int arg) {
// 条件一： tryAcquire 尝试获取锁，获取成功返回true，获取失败返回false
// 条件二： 2.1：addWaiter 将当前线程封装成node入队
// 2.2：acquireQueued 挂起当前线程 唤醒后相关的逻辑
// acquireQueued 返回true 表示挂起过程中线程被中断唤醒过 false表示未被中断过
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
// 再次设置中断标记
selfInterrupt();
}
- tryAcquire方法
// 抢占成功：返回true
// 抢占失败：返回false
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
/**
* Fair version of tryAcquire. Don't grant access unless
* recursive call or no waiters or is first.
*/
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {
// 当前线程
final Thread current = Thread.currentThread();
// AQS state值
int c = getState();
// 条件成立：c==0 表示当前AQS处于无锁状态
if (c == 0) {
// 条件一：
// 因为failSync是公平锁，任何时候都需要检查 同步队列中是否在当前线程之前有等待者
// hasQueuedPredecessors 返回true 表示当前线程前面有等待者，当前线程需要入队等待 返回false 表示无等待 直接尝试获取锁
if (!hasQueuedPredecessors() &&
// 条件二： compareAndSetState(0,acqures)
// 成功 说明当前线程获取锁成功 失败 说明存在竞争，且当前线程竞争失败
compareAndSetState(0, acquires)) {
// 设置当前线程为独占者 线程
setExclusiveOwnerThread(current);
return true;
}
}
// 执行到这里，有几种情况？
// c !=0 c>0 这种情况下就需要检查一下当前线程是不是独占锁的线程，因为reentrantlook是可以重入的
else if (current == getExclusiveOwnerThread()) {
// 锁重入的逻辑
int nextc = c + acquires;
// 越界判断，当重入的深度很深时，会导致 nextc<0,int值达到最大之后+1，会变成负数
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
// 更新操作
setState(nextc);
return true;
}
return false;
}
// 执行到这里时候会有哪些情况？
// 1.cas 失败了，表示当前线程在抢锁的时候，被别人抢到了
// 2.c>0 且ownerThread != currentThread 表示锁已经被别人抢了，但是获取锁的线程不是当前线程
}
- 如果现在没有锁（state=0）那就去抢锁，这里可以cas失败，被别人抢去了
- 如果锁已经被占用了，判断一下当前线程是不是获取锁线程，是的话，state+1
- 上面如果都没有执行的话，要么在判断state=0抢锁的时候没有抢到，要么就是锁被抢占了，并且自己不是获取锁线程
- addWaiter方法 （将当前线程封装成node入队）
private Node addWaiter(Node mode) {
// Node.EXCLUSIVE
// 构建node，把当前线程封装到对象node中
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
// 快速入队
// 获取队尾节点，保存到pred变量中
Node pred = tail;
if (pred != null) {
// 把当前节点的prev指向原来的尾节点
node.prev = pred;
if (compareAndSetTail(pred, node)) {
// 把原来的尾节点的next指向插入节点
pred.next = node;
return node;
}
}
// 什么时候会执行到这里？
// 1. 当前队列是空队列，tail == null (在判断pred != null)
// 2. cas竞争失败
// 完整入队
enq(node);
return node;
}
- 把当前线程封装成node节点
- 执行快速入队操作，首先看一下尾节点是不是空，就是等待队列空不空，如果不空的就cas操作，插入队尾，把当前节点的prev指向原来的tail，原来的tail指向插入节点
- enq方法 （自旋入队）
private Node enq(final Node node) {
// 自旋入队，只有当前node入队成功后，才会跳出循环
for (;;) {
Node t = tail;
// 1.第一种情况下：
// 当前队列是空队列，tail == null，说明当前线程被锁占用，且当前线程，
// 有可能是第一个获取锁失败的线程（当前可能有一批获取锁失败的线程）
if (t == null) { // Must initialize
// 作为当前持锁线程的第一个 后续线程，需要做什么事？
// 1.因为当前持锁的线程，它获取锁的时候，直接tryAcquire成功了，没有向同步队列插入node，所以作为后续需要给它
// 擦屁股
// CAS 成功，说明当前线程成功创建head节点
if (compareAndSetHead(new Node()))
tail = head;
// 注意：这里没有return，会继续for
} else {
// 第一个获取锁失败的线程同时还有再把自己加到tail上 + 其他普通节点的入队方式
//
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
- 如果上面快速入队失败了，就进入enq自旋入对，第一个持锁线程不会创建head节点（懒加载），所以队列还是空，第二个节点会帮助创建head节点，并且自己进入下一次自旋
- �第二个节点会把自己插入到tail后面，并把tail指向他
- acquireQueued方法 （当前线程没有被park，需要去挂起 唤醒之后的逻辑在哪里？ 唤醒之后的逻辑）
// 参数一：node就是当前线程包装出来的node，且当前时刻，已经入队成功了
// 参数二：当前线程抢占资源成功后，设置state值时，会用到
final boolean acquireQueued(final Node node, int arg) {
// true 表示当前线程抢占锁成功，普通情况下，lock早晚会拿到锁
// false 表示失败，需要执行出队的逻辑
boolean failed = true;
try {
// 当前线程是否被中断过
boolean interrupted = false;
// 自旋
for (;;) {
// 什么时候会执行这里？？
// 1、第一次进入for循环，在线程尚未被park之前会执行
// 2.线程park之后，被唤醒之后，会重新进入自旋
// 获取当前节点的前置节点
final Node p = node.predecessor();
// 条件一成立：p == head 说明当前节点为head.next节点，head.next节点任何时候都有权利去争取锁
// 条件二：tryAcquire(arg)
// 成立说明：head对应线程已经释放，head.next对应线程，正好获取到锁
// 不成立：说明head对应的线程 还没释放 head.next仍然需要park
if (p == head && tryAcquire(arg)) {
// 拿到锁之后，需要做什么？？？
// 设置自己为head节点
setHead(node);
// 将上个线程对应的node的next引用设置为null，协助老的head出队
p.next = null; // help GC
// 当前线程获取锁的过程没有异常，返回中断标记
failed = false;
return interrupted;
}
// shouldParkAfterFailedAcquire 这个方法是干嘛的？
// 当前线程获取锁资源失败后，是否需要挂起
// 如果是true 当前线程需要挂起 false 不需要
if (shouldParkAfterFailedAcquire(p, node) &&
// parkAndCheckInterrupt 这个方法什么作用？
// 挂起当前线程，并且唤醒之后，返回当前线程的中断标记（interrupted 唤醒方式：unpark 或者其他线程发起中断信号
parkAndCheckInterrupt())
// 表示当前node对应的线程是被中断信号唤醒的
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
- �
- shouldParkAfterFailedAcquire
// 参数一：pred 当前线程的前置节点
// 参数二：node 当前线程对应的node
// 返回值：boolean true表示当前线程需要挂起
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
// 获取前置节点的状态
// waitStatus：0 默认状态 new Node()、-1 Signal状态，表示当前节点释放锁之后会唤醒后面的节点、
// >0 表示当前节点是CANCELED状态
int ws = pred.waitStatus;
// 条件成立：表示前置节点 是个可以唤醒当前节点的节点，返回true => parkAndCheckInterrupt park当前线程
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
// 找爸爸的过程，结束条件是什么？前置节点 waitStatus <= 0
// 就是将双向队列其中cancel节点出队，让新加入的节点连上 SIGNAL节点
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 当前node前置节点的状态就是0的一种情况
// 将当前线程node的前置节点设置成SIGNAL，表示前置节点释放锁之后要喊醒我
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
- 当前节点的前置节点是 取消状态时，第一来到这个方法时，会越过取消状态的节点，第二次会返回true，然后park
- 当前节点的前置节点是0，当前线程会cas设置前置节点为-1，第二次来到返回true，然后park
shouldParkAfterFailedAcquire 这个方法确保前面的节点状态一定是-1，不管是跨过前面>0的，还是说先把前置节点设置成-1，当这一次return false再进来的时候，会看前置节点是不是-1，然后return true退出
- parkAndCheckInterrupt （park线程，并且在唤醒之后，返回为否为中断唤醒）
�
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
- cancelAcquire （取消指定node参与竞争）
// 取消指定node参与竞争
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
Node predNext = pred.next;
node.waitStatus = Node.CANCELLED;
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
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
}
}
3.unlock解锁过程
- AQS的release方法
public final boolean release(int arg) {
// 尝试释放锁，tryRelease 返回true，表示当前线程已经完成释放锁
// 返回false,说明当前线程尚未完全释放锁
if (tryRelease(arg)) {
// head节点什么情况下会创建出来？
// 当持锁线程未释放线程时，且持续期间，有其他线程想要获取锁，其他线程发现了锁，而且队列是空队列，此时后续线程会为当前
// 持锁线程构建出一个head节点，并把自己追加到head后面
Node h = head;
// 条件一成立：说明队列中的head节点已经初始化了，ReentrantLock在使用期间，发生过多线程竞争了
// 条件二成立：当前head后面一定插入了node节点
if (h != null && h.waitStatus != 0)
// 唤醒后续节点
unparkSuccessor(h);
return true;
}
return false;
}
- FailSync extends Sync 的tryRelease方法
protected final boolean tryRelease(int releases) {
// 减去释放的值
int c = getState() - releases;
// 条件成立：说明当前线程并未持锁，直接异常...
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
// 当前线程持有锁
// 是否已经完全释放锁
boolean free = false;
// 条件成立：说明当前线程已经达到完全释放锁的条件，c==0
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
// 更新state的值
setState(c);
return free;
}
- AQS的unparkSuccessor方法
// 唤醒当前节点的下一个节点
private void unparkSuccessor(Node node) {
// 获取当前节点（head）的状态，改成0是因为当前节点已经完成喊后续节点的任务
int ws = node.waitStatus;
// -1 Signal，改成0的原因：因为当前节点已经完成喊后续节点的任务
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
// s 是当前节点的第一个后续节点
Node s = node.next;
// 条件一：
// s 什么时候等于null
// 1.当前节点就是tail节点， s==null
// 2.当新节点入队未完成是，（1.设置新节点的prev 指向pred 2.cas设置新节点为tail 3.（未完成）pred.next->新节点）
// 条件二：成立 s.waitStatus >0 前提: s!= null
// 成立：说明 当前node节点的后续节点是 取消状态。。。 需要找一个合适的可以被唤醒的节点
//
if (s == null || s.waitStatus > 0) {
// 查找可以唤醒的节点
s = null;
// 从tail往前到head找一个离head最近的可以唤醒的节点，然后赋值给s
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
// 不管是head的后续节点还是 从tail往前找到离head最近可以唤醒的节点
if (s != null)
LockSupport.unpark(s.thread);
}
- 不过unparkSuccessor这个方法还是没有把head出队，然后被唤醒完的线程又走到如下代码：
被挂起的线程先从parkAndCheckInterrupt中醒来，然后进入外面的自旋，先判断前置节点是不是head，然后又去拿锁，将head至为空，完成head节点的出队操作
4. lockInterruptily 响应中断的加锁
- doAcquireInterruptibly
// 响应中断的获取锁
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);
}
}
上面parkAndCheckInterrput方法在接受到中断信号的时候， 会抛出InterruptedException，然后跳出方法，最终执行cancelAcquire方法
- cancelAcquire方法 （这一快的逻辑不看了有机会再研究）
// 取消指定node竞争
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
// 空判断
if (node == null)
return;
// 因为取消排队了，所以node内部的线程置为空
node.thread = null;
// 获取当前排队node的前驱
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
// 拿到前驱的next节点
// 1.当前node
// 2.可能也是cancel状态
Node predNext = pred.next;
// 将当前node状态设置为1.cancel状态
node.waitStatus = Node.CANCELLED;
// 当前取消排队的node所处位置不同，出队的策略是不同的，一共分为三种
// 1.当前node是队尾
// 2.当前node不是head.next也不是tail，
// 3.当前node是head.next节点
// 条件一成立：当前node是队尾，队尾出队
// 条件二成立：compareAndSetTail(node,pred) 成功，说明修改tail成功
if (node == tail && compareAndSetTail(node, pred)) {
// 修改pred.next = null 完成node出队
compareAndSetNext(pred, predNext, null);
} else {
// 保存节点节点状态
int ws;
// 当前node不是head.next也不是tail
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
}
}
文章来源:https://blog.csdn.net/LittleStar_Cao/article/details/135541149
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：chenni525@qq.com进行投诉反馈，一经查实，立即删除！
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：chenni525@qq.com进行投诉反馈，一经查实，立即删除！
最新文章
- Python教程
- 深入理解 MySQL 中的 HAVING 关键字和聚合函数
- Qt之QChar编码（1）
- MyBatis入门基础篇
- 用Python脚本实现FFmpeg批量转换
- 代码随想录算法训练营第十一天 | 20.有效的括号、1047. 删除字符串中的所有相邻重复项、150. 逆波兰表达式求值
- 2024年危险化学品生产单位安全生产管理人员证模拟考试题库及危险化学品生产单位安全生产管理人员理论考试试题
- ＜sa8650＞sa8650 qcxser-之-串行解串器GPIO的map
- 推荐EasyImages简单图床源码
- 句子中的最多单词数
- sphinx，一个神奇的 Python 库！
- python3处理xls并flask显示
- AutoCAD图纸打印后内容不见
- 用C语言实现动态数组Vector
- 网络基础面试题（四）