当前位置 : 主页 > 编程语言 > 其它开发 >

AQS源码探究之竞争锁资源

来源:互联网 收集:自由互联 发布时间:2022-05-20
AQS源码探究---竞争锁资源 我们进入ReentrantLock源码中查看其内部类 Sync 对AQS进行扩展公共方法并定义抽象方法的抽象类 FaireSync 实现公平锁的AQS的实现类 UnFairSync 实现非公平锁的ASQ的实现
AQS源码探究---竞争锁资源

我们进入ReentrantLock源码中查看其内部类

  • Sync 对AQS进行扩展公共方法并定义抽象方法的抽象类
  • FaireSync 实现公平锁的AQS的实现类
  • UnFairSync 实现非公平锁的ASQ的实现类

我使用例子进行的debug,然后一步一步看源码。例子在文章最后面

以下流程皆以非公平锁为例

线程竞争锁资源

AQS的state解释:

  • 0 表示锁没有被占用
  • 1 表示锁被占用了
  • > 1 表示锁被重入了 PS: ReentrantLock是可重入锁
获得锁执行流程
  1. 创建ReentrantLock对象
// ReetrantLock 默认创建一个非公平锁的AQS
public ReentrantLock() {
    sync = new NonfairSync();
}
  1. 然后我们调用lock方法请求锁
    • 成功,即将锁的owner主人设置为当前线程,接下来就是回到线程中执行线程的任务。
    • 失败,即进入acquire的流程。
static final class NonfairSync extends Sync {
    
    final void lock() {
        // 请求锁资源,如果将锁的state状态0改成1,即为成功获得锁资源
        if (compareAndSetState(0, 1))
            // 将锁的拥有者设置为当前线程,里面就一句话没啥好看的
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
    
}

下面是AQS阻塞链表是由一个双向链表组成的。

阻塞链表的成员对象Node的waitState状态解释:

  • CANCELLED = 1 表示线程已经被取消了
  • SIGNAL = -1 表示后继线程需要unpark解除阻塞,下图即表示。

image

锁竞争失败流程
  1. 进入acquire方法
public final void acquire(int arg) {
	// 首先再次请求锁
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
  1. 首先会执行tryAcquire方法
protected final boolean tryAcquire(int acquires) { // 注意:我们进入的是非公平锁的tryAcquire实现
    return nonfairTryAcquire(acquires);  
}

再次进入nonfairTryAcquire(acquires)方法

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();  // 获得当前线程
    int c = getState();    // 获得当前线程的状态
    if (c == 0) {     // 如果状态为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;   // 失败返回false。如果是成功获得锁或者是重入都会返回true。需要了解
}
  1. 回到步骤1代码,如果是失败返回false取反true,就会继续执行if语句。成功取反后false就直接结束当前语句,就会直接回到线程执行线程代码了。
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) // 这里是两个方法,需要一个一个来
        selfInterrupt();
}
// acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  1. 执行addWaiter方法,概括就是将没有获得锁的加入一个等待链表中。
private Node addWaiter(Node mode) {  // 刚创建的时候mode为null的
    Node node = new Node(Thread.currentThread(), mode);  // 首先创建一个node
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;  // 将尾部的引用给pred变量
    if (pred != null) {   // 刚开始创建的时候pred是null的
        node.prev = pred;   
        if (compareAndSetTail(pred, node)) {   // 这个代码块就是cas尝试加入双向链表尾部
            pred.next = node;   
            return node;
        }
    }
    enq(node);    // 这里是创建head和tail进的方法,和if (compareAndSetTail(pred, node))失败进入
    return node; // 方法返回由当前线程创建的node
}

enq方法的进入条件

  • 进行head和tail的初始化。
  • 多线程下如果调用enq方法失败,就是当别的线程也进入了等待链表,此时tail就会改变,上面的cas就会false,没有返回,就会进行enq方法
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;  // 如果尾部为空就会进行初始化,没有的话不断进行cas尝试插入链表尾部。
        if (t == null) { // Must initialize 初始化链表
            if (compareAndSetHead(new Node()))   // 我们可以看到head是指向一个没有参数的node对象的
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) { 
                t.next = node; // 注意t还是引用旧值,而tail已经更新引用为node了。
                return t;  
            }
        }
    }
}

疑问:

compareAndSetTail(t, node) 方法在我初次遇见的时候很奇怪。为什么t还算指向了旧的node对象

因为这个compareAndSetTail只是将tail的引用改变成了node,注意这边改变的是tail的引用。并没有去改变pred的引用。传入pred只是保证我们获得的尾部和现在的尾部是一样的,才能进行安全的尾部连接。

这也是我基础不太扎实的原因吧。

  1. 执行acquireQueued方法,再次尝试获得锁,和进行阻塞
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;  
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();  // 获得node前驱
            if (p == head && tryAcquire(arg)) {  // 如果是第一个等待锁的线程,再次请求锁
                setHead(node);  // 请求成功就将该线程的node直接移出等待链表
                p.next = null; // help GC
                failed = false;
                return interrupted; 
            }
            if (shouldParkAfterFailedAcquire(p, node) &&   // 检查状态并更新前驱状态为-1,即表示有后继节点阻塞了。
                parkAndCheckInterrupt()) // 进入park,如果被中断返回true
                interrupted = true;      
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

在parkAndCheckInterrupt方法时进行park阻塞。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}
线程释放锁
  1. 调用unlock方法
public void unlock() {
    sync.release(1);
}
  1. 调用release方法
public final boolean release(int arg) {
    if (tryRelease(arg)) {  // 进入tryRelease即尝试释放
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

进入tryRelease的ReentrantLock实现

protected final boolean tryRelease(int releases) { 
    int c = getState() - releases;   // 获得当前的状态
    if (Thread.currentThread() != getExclusiveOwnerThread())  // 非获得锁线程抛异常
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {   // 如果没有重入直接释放锁将owner置为null
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);   // 由于锁资源只有一个只有一个线程能更新状态,所以更新AQS状态不需要cas
    return free;
}
  1. 继续回到release方法,释放锁成功返回true,进入条件语句
public final boolean release(int arg) {
    if (tryRelease(arg)) {  // 进入tryRelease即尝试释放
        Node h = head;
        if (h != null && h.waitStatus != 0) // 阻塞队列存在即头节点不为空且头节点的状态不为0,为0表示后面没节点阻塞了
            unparkSuccessor(h);   
        return true;
    }
    return false;
}
  1. 进入unparkSuccessor方法,就不贴源码了,简单介绍一下就是将头节点置空,将阻塞队列中第一个等待的node解除阻塞,将他放出来去抢锁资源。
非公平锁和公平锁的区别

看完源码,整明白了就是锁资源释放后会放第一个等待线程去抢锁。

我就疑惑了,那明明就是公平的啊。

其实只是释放了线程,但是同时有其他的线程进行争抢,就又会变成争抢的情况,还是可能被其他线程抢走锁资源。

公平锁

就会判断如果阻塞链表是否为空,为空才能进行获取锁资源,又或者是锁重入

不然就是直接加入阻塞链表,从而实现了公平。

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
DEBUG例子
@Slf4j
public class Test1 {

    public static void main(String[] args) {

        ReentrantLock lock = new ReentrantLock();
        // Reentrantlock锁资源被拥有
        new Thread(()->{
            lock.lock();
            try{
                log.debug("运行中");
                try {
                    Thread.sleep(2000000000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }finally {
                lock.unlock();
            }
        }).start();
        // ReentrantLock阻塞链表初始化
        new Thread(()->{
            lock.lock();
            try{
                log.debug("运行中");
                try {
                    Thread.sleep(2000000000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }finally {
                lock.unlock();
            }
        }).start();
        // ReentrantLock 再次向阻塞链表添加线程
        new Thread(()->{
            lock.lock();
            try{
                log.debug("运行中");
                try {
                    Thread.sleep(2000000000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }finally {
                lock.unlock();
            }
        }).start();
    }

}
上一篇:.net 项目使用 JSON Schema
下一篇:没有了
网友评论