AQS源码探究之竞争锁资源

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(); } 

2. 然后我们调用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解除阻塞，下图即表示。

锁竞争失败流程

1. 进入acquire方法

public final void acquire(int arg) { 	// 首先再次请求锁     if (!tryAcquire(arg) &&         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))         selfInterrupt(); } 

2. 首先会执行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。需要了解 } 

3. 回到步骤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)) 

4. 执行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只是保证我们获得的尾部和现在的尾部是一样的，才能进行安全的尾部连接。

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

5. 执行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); } 

2. 调用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; } 

3. 继续回到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; } 

4. 进入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();     }  }