并发编程-AQS

末蓝、 2023-02-28 05:54 86阅读 0赞

# AQS #

### 文章目录 ###

*  AQS
 *   *  一，简介
     *  二，AQS数据结构
     *  三，主要方法源码分析
     *  四，AQS实现类
     *   *  顶层锁接口
         *  ReentrantLock
         *  ReentrantReadWriteLock

## 一，简介 ##

> AQS是抽象队列同步器
> 
>  *  抽象：抽象类，只实现了一些主要逻辑，有些方法由子类实现
>  *  队列：使用先进先出队列存储数据
>  *  同步：可以实现同步功能
> 
> AQS是用来构建锁和同步器的框架，JUC包中很多同步工具类都基于AQS实现

## 二，AQS数据结构 ##

AQS内部使用一个volatile变量state来作为资源的标识，同时定义了几个获取和修改state的protectd方法，子类可以覆盖这些方法来实现自己的逻辑

protected final int getState()
    protected final void setState(int newState) 
    protected final boolean compareAndSetState(int expect, int update)

这三个方法都是原子操作，其中compareAndSetState方法底层依靠的是CAS来保证原子更新

protected final boolean compareAndSetState(int expect, int update) {
        
            // See below for intrinsics setup to support this
            return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
        }

AQS内部维护了一个等待获取同步状态的队列：

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6TNl39yz-1595237478449)(https://gblobscdn.gitbook.com/assets%2F-L\_5HvtIhTFW9TQlOF8e%2F-L\_5TIKcBFHWPtY3OwUo%2F-L\_5TJQjOPACL\_iNG1yE%2FAQS%E6%95%B0%E6%8D%AE%E7%BB%93%E6%9E%84.png?alt=media)\]

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;
    
            /**
             *判断是否是共享锁模式
             */
            final boolean isShared() {
        
                return nextWaiter == SHARED;
            }
    
            /**
             * 返回当前节点的前驱节点
             * @return the predecessor of this node
             */
            final Node predecessor() throws NullPointerException {
        
                Node p = prev;
                if (p == null)
                    throw new NullPointerException();
                else
                    return p;
            }
    
            Node() {
            // Used to establish initial head or SHARED marker
            }
    
            Node(Thread thread, Node mode) {
             // Used by addWaiter
                this.nextWaiter = mode;
                this.thread = thread;
            }
    
            Node(Thread thread, int waitStatus) {
         // Used by Condition
                this.waitStatus = waitStatus;
                this.thread = thread;
            }
        }

看源码的过程中发现了节点有几个状态：

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

**资源共享模式**

*  共享模式：资源可以同时被多个线程获取，具体的资源数可以通过参数设定，具体实现：Semaphore/CountDownLatch。
 *  独占模式：资源只能同一时间被一个线程获取

**AQS内部维护了两个队列，一个是同步队列，一个是等待队列**

## 三，主要方法源码分析 ##

> AQS的设计基于模板方法设计模式，有一些方案必须要子类去实现，AQS中没有实现

protected boolean tryAcquire(int arg) {
        
            throw new UnsupportedOperationException();
        }
        protected boolean tryRelease(int arg) {
        
            throw new UnsupportedOperationException();
        }
        protected int tryAcquireShared(int arg) {
        
            throw new UnsupportedOperationException();
        }
        protected boolean tryReleaseShared(int arg) {
        
            throw new UnsupportedOperationException();
        }
    	// 是否独占锁
        protected boolean isHeldExclusively() {
        
            throw new UnsupportedOperationException();
        }

对于加锁和解锁，AQS实现了一系列主要的逻辑：但是具体细节需要子类去实现

**获取资源**

*  尝试去非阻塞的获取同步状态，获取不到同步状态后封装为节点链接入同步队列自选等待获取锁

public final void acquire(int arg) {
        
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    	// 链接入同步队列
        private Node addWaiter(Node mode) {
        
            Node node = new Node(Thread.currentThread(), mode);
            // 尾节点CAS插入
            Node pred = tail;
            if (pred != null) {
        
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
        
                    pred.next = node;
                    return node;
                }
            }
            // 如果等待队列为空或者上述CAS失败，再自旋CAS插入
            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;
                    }
                }
            }
        }

**为什么需要CAS插入同步队列尾部？**

*  可能在同一时间存在多个线程竞争同步状态，就可能出现多个线程同时往同步队列插入尾节点的情况

> 回到获取资源的方法

*  封装好进入同步队列后，就是在同步队列中等待获取同步状态了，获取同步状态的过程也是一个自旋的过程，若是公平锁则是从头结点一个一个去获取同步状态，非公平锁则是随机

final boolean acquireQueued(final Node node, int arg) {
        
        boolean failed = true;
        try {
        
            boolean interrupted = false;
            // 自旋
            for (;;) {
        
                // p：前驱节点
                final Node p = node.predecessor();
                // 如果node的前驱结点p是head，表示node是第二个结点，就可以尝试去获取资源了（公平锁的情况下）
                if (p == head && tryAcquire(arg)) {
        
                    // 拿到资源后，将head指向该结点。
                    // 所以head所指的结点，就是当前获取到资源的那个结点或null。
                    setHead(node); 
                    p.next = null; // help GC
                    failed = false;
                    // 成功获取到锁，退出自旋
                    return interrupted;
                }
                // 获取不到锁则调用park()是自己阻塞
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
        
            if (failed)
                // 获取锁失败就将节点状态更新为cancle
                cancelAcquire(node);
        }
    }

这里parkAndCheckInterrupt方法内部使用到了LockSupport.park(this)，顺便简单介绍一下park。

LockSupport类是Java 6 引入的一个类，提供了基本的线程同步原语。LockSupport实际上是调用了Unsafe类里的函数，归结到Unsafe里，只有两个函数：

*  park(boolean isAbsolute, long time)：阻塞当前线程
 *  unpark(Thread jthread)：使给定的线程停止阻塞

所以**结点进入等待队列后，是调用park使它进入阻塞状态的。只有头结点的线程是处于活跃状态的**。

所以AQS为我们实现了主要逻辑，我们只需要子类重写尝试获取资源的protectd方法即可

**释放资源：**

public final boolean release(int arg) {
        
            if (tryRelease(arg)) {
        
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    		
        private void unparkSuccessor(Node node) {
        
            // 如果状态是负数，尝试把它设置为0
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
            // 得到头结点的后继结点head.next
            Node s = node.next;
            // 如果这个后继结点为空或者状态大于0
            // 通过前面的定义我们知道，大于0只有一种可能，就是这个结点已被取消
            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);
        }

## 四，AQS实现类 ##

synchronized的不足：

*  如果synchronized保护的临界区是只读操作，其实是可以多线程并发读的，但是synchronized是保证了只能一个线程去读
 *  synchronized无法知道是否成功获取到锁
 *  如果获取到锁的线程在synchronized保护的临界区内调用sleep()且不释放锁的话，那么会导致其他线程也一同等待

公平锁和非公平锁：

这里的“公平”，其实通俗意义来说就是“先来后到”，也就是FIFO。如果对一个锁来说，先对锁获取请求的线程一定会先被满足，后对锁获取请求的线程后被满足，那这个锁就是公平的。反之，那就是不公平的。

**一般情况下，非公平锁能提升一定的效率。但是非公平锁可能会发生线程饥饿（有一些线程长时间得不到锁）的情况。所以要根据实际的需求来选择非公平锁和公平锁。**

读写锁：

*  写锁：同一时刻只有一个线程能写资源，其他写线程和读线程都会被阻塞
 *  读锁：同一时刻允许多个读线程读资源，大大提高性能，适合读多写少的场景

### 顶层锁接口 ###

**lock**

lock接口定义了获取锁和释放锁的方法

public interface Lock {
        
        // 正常的lock，获取不到锁则阻塞等待
        void lock();
        // 可中断的获取锁：线程B获取不到锁在阻塞等待，但是现在线程B不想再等了，于是可以自己中断自己或者让别人中断自己，从而不再阻塞等待，去干别的事
        void lockInterruptibly() throws InterruptedException;
        // 获取不到锁不会阻塞等待，直接返回
        boolean tryLock();
        // 在指定时间内获取不到锁则返回
        boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
        // 释放锁
        void unlock();
        // 等待队列
        Condition newCondition();
    }

**ReadWriteLock**

ReadWriteLock接口定义了获取读锁和写锁的方法

public interface ReadWriteLock {
        
        Lock readLock();
        Lock writeLock();
    }

**lock和synchronized的区别：**

*  lock是API级别的锁，基于AQS实现同步状态的获取和释放，synchronized是JVM级别的锁,JVM字节码指令实现同步的获取和释放
 *  lock在获取不到锁时可以直接返回，不阻塞等待，synchronized获取不到必须阻塞等待，且无法知道是否成功获取到锁
 *  lock可以响应中断，synchronized不可以

### ReentrantLock ###

reentranlock实现了非公平锁和公平锁，默认是非公平锁

public ReentrantLock() {
        
            sync = new NonfairSync();
        }

也可以传入true实现公平锁

public ReentrantLock(boolean fair) {
        
            sync = fair ? new FairSync() : new NonfairSync();
        }

公平锁和非公平锁其实内部依赖是公平同步器和非公平同步器，上文我们提到过，AQS为我们实现了主要的逻辑，JUC包中的工具类基于AQS实现，都实现AQS中的protectd方法

reentranlock中公平锁和非公平锁的区别在加锁，

// 非公平同步器
       static final class NonfairSync extends Sync {
        
            private static final long serialVersionUID = 7316153563782823691L;
            final void lock() {
        
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
            }
    
            protected final boolean tryAcquire(int acquires) {
        
                return nonfairTryAcquire(acquires);
            }
        }
    
        // 公平同步器
        static final class FairSync extends Sync {
        
            private static final long serialVersionUID = -3000897897090466540L;
    
            final void lock() {
        
                acquire(1);
            }
    
            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;
            }
        }

非公平的获取锁：

protected final boolean tryAcquire(int acquires) {
        
                return nonfairTryAcquire(acquires);
            }
    
            final boolean nonfairTryAcquire(int acquires) {
        
                // 当前线程
                final Thread current = Thread.currentThread();
                // 当前同步状态
                int c = getState();
                if (c == 0) {
        
                    // CAS修改同步状态，将获取同步状态的线程修改为当前线程
                    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;
            }

公平的获取锁：

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

**公平锁获取同步状态的一个重要条件：**

*  当前线程对应的节点没有前驱节点

**释放锁：**

protected final boolean tryRelease(int releases) {
        
                int c = getState() - releases;
                // 先判断是否是持有锁的线程
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
        
                    // 对于锁重入的情况，只有当同步状态为0时才真正返回true
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                // 修改同步状态
                setState(c);
                return free;
            }

### ReentrantReadWriteLock ###

ReentrantReadWriteLock的锁还是通过state变量作为获取锁的标准，高16位表示读锁，低16位表示写锁

**读锁加锁**

final boolean tryReadLock() {
        
                Thread current = Thread.currentThread();
                for (;;) {
        
                    int c = getState();
                    // 如果有其他写线程，则直接获取失败，退出自旋
                    if (exclusiveCount(c) != 0 &&
                        getExclusiveOwnerThread() != current)
                        return false;
                    // 获取读状态     
                    int r = sharedCount(c);
                    // 最大读状态，不能再读，资源上限了
                    if (r == MAX_COUNT)
                        throw new Error("Maximum lock count exceeded");
                    // CAS更新读状态
                    if (compareAndSetState(c, c + SHARED_UNIT)) {
        
                        // r = 0：第一个获取了读锁
                        if (r == 0) {
        
                            firstReader = current;
                            firstReaderHoldCount = 1;
                        } else if (firstReader == current) {
        
                            // 锁重入
                            firstReaderHoldCount++;
                        } else {
        
                            //cachedHoldCounter 为缓存最后一个获取锁的线程
                            HoldCounter rh = cachedHoldCounter;
                            if (rh == null || rh.tid != getThreadId(current))
                                cachedHoldCounter = rh = readHolds.get(); //缓存最后一个获取锁的线程
                            else if (rh.count == 0)// 当前线程获取到了锁，但是重入次数为0，那么把当前线程存入进去
                                readHolds.set(rh);
                            rh.count++;
                        }
                        return true;
                    }
                }
            }

**写锁获取**

public boolean tryLock( ) {
        
                return sync.tryWriteLock();
            }
            
            
     final boolean tryWriteLock() {
        
                Thread current = Thread.currentThread();
                int c = getState();
                if (c != 0) {
        //状态不等于0，说明已经锁已经被获取过了
                    int w = exclusiveCount(c);//这里是判断是否获取到了写锁，后面会详细分析这段代码
                    // 这里就是判断是否是锁重入：2种情况
                    // 1.c!=0说明是有锁被获取的，那么w==0，
                    // 说明写锁是没有被获取，也就是说读锁被获取了，由于写锁和读锁的互斥，为了保证数据的可见性
                    // 所以return false.
                    //2. w!=0，写锁被获取了，但是current != getExclusiveOwnerThread() ，
                    // 说明是被别的线程获取了，return false;
                    if (w == 0 || current != getExclusiveOwnerThread())
                        return false;
                    if (w == MAX_COUNT)//判断是否溢出
                        throw new Error("Maximum lock count exceeded");
                }
                // 尝试获取锁
                if (!compareAndSetState(c, c + 1))
                    return false;
                setExclusiveOwnerThread(current);
                return true;
            }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTkyMjI4OQ_size_16_color_FFFFFF_t_70]: /images/20230209/bd93d0ce466c44af8c7488587defa75b.png