AQS源码分析之独占式获取锁和释放锁

本是古典何须时尚 2022-12-09 11:55 228阅读 0赞

# AQS源码分析之独占式获取锁和释放锁 #

队列同步器AbstractQueuedSynchronizer（以下简称AQS），是用来构建锁或者其他同步组件的基础框架，ReentrantLock、ReentrantReadWriteLock和CountDownLatch等并发工具类底层都是通过AQS实现的。

## AQS的数据结构 ##

### 同步状态 ###

AQS内部使用一个被volatile关键字修饰的state属性来表示同步状态，并提供了下面三个方法来对这个同步状态进行操作：

private volatile int state; // 同步状态
    
    protected final int getState() {  // 获取当前同步状态
        return state;
    }
    
    protected final void setState(int newState) {  // 设置当前同步状态
        state = newState;
    }
    
    protected final boolean compareAndSetState(int expect, int update) {  // 使用cas来修改state，保证高并发情况下操作的原子性
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }

一般来说，state=0表示没有线程来获取锁，state=1表示有一个线程获取锁，state大于表示锁的重入次数。

疑问：在高并发下，不应该都是使用cas来修改state吗，为什么还要提供一个setState()方法？这里卖个关子，后面解答。

### 同步队列 ###

AQS依赖内部的同步队列（一个先进先出的双向链表）来完成同步状态的管理，当前线程获取同步状态失败时，同步器会将当前线程以及等待状态等信息构造成为一个节点（Node）并将其加入同步队列，同时会阻塞当前线程，当同步状态释放时，会把首节点中的线程唤醒，使其再次尝试获取同步状态。

同步队列中的节点Node的数据结构：

static final class Node { 
    
        // 线程的等待状态
        // SIGNAL:当前节点的后继节点处于等待状态时,如果当前节点的同步状态被释放或者取消,必须唤起它的后继节点
        // CANCELLED: 一个节点由于超时或者中断需要在CLH队列中取消等待状态,被取消的节点不会再次等待
        // CONDITION: 当前节点在等待队列中,只有当节点的状态设为0的时候该节点才会被转移到同步队列
        // PROPAGATE: 下一次的共享模式同步状态的获取将会无条件的传播
        // waitStatus的初始值时0,使用CAS来修改节点的状态
        volatile int waitStatus;
    
        volatile Node prev; // 前驱节点
    
        volatile Node next; // 后继节点
    
        volatile Thread thread; // 当前节点对应的线程
    }

AQS有头节点和尾节点，这样就构成了一个双向链表：

private transient volatile Node head; // 头节点
    
    private transient volatile Node tail; // 尾节点

## 没有资源竞争，就一个线程获取锁 ##

先来看一个例子，然后根据例子追踪源码：

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 没有资源竞争，就一个线程获取锁 */
    public class NoCompetitionDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            Thread t1 = new Thread(() -> { 
                ReentrantLock reentrantLock = new ReentrantLock();
                reentrantLock.lock();
                try { 
                    System.out.println("do something...");
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t1");
            t1.start();
            t1.join();
        }
    }

下面的源码分析会根据上面的例子来，抓要点和主干，也就是没有进入的代码暂时不分析，屏蔽一些细节，后面通过其他的场景实例来分析，不然一开始就会很复杂，容易晕。

reentrantLock.lock()方法实现如下：

public void lock() { 
        sync.lock(); 
    }}

sync.lock()方法的实现有两个，一个是公平锁FairSync，一个是非公平锁NonfairSync，默认是非公平锁，这里先看非公平锁lock方法的实现：

final void lock() { 
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread()); // 获取锁成功，标记一下哪个线程获得了锁，便于后面的重入
        else
            acquire(1); // 获取锁失败
    }

在上面的例子中，没有资源竞争，就一个线程获取锁，这样只需要使用cas将state设置为1，获取锁成功，然后就直接返回了。

再来看一下释放锁的过程，ReentrantLock\#unlock：

public void unlock() { 
        sync.release(1);
    }

AbstractQueuedSynchronizer\#release：

public final boolean release(int arg) { 
        if (tryRelease(arg)) { 
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h); // 同步队列中暂时没有节点，不需要唤醒，t1不会进入
            return true;
        }
        return false;
    }

ReentrantLock.Sync\#tryRelease：

protected final boolean tryRelease(int releases) { 
        int c = getState() - releases; // state-1，如果没有重入，此时c=0
        if (Thread.currentThread() != getExclusiveOwnerThread()) // 不是持有锁的线程不能释放锁
            throw new IllegalMonitorStateException();
        boolean free = false;
        if (c == 0) { 
            free = true;
            setExclusiveOwnerThread(null); // exclusiveOwnerThread=null
        }
        setState(c); // state=0，释放锁的时候没有并发，不需要cas，疑问的答案在这。
        return free;
    }

AQS中锁的变化过程如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

上面的例子很简单，下面来看一个复杂的。

## 有资源竞争，多个线程抢锁 ##

package com.morris.concurrent.lock.reentrantlock.trace;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.ReentrantLock;
    
    /** * 有资源竞争，多个线程抢锁 */
    public class MultiThreadDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            ReentrantLock reentrantLock = new ReentrantLock();
            new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    System.out.println("t1 do something...");
                    TimeUnit.SECONDS.sleep(100); // 休眠，便于调试与观察
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t1").start();
    
            TimeUnit.SECONDS.sleep(1); // 让t1跑起来获得锁
    
            new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    System.out.println("t2 do something...");
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t2").start();
    
            TimeUnit.SECONDS.sleep(1); // 让t2跑起来获得锁
    
            new Thread(() -> { 
                reentrantLock.lock();
                try { 
                    System.out.println("t3 do something...");
                } finally { 
                    reentrantLock.unlock();
                }
            }, "t3").start();
        }
    
    }

上面的例子中，t1会一直持有锁，t2、t3只能进入同步队列等待t1释放锁后被唤醒，t1的加锁流程与之前的例子中的加锁流程一样，这里不再赘述。

t1加锁后，t2来加锁前，AQS中各个数据结构的值如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]

此时t2来加锁，他会向t1一样先尝试cas设置的值为1，结果失败了，因为锁被t1持有了，然后就会进入AbstractQueuedSynchronizer\#acquire方法：

public final void acquire(int arg) { 
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

acquire()采用了模板方法模式，tryAcquire()的具体实现在子类NonfairSync\#tryAcquire中：

protected final boolean tryAcquire(int acquires) { 
        return nonfairTryAcquire(acquires);
    }

ReentrantLock.Sync\#nonfairTryAcquire，这里的条件都不满足，所以返回false：

final boolean nonfairTryAcquire(int acquires) { 
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {  // t2进来时state=1，这里不会进入
            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;
    }

所以上面的NonfairSync\#tryAcquire会返回false，然后就会执行AbstractQueuedSynchronizer\#addWaiter：

private Node addWaiter(Node mode) { 
        Node node = new Node(Thread.currentThread(), mode); // 创建一个Node节点
    
        Node pred = tail;
        if (pred != null) {  // t2进来的时候，同步队列还没有被初始化，head=tail=null这里不会进入
            node.prev = pred;
            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) {  // t2第一遍循环进来这个if，初始化队列
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else { 
                node.prev = t;
                if (compareAndSetTail(t, node)) {  // t2第二遍循环进来这个else，将node添加到双向链表的尾部
                    t.next = node;
                    return t;
                }
            }
        }
    }

此时，队列大概长这样：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]

然后程序接着往下执行，会进入到AbstractQueuedSynchronizer\#acquireQueued：

final boolean acquireQueued(final Node node, int arg) {  // node为t2
        boolean failed = true;
        try { 
            boolean interrupted = false;
            for (;;) { 
                final Node p = node.predecessor(); // 拿到t2的前驱节点，也就是h
                if (p == head && tryAcquire(arg)) {  // h是头节点，所以又会调用tryAcquire再次尝试获取锁，这里还是失败，不会进if
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

AbstractQueuedSynchronizer\#shouldParkAfterFailedAcquire

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {  // pred为h，node为t2
        int ws = pred.waitStatus; 
        if (ws == Node.SIGNAL) 
            return true; // t2第二次进入，直接返回true
        if (ws > 0) { 
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL); // t1第一次进入，将h的waitStatus改为Node.SIGNAL（-1），也就是告诉h，你释放锁后要唤醒我
        }
        return false;
    }

shouldParkAfterFailedAcquire这个方法t2第一次进来会把h的waitStatus改为-1，以便h释放锁后，可以通知t2，返回false，然后又会再次进入AbstractQueuedSynchronizer\#acquireQueued中的死循环，开始第二次自旋，此时t1还是没有释放锁，所以前面的执行结果还是一样，然后第二次进入shouldParkAfterFailedAcquire这个方法，直接返回true。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 3]

最后t2会进入AbstractQueuedSynchronizer\#parkAndCheckInterrupt，开始阻塞休眠，直接t1释放锁后，将它唤醒：

private final boolean parkAndCheckInterrupt() { 
        LockSupport.park(this); // 睡了
        return Thread.interrupted();
    }

下面来分析t3的加锁过程：

t3和t2的逻辑基本一样，有两个稍微不同的地方：

1.  t3加入队列的时候，队列已经初始化完成了，不需要再去初始化了。
2.  t3的前驱节点不是头结点，在AbstractQueuedSynchronizer\#acquireQueued方法不会再去尝试加锁了。

t3最后也会被进入到同步队列中，然后阻塞休眠，直接t2释放锁后，将它唤醒，此时同步队列中的结构如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 4]

下面来看一下t1解锁后是怎么唤醒t2的：

public final boolean release(int arg) { 
        if (tryRelease(arg)) { 
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h); // t1在释放锁时，进入此
            return true;
        }
        return false;
    }

AbstractQueuedSynchronizer\#unparkSuccessor：

private void unparkSuccessor(Node node) {  // 这个node是h，也就是头节点
    
        int ws = node.waitStatus; // -1
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0); // 先把-1改为0，表示不需要再通知了
    
        Node s = node.next; // s=t1
        if (s == null || s.waitStatus > 0) {  // s不会为null，不会进入
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread); // 唤醒t1
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 5]

t2之前是在AbstractQueuedSynchronizer\#parkAndCheckInterrupt里面睡的，被t1唤醒后，就会接着往下执行：

final boolean acquireQueued(final Node node, int arg) { 
        boolean failed = true;
        try { 
            boolean interrupted = false;
            for (;;) {  // t2醒来又会进入循环中
                final Node p = node.predecessor(); // p是头结点
                if (p == head && tryAcquire(arg)) {  // 这里获取锁会成功，因为t1已经释放了
                    setHead(node); // t2变成头结点
                    p.next = null; // 之前的头结点没有引用指向它，就会被GC回收了
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }
    
    private void setHead(Node node) { 
        head = node;
        node.thread = null;
        node.prev = null;
    }

t2获得了锁，同步队列的结构如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 6]

好了，后面的流程都类似。

## 公平锁的加锁流程 ##

ReentrantLock.FairSync\#lock，直接调用acquire方法。不像非公平锁，一上来就直接加锁，不成功才调用acquire：

final void lock() { 
        acquire(1);
    }

acquire方法中会尝试加锁，ReentrantLock.FairSync\#tryAcquire：

protected final boolean tryAcquire(int acquires) { 
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) { 
            if (!hasQueuedPredecessors() && // 只有这一行与非公平锁不一样
                compareAndSetState(0, acquires)) { 
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) { 
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }

在加锁之前得先判断同步队列中是否有线程在等待，如果有就不会去加锁，然后把自己加入到等待队列中等待：

public final boolean hasQueuedPredecessors() { 
        Node t = tail;
        Node h = head;
        Node s;
        return h != t &&
            ((s = h.next) == null || s.thread != Thread.currentThread());
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/507d359284524d6c9ccd278bac10026e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/f6bb70f7697b4363be842ede1cb7f660.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/75a27226e50144c290405c10a1f0464c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/ae879e14e24e468cbcb7fca5365e9e85.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/d57d957a5c004900a30735de07711b4a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/eb55413c8bd14af4a56d7575aac452c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/34f043b8462e40e791fa41b819de9aa2.png