JUC-AQS原理流程

冷不防 2022-12-27 09:30 160阅读 0赞

**AQS指AbstractQueuedSynchronizer, 抽象的队列同步器**

是用来构建锁或者其他同步器组件的重量级基础框架及整个JUC体系的基石,通过内置的FIFO队列来完成资源获取和线程的判断工作,并通过一个int类型变量,表示持有锁的状态

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70][]

AQS使用一个volatile的int类型成员变量来表示同步状态,通过内置的FIFO队列来完成资源获取和线程排队工作,将暂时获取不到锁的线程加入到队列中,这个队列就是AQS的抽象表现.

它将请求共享资源的线程封装成队列的节点Node,通过CAS、自旋以及LockSupport.park()的方式维护Sate变量的状态,使并发达到同步控制的效果

sate变量+CLH变种的双端队列

AQS同步队列的基本结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 1][]

// AbstractQueuedSynchronizer内部类Node
            static final class Node {
                static final Node SHARED = new Node(); // 共享
                static final Node EXCLUSIVE = null; // 独占
    
                static final int CANCELLED =  1;
                static final int SIGNAL    = -1;
                static final int CONDITION = -2;
                static final int PROPAGATE = -3;
    
                volatile int waitStatus;
    
                volatile Node prev;
    
                volatile Node next;
    
                volatile Thread thread;
    
                Node nextWaiter;
        }

**AQS源码主要方法**

**由lock()开始**

// 分为公平FairSync和非公平NonfairSync
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

**acquire()**

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

**tryAcquire(arg); arg=1**

final boolean nonfairTryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 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;
            }

**addWaiter(Node.EXCLUSIVE)**

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;
                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) { // Must initialize 创建哨兵节点
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }

**acquireQueued(addWaiter(Node.EXCLUSIVE)**

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false; // 中断
                for (;;) {
                    final Node p = node.predecessor(); // prev节点
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    // 取消排队
                    cancelAcquire(node);
            }
        }
    
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            // 获取前驱节点的状态
            int ws = pred.waitStatus; // waitStatus默认0
            // 如果是SIGNAL状态,即等待被占用的资源释放,返回true.程序会继续向下执行parkAndCheckInterrupt()方法,将当前线程park挂起
            if (ws == Node.SIGNAL) // Node.SIGNAL=-1
                return true;
            if (ws > 0) {
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                // 将当前节点的前驱节点状态设置为SIGNAL,用于后续唤醒操作
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
    
        private final boolean parkAndCheckInterrupt() {
            // 线程被阻塞,进入队列等待
            LockSupport.park(this);
            // 根据park方法API描述,程序在下面三种情况会继续向下执行
            // 1.被unpark
            // 2.被中断interrupt
            // 3.其他不合逻辑的返回才会继续向下执行
            // 因上述三种情况程序执行至此,返回当前线程的中断状态,并清空中断状态
            // 由于被中断,该方法会返回true
            return Thread.interrupted();
        }

addWaiter()入队流程图,双向链表中,第一个节点为虚节点(哨兵节点),不存储任何信息,只是占位.真正的第一个有数据的节点,是从第二个节点开始的.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 2][]

**由unlock()释放锁**

public void unlock() {
            sync.release(1);
        }
    
        public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    
            protected final boolean tryRelease(int releases) {
                int c = getState() - releases; // 1-1=0
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }

**unparkSuccessor()**

private void unparkSuccessor(Node node) {
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                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);
        }

**阻塞的线程被唤醒后,会继续执行parkAndCheckInterrupt()方法返回false**

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

**接着进入acquireQueued()方法继续自旋,返回false**

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false; // 中断
                for (;;) {
                    final Node p = node.predecessor(); // prev节点
                    // 此时tryAcquire()成功返回true
                    if (p == head && tryAcquire(arg)) {
                        // 设置新的head节点
                        setHead(node);
                        // 旧的哨兵节点没有引用,GC回收
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    // 取消排队
                    cancelAcquire(node);
            }
        }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70]: /images/20221120/bb42f6addbc5465682b5e340dce99038.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201210164025472.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201211102749142.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201211150002607.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5OTQwMjA1,size_16,color_FFFFFF,t_70