ReentrantLock 互斥锁(独占锁)
锁
我们知道锁的基本原理是，基于将多线程并行任务通过某一种机制实现线程的串 行执行，从而达到线程安全性的目的。而Lock是juc中实现的锁接口，他定义了锁的一些行为规范，他的设计目的是为了解决synchronized关键字在一些并发场景下不适用的问题。
Lock
juc 包下的接口，定义了锁的规范。有多种实现类。
image-20200728144916359.png
ReentrantLock
ReentrantLock 重入锁 一个持有锁的线程，在释放锁之前。此线程如果再次访问了该同步锁的其他的方法，这个线程不需要再次竞争锁，只需要记录重入次数。重入锁的设计目的是为了解决死锁的问题
public class ReentrantLockDemo {
private static Lock lock = new ReentrantLock();
private static int count = 0;
private static void inrc() {
try {
//加锁
lock.lock();
Thread.sleep(10);
count++;
//模拟重入锁
dec();//2
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
private static void dec(){
lock.lock();
count--;
lock.unlock();
}
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 100; i++) {
new Thread(() -> {
inrc();//1
}).start();
}
TimeUnit.SECONDS.sleep(3);
System.out.println(count);
}
}
inr() 方法获取锁成功并没有释放锁的情况下调用dec()再次获取锁，假如没有重入锁的话这里会导致死锁。此线程如果再次访问了该同步锁的其他的方法，这个线程不需要再次竞争锁，只需要记录重入次数。
<span style='color:red'>内部是如何实现的？假如线程中断锁没有及时释放怎么办呢</span>
架构分析
- 在多线程环境下如何实互斥的逻辑？ 多线程环境下通过更改边界值state来实现获取锁与释放锁的逻辑。通过原子操作更改成员变量state，如果state=1则代表当前线程获取了锁其他线程阻塞。
- 获取锁失败的线程如何处理？获取锁失败的线程进行阻塞而不是循环。之所以阻塞是为了让这些获取锁失败的线程让出cpu资源
- 如何存储阻塞的线程呢？用一个双端列队queue存储线程
- 当前线程释放锁之后，如何唤醒阻塞的？释放锁的时候如果head节点的waitState= -1 那么通过LockSupport.unpark(thread) 唤醒header的后续waitState<= 0的节点
- 公平锁与公平锁什么区别？ 阻塞列队中的阻塞线程按照先进先出的顺序被唤醒的，第一个进来的第一个被唤醒去抢占锁，但是这个时候新的线程T2获取锁，那么T2并不是先插入阻塞列队中，而是先尝试获取一个锁，那么这种行为对于那些先进入阻塞列队中排队的线程来说是不公平的，就好比说大家排队买票一样，一个新来的没有排队直接去窗口买票了，所以叫做非公平锁反之则是公平锁.
使用
private final Sync sync; //内部类 实现类AbstractQueuedSynchronized抽象类
//默认实现是非公平锁
public ReentrantLock() {
sync = new NonfairSync();
}
//true 公平锁 false 非公平锁
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync() : new NonfairSync();
}
//获取锁
public void lock() {
sync.lock();
}
源码分析
获取锁
NonfairSync 重入锁的核心实现
//中间态变量，通过更改state的值判断是否占有锁 state=0 代表无线程占用
//state >= 1 代表有锁 state =1 代表有一个线程占用锁 state=5假如是重入锁那么代表这个线程获取了5次锁，相应的需要释放5次锁使得state=0从而达到释放的目的。
private volatile int state;
//当前占有锁的线程
private transient Thread exclusiveOwnerThread;
//尝试获取锁
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
compareAndSetState 通过调用unsafe的提供的cas本地方法(调用c语言) 尝试把state=0的值更改为1，如果成功设置占有锁的线程为当前线程，如果获取失败则调用acquire尝试获取1次
acquire
public final void acquire(int arg) {
//尝试获取一个锁，如果失败了添加线程到等待的双端队列中
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)){
selfInterrupt();
}
}
//尝试获取锁
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
//此处体现了nonfair非公平的意思
if (c == 0) {
//说明已经有线程释放了锁，那么当前线程尝试获取一次，如果获取成功返回true
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//次数体现了重入锁的逻辑
//如果获取锁的线程是当前线程
else if (current == getExclusiveOwnerThread()) {
//获取锁的次数+acquires
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
//这里无需cas操作，因为必然已经获取锁了
setState(nextc);
return true;
}
return false;
}
acquireQueued(addWaiter(Node.EXCLUSIVE), arg) 调用tryAcquire尝试获取锁失败的时候才会执行这个方法，先执行里面的类addWaiter(Node.EXCLUSIVE)
addWaiter
AQS中维护了一个存储了等待线程的Node节点的双端链表，有首节点head与尾节点tail，创建一个Node节点里面存储的是当前线程，如果已经有了tail节点则尝试cas操作添加当前节点到链表的尾结点，如果没有则进行初始化操作cas操作创建一个head节点并且自旋(没有任何结束条件的循环)cas操作添加尾结点到链表的尾部，最终返回新增的Node节点。
//线程被中断 或者 等待超时
static final int CANCELLED = 1;
//线程之间通讯的标识，是否需要唤醒后续节点 NodeA = -1 那么当NodeA释放锁的时候就需要唤醒被阻塞的后续节点NodeB
static final int SIGNAL = -1;
//
static final int CONDITION = -2;
static final int PROPAGATE = -3;
//节点的等待状态
volatile int waitStatus;
//添加当前线程到等待列队中
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;
if (pred != null) {
node.prev = pred;
// cas设置尾部节点
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) {
//初始化header节点
if (compareAndSetHead(new Node()))
tail = head;
} else {
//cas操作添加尾结点
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
image-20210407160316216.png
acquireQueued
对于插入到等待队列中的Node节点通过 addWaiter 方法把线程添加到链表后，会接着把 Node 作为参数传递给 acquireQueued 方法，去再次竞争锁
-
获取当前节点的 prev 节点
-
如果 prev 节点为 head 节点，那么它就有资格去争抢锁，调用 tryAcquire 抢占
锁
-
抢占锁成功以后，把获得锁的节点设置为 head，并且移除原来的初始化 head
节点
-
如果获得锁失败，则根据shouldParkAfterFailedAcquire决定是否需要挂起线程()
-
最后，通过 cancelAcquire 取消获得锁的操作
//等待队列中的线程尝试获取一个锁
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取当前节点的pre节点
final Node p = node.predecessor();
//如果当前节点是等待队列中的第2个节点 这里会再次尝试获取一次锁
if (p == head && tryAcquire(arg)) {
//成功了 设置当前节点为头节点 也就是说首节点是获取了锁的节点
setHead(node);
//这里帮助不必要的Node被gc回收
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
//线程挂起并让出cup,当被唤醒的时候返回中断状态
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
//取消锁
cancelAcquire(node);
}
}
shouldParkAfterFailedAcquire
//是否应该阻塞获取锁失败的节点 pred 前置节点 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;
}
shouldParkAfterFailedAcquire 判断节点是否被挂起，如果节点被挂起了那么一定是需要一种机制它前面的节点在释放锁的时候能够唤醒它，如果没有这种机制线程被挂起了没有唤醒岂不是有很大问题。这种机制就是通过设置前节点的共享变量waitStatus=-1来实现的。如果前节点为-1那么我就可以安心的挂起了因为前节点会在释放锁的时候唤醒我。如果前节点>0也就是被删除了，那么就删除处于waitStatus=1状态的节点直到找到一个前节点的waitStatus <=0并相连接。
我当时看到这里的时候一直不理解为什么要for (;;) 循环中判断是否需要阻塞呢？ 直接把当前Node对应的线程阻塞不就行了吗。假如TreadA 与 TheadB ，TreadA获取了锁，TreadB添加到等待队列中，那么这个时候TreadA释放了锁，
ThreadB进入到了acquireQueued方法还未阻塞并且还未获取锁，所以ThreadA不需要唤醒ThreadB，因为ThreadB没有阻塞，它会在acquireQueued的后续方法中获取锁的。
//释放锁逻辑 if (h != null && h.waitStatus != 0) unparkSuccessor(h);ThreadB进入到了acquireQueued方法的，这里已经获取锁失败了。TreadA锁已经被释放了，所以ThreadB调用shouldParkAfterFailedAcquire方法并设置前置节点的状态为-1并放回false不阻塞，为的是能够再次循环一遍尝试获取锁。如果这里设置节点状态为-1没有重新获取的话，会导致ThreadB一直阻塞不会获取锁，因为ThreadA已经释放了。
parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
挂起当前线程。这里调用了LockSupport.park(this)把线程挂起了并返回 Thread.interrupted()线程复位。
我这里一直不明白线程被中断后为什么调用 Thread.interrupted()进行复位然后节点继续去抢占锁。
if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) interrupted = true;这里只是单纯的标识了线程被中断过，在获取锁成功后会再次发起一次中断。
static void selfInterrupt() { Thread.currentThread().interrupt(); }线程被中断后直接退出不就行了吗？ 线程被干掉并不是简单的kill就完事了，直接删除线程是不合理的比如说一些内存资源没有被释放，所以Thread提供了一个interrupt()方法发送一个中断的信号，当开发人员收到这个信号之后去进行处理比如说释放资源然后执行完Runable方法让系统去回收Thread。所以在重入锁中业务代码需要明确收到中断信息后进行释放资源等操作，前提是线程需要获取锁，释放资源的操作肯定也是Lock.lock()中。所以AQS收到中断信号后并没有直接释放线程，而是记录标识然后继续在等待列队中尝试获取锁，获取锁之后调用 interrupt()方法再次中断。
image-20210407170850313.png
释放锁
更新state的值如果为0则唤醒后续节点
release
public final boolean release(int arg) {
if (tryRelease(arg)) {
//释放锁成功
Node h = head; //得到 aqs 中 head 节点
if (h != null && h.waitStatus != 0)
//如果 head 节点不 为空并且状态!=0.
//调用 unparkSuccessor(h) 唤醒后续节点
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease
释放锁的业务逻辑不需要考虑多线程的问题，他还是被一个线程持有。因为重入锁的机制state>=1 释放就是 getState() - releases并跟新state为最新值，如果state=0则返回。
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;
}
网友评论