手写一个模拟的ReentrantLock

南七星之家 · 2022-8-24 07:03:10

package cn.daheww.demo.juc.reentrylock;
import sun.misc.Unsafe;
import java.lang.reflect.Field;
import java.util.concurrent.locks.LockSupport;
/**
* @author daheww
* @date 2022/7/7
*/
public class MiniReentryLock implements Lock {
/**
* 锁的是什么 --> 资源 --> state
* 0 --> 未加锁
* >0 -> 加锁
*/
private volatile int state;
/**
* 独占模式
* 同一时刻只有一个线程可以持有锁，其它线程在未获取到锁的时候会被阻塞
*
* 当前独占锁的线程（占用锁的线程）
*/
private Thread exclusiveOwnerThread;
/**
* 需要有两个节点去维护阻塞队列
* Head 指向队列的头节点
* Tail 指向队列的尾节点
*
* 比较特殊：Head节点对应的线程就是当前占用锁的线程
*/
private Node head;
private Node tail;
/**
* 获取锁
* 假设当前锁被占用，则会阻塞调用者线程，直到它抢占到锁为止
*
* 模拟公平锁
* --> 先来后到
*
* lock的过程
* 情景1.线程进来后发现，当前state == 0 --> 直接去抢锁
* 情景2.线程进来后发现，当前state > 0 --> 将当前线程入队
*/
@Override
public void lock() {
// 第一次获取到锁时，将state设置为1
// 第n次重入时，将state设置为n
acquire(1);
}
@Override
public void unlock() {
release(1);
}
private void release(int arg) {
// 条件成立：说明线程已经完全释放锁了
if (tryRelease(arg)) {
// 阻塞队列里面，还有睡觉的线程，应该唤醒一个线程
// 首先需要知道有没有等待的node --> head.next == null
Node head = this.head;
if (head.nx != null) {
// 公平锁，唤醒head.nx节点
unparkSuccessor(head);
}
}
}
private void unparkSuccessor(Node node) {
Node s = node.nx;
if (s != null && s.thread != null) {
LockSupport.unpark(s.thread);
}
}
/**
* 完全释放锁成功则返回true
*/
private boolean tryRelease(int arg) {
int c = getState() - arg;
if (getExclusiveOwnerThread() != Thread.currentThread()) {
throw new RuntimeException("must get lock first");
}
// 如果执行到这里，不存在并发，只会有一个线程会来到这里
// 条件成立，则说明当前线程持有的lock锁已经完全释放了
if (c == 0) {
this.exclusiveOwnerThread = null;
this.state = c;
return true;
} else {
this.state = c;
return false;
}
}
/**
* 竞争资源
* 1.尝试获取锁。成功则占用锁，且返回
* 2.抢占锁失败，阻塞当前线程
* @param arg
*/
private void acquire(int arg) {
if (!tryAcquire(arg)) {
// 抢锁失败
// step1.将当前线程封装成node，加入到阻塞队列中
Node node = addWaiter();
// step2.将当前线程park，使线程处于挂起状态
acquireQueued(node, arg);
}
// 抢锁成功
// 1.抢到了锁
// 2.重入了锁
}
/**
* 尝试抢锁失败，需要做的事：
* 1.需要将当前线程封装成node，加入到阻塞队列中
* 2.需要将当前线程park，使线程处于挂起状态
*
* 唤醒的流程：
* 1.检查当前node是否为head.next节点
* head.next是拥有抢占权限的线程，其它node都没有抢占的权限
* 2.抢占：
* 成功：
* 1.将当前node设置为node，将老的head出队，返回到业务层面
* 2.继续park等待被唤醒
*
* ----------------------------------------------
* 1.添加到阻塞队列的逻辑 addWaiter()
* 2.竞争资源的逻辑 acquireQueued()
*/
private void acquireQueued(Node node, int arg) {
// 当前线程已经放到queue中了
// 只有当前node成功获取到锁以后才会跳出自旋
for (; ; ) {
// 什么情况下，当前node被唤醒后可以尝试去获取锁呢？
// 只有一种情况，当前node是head的后继节点，才有这个权限
// 不是就先来后到
Node pvNode = node.pv;
// 条件1：pvNode == head
// true --> 说明当前node拥有抢占权限
// queue中的第一个节点代表的是当前锁正在执行的线程 --> head指向的线程
// head后面的线程代表的是正在排队的线程 --> 所以只有head.nx节点拥有抢锁的权利
// 条件2：tryAcquire(arg)
// true --> 当前线程获取到了锁
//
if (pvNode == head && tryAcquire(arg)) {
// 进入到这里面说明当前线程竞争锁成功了
// 需要做的操作：
// 1.设置当前head为当前线程的node
// 2.协助原来的对象出队
setHead(node);
pvNode.nx = null;
// 因为获取到了锁，所以就return了
return;
}
// 当前不是head.nx节点，或者去尝试获取锁失败了，这个时候都需要去把当前线程park掉
System.out.println("线程：" + Thread.currentThread().getName() + " 挂起");
LockSupport.park();
// 直到某个线程做了当前线程的unPark操作，这个线程才会继续执行
/*
所以总结一下，lock的逻辑就是：
1.在没锁的情况下，如果有个线程调用了lock方法，它就会改变lock中的state值。此时state值就不会为0了。
那么其它线程调用lock方法时，会看到这个state不为0。
2.然后这个线程会被封装成一个node节点
3.然后会去尝试竞争一下锁，做一下最后的挽救工作，如果实在挽救不了，就park了
--> 线程就在这个lock的lock()方法里被阻塞了。就达到了锁的效果
--> 所有调用这个锁对象lock的方法只能有一个线程能继续执行，然后其它线程会被阻塞，直到这个线程做了unlock操作
*/
System.out.println("线程：" + Thread.currentThread().getName() + " 唤醒");
// 什么时候唤醒被park的线程？--> unlock()
}
}
/**
* 把当前线程入队
* 返回当前线程对应的node节点
*
* addWaiter执行完成后，保证当前线程已经入队成功
*/
private Node addWaiter() {
Node newNode = new Node(Thread.currentThread());
// 如何入队？
// Case1.当前node不是第一个入队的node，队列已经有等待的node了
// 1.找到newNode的pv节点
// 2.更新newNode.pvNode = pv节点
// 3.CAS更新tail为newNode
// 4.更新pv节点
Node pvNode = tail;
if (pvNode != null) {
newNode.pv = pvNode;
// 条件成立，说明当前线程成功入队
if (compareAndSetTail(pvNode, newNode)) {
pvNode.nx = newNode;
return newNode;
}
}
// 执行到这里的几种情况
// 1.tail == null队列是空
// 2.cas设置当前newNode为tail时失败了 --> 循环入队 --> 自旋
enq(newNode);
return newNode;
}
/**
* 自旋入队，只有成功之后才返回
* 1.tail == null 队列是空队列
* 2.cas设置当前newNode为tail时失败了
*/
private void enq(Node node) {
for (; ; ) {
// 第一种情况：队列是空队列
// --> 当前线程是第一个抢占锁的线程...
// 当前持有锁的线程，并没有设置过任何node，所以作为该线程的第一个后驱节点
// 需要给他擦屁股
// 给当前持有锁的线程补充一个node作为head节点
// head节点任何时候都代表当前占用锁的线程
if (tail == null) {
// 条件成立：说明当前线程给当前持有锁的线程补充head操作成功了
if (compareAndSetHead(new Node())) {
tail = head;
// 注意，并没有直接返回，而是会继续自旋
}
} else {
// 当前队列中已经有node了，说明这是一个追加node的过程
// 如何入队呢？
// 1.找到newNode的pv节点 --> 最新的tail节点
// 2.更新newNode.pvNode = pv节点
// 3.CAS更新tail为newNode
// 4.更新pv节点
Node pvNode = tail;
node.pv = pvNode;
// 条件成立，说明当前线程成功入队
if (compareAndSetTail(pvNode, node)) {
pvNode.nx = node;
return;
}
}
}
}
/**
* 尝试获取锁，不会去阻塞线程
* true --> 抢占成功
* false --> 抢占失败
*/
private boolean tryAcquire(int arg) {
if (state == 0) {
// 当前state为0
// 不能直接抢锁 --> 公平锁 --> 先来后到
// 条件一：!hasQueuedPredecessors() ---> 取反之后为true，表示当前线程前面没有等待着的线程
// 条件二：compareAndSetState(0, arg) -> 使用cas的原因：lock方法可能有多线程调用的情况
// true --> 当前线程抢锁成功
// (1) volatile --> state被volatile修饰了，所以其它线程能第一时间知道这个值不为0了 --> 缓存能够一致了
// (2) cas -------> state从0变为arg的操作用cas实现，用于保证只会有一个线程能够改变state的值（0->arg） --> 只会有一个线程能够执行接下来的操作 --> 锁
// 1.如果cas的变量不用volatile修饰就没有意义：
// 因为A线程改变了state的值，但是B线程并不知道
// （可见性，volatile会让B线程中的副本马上失效，然后获取最新的state的值，此时B线程工作空间中的state值就不为0了）
// 2.如果volatile的变量不用cas去改变它的值的话，也没有意义：
//· step1.A线程，B线程都拿到了state的副本信息，此时state值为0
// step2.A线程改变了state的值。B线程还在写，因为state的值改变了，所以B线程工作空间中的state值改变，然后B继续写。
// 所以所有判断出state值为0的线程都能写成功，并且能执行写成功后续的操作
// 所以要用cas+volatile去保证只会有一个线程能够写成功这个值
// Ps.可以看到，如果这些线程想写的值都是同一个值的话，多写了几次，但是结果和只写一次是一致的
// cas+volatile主要还是去控制写成功之后的操作只会被执行一次，这样就像一个锁一样了
if (!hasQueuedPredecessors() && compareAndSetState(0, arg)) {
// 抢锁成功
// 1.将exclusiveOwnerThread设置为当前线程
this.exclusiveOwnerThread = Thread.currentThread();
return true;
// 不会入队任何node，接返回true
// 接下来第一个竞争失败的线程会先去帮忙创建一个node，然后再执行后续的操作
}
// 当前线程前面有线程在等待 || 多个线程和当前线程一起在尝试获取这个锁，然后当前线程失败了 --> return false;
} else if (Thread.currentThread() == this.exclusiveOwnerThread) {
// 执行的时机：
// 1.当前锁被占用
// 2.当前线程即为持锁线程
// 这里面不存在并发。只有当前加锁的线程才有权限修改state
// 即使是同一个线程多次进入到这，设置state的值，那么它们都是使用的同一个工作空间
// 不存在不同工作空间下，这个值的不一样的情况（因为没有了缓存）
// 锁重入的流程
int c = getState();
c += arg;
// TODO 越界判断
this.state = c;
return true;
}
// 什么时候会返回false？
// 1.cas加锁失败
// 2.state大于0，且当前线程不是持锁线程
return false;
}
/**
* 当前线程前面是否有等待着的线程
* true --> 当前线程前面有等待着的线程
* false -> 当前线程前面没有其它等待着的线程
*
* 调用链
* lock --> acquire -> tryAcquire -> hasQueuedPredecessors(state值为0时，即当前lock为无主状态)
*
* 什么时候返回false？
* 1.当前队列是空
* 2.当前线程为head.next节点 --> head.next在任何时候都有权力去争取lock
*/
private boolean hasQueuedPredecessors() {
Node h = head;
Node t = tail;
Node s;
// 条件一：h != t
// true --> 当前队列已经有node了
// false -> h == t
// case1. h == t == null --> 还没初始化过queue
// case2. h == t == head
// 第一个获取锁失败的线程会为当前持有锁的线程补充创建一个head node
// 条件二：
// 前置条件：条件一成立
// 排除几种情况：
// 条件2.1：极端情况 --> 第一个获取锁失败的线程，会为持锁的线程补充创建head节点，然后在自旋入队
// step1.cas设置tail成功了
// step2.head.next = node
// 在这两步中间的时候，有线程来检查前面是否有等待的线程
// 这种情况应该返回true：已经有head.next节点了，其它线程来这的时候需要返回true
// 条件2.2：
// 前置条件：h.next不是null
// true --> 条件成立说明当前线程就是持有锁的线程
// false -> 说明当前线程就是h.next节点对应的线程，需要返回false。回头线程就会去竞争锁了
return h != t && ((s = h.nx) == null || s.thread != Thread.currentThread());
}
private static final Unsafe UNSAFE;
private static final long STATE_OFFSET;
private static final long HEAD_OFFSET;
private static final long TAIL_OFFSET;
static {
try {
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
UNSAFE = (Unsafe) f.get(null);
STATE_OFFSET = UNSAFE.objectFieldOffset(MiniReentryLock.class.getDeclaredField("state"));
HEAD_OFFSET = UNSAFE.objectFieldOffset(MiniReentryLock.class.getDeclaredField("head"));
TAIL_OFFSET = UNSAFE.objectFieldOffset(MiniReentryLock.class.getDeclaredField("tail"));
} catch (Exception e) {
throw new Error(e);
}
}
private boolean compareAndSetHead(Node update) {
return UNSAFE.compareAndSwapObject(this, HEAD_OFFSET, null, update);
}
private boolean compareAndSetTail(Node expect, Node update) {
return UNSAFE.compareAndSwapObject(this, TAIL_OFFSET, expect, update);
}
private boolean compareAndSetState(int expect, int update) {
return UNSAFE.compareAndSwapInt(this, STATE_OFFSET, expect, update);
}
/**
* 阻塞的线程被封装成node节点，然后放进FIFO队列
*/
static final class Node {
/**
* 封装的线程本身
*/
Thread thread;
/**
* 前置节点引用
*/
Node pv;
/**
* 后置节点引用
*/
Node nx;
public Node(Thread thread) {
this.thread = thread;
}
public Node() {
}
}
public int getState() {
return state;
}
private void setHead(Node node) {
this.head = node;
// 当前线程已经是获取到锁的线程
node.thread = null;
node.pv = null;
}
public void setState(int state) {
this.state = state;
}
public Thread getExclusiveOwnerThread() {
return exclusiveOwnerThread;
}
public void setExclusiveOwnerThread(Thread exclusiveOwnerThread) {
this.exclusiveOwnerThread = exclusiveOwnerThread;
}
public Node getHead() {
return head;
}
public Node getTail() {
return tail;
}
public void setTail(Node tail) {
this.tail = tail;
}
}

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

		自动登录	找回密码
密码			立即注册

手写一个模拟的ReentrantLock

0 个回复

快速回复

楼主热帖

标签云