AQS学习小随笔

川长思鸟来 2022-11-25 11:50 181阅读 0赞

### AQS（AbstractQueuedSynchronizer） ###

更像是一个管理线程的工具类，线程的阻塞和接触阻塞，等待队列的管理，高效解决这些繁琐的问题，工具类只需要去关注业务逻辑实现；

> AQS是一个用于构建锁、同步器、协作工具类的工具类（框架），有了AQS之后方便了很多构建线程协作的并发工具功能实现。

//AQS是抽象类，继承AbstractOwnableSynchronizer为了查看当前获取获取锁的线程，方便监控
    //很多工具类对AQS进行了继承
    public abstract class AbstractQueuedSynchronizer
        extends AbstractOwnableSynchronizer
        implements java.io.Serializable {

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

> 使用到AQS的并发工具类主要有：ReentrantLock，ReentrantReadWirteLock，CountDownLatch，Semaphore，ThreadPoolExector等；

#### 1. state状态 ####

state是一个**int**类型成员变量来表示**同步状态**，并且被**volatile**修饰保证可见性，所以所有修改state的方法，例如getState，setState以及CAS操作都会保证并发安全。

/**
     * The synchronization state.
     */
    private volatile int state;

**state的具体含义会根据具体的实现来确定**：

*  在ReentrantLock中，state代表可以锁可重入次数；
 *  在CountDownLatch中，state代表还需要倒数的次数；
 *  在Semaphore中，代表剩余的许可证数量；

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

#### 2.控制线程抢锁和配合的FIFO队列 ####

队列最主要是作用就是用来存放\*\*“等待的线程“\*\*，AQS就相当于是排队管理器，当多个线程争取一把锁的时候，必须使用排队机制将没有拿到锁的线程进行排队；当锁释放的时候，就会选择一个合适的线程来获取这个刚释放的锁。

AQS会维护一个等待的线程队列，将**等待的线程都存放到这个队列里；**

队列是一个**双向链表**形式；

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

#### 3. 期望协作工具类去实现的获取/释放等方法 ####

这里**获取/释放**的方法，是利用AQS的协作工具类里最重要的方法，是由**协作类自己去实现的**，并且含义各不相同；

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 2]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 4]

#### AQS用法 ####

1.  写一个类，想好协作的逻辑，实现获取/释放方法；
2.  内部写一个Sync类继承AbstractQueuedSynchronizer；
3.  根据释放独占来重写方法：如果是独占需要重写tryAcquire/tryRelease；如果是共享需要重写tryAcquireShared/tryReleaseShared；然后进行方法调用；

#### AQS在CountDownLatch的应用 ####

CountDownLatch中有内部类Sync继承了AQS；

##### 构造器 #####

//创建一个Sync引用
    private final Sync sync;
    //传入count
    public CountDownLatch(int count) {
        
        if (count < 0) throw new IllegalArgumentException("count < 0");
      //初始化Sync，传入count
        this.sync = new Sync(count);
    }
    //传入的count实则调用AQS的setState方法
    Sync(int count) {
        
        setState(count);
     }
    //count即AQS的state
    //这个state即CountDownLatch中的倒数次数
    protected final void setState(int newState) {
        
            state = newState;
    }

##### getCount #####

int getCount() {
        
        return getState();
    }
    //获取count即返回AQS的state
     protected final int getState() {
        
            return state;
        }

##### await #####

//让当前线程等待倒数结束再执行
    public void await() throws InterruptedException {
        
        sync.acquireSharedInterruptibly(1);
    }
    
    
    public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
        
            if (Thread.interrupted())
                throw new InterruptedException();
      			//跳17
            if (tryAcquireShared(arg) < 0)
              //让当前线程进入等待队列并阻塞
              //跳24
                doAcquireSharedInterruptibly(arg);
        }
    
    //该方法是CountDownLatch实现AQS的
    //返回AQS的state，判断state是否倒数至0
    protected int tryAcquireShared(int acquires) {
        
                return (getState() == 0) ? 1 : -1;
    }
    
     //让当前线程进入等待队列并阻塞
    private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
        
      //包装节点（AQS等待队列的节点）
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
        
                for (;;) {
        
                    final Node p = node.predecessor();
                    if (p == head) {
        
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
        
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
    //parkAndCheckInterrupt调用LockSupport.park，底层是Unsafe.park方法，阻塞当前线程
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
        
                if (failed)
                    cancelAcquire(node);
            }
        }

##### countDown #####

public void countDown() {
        
      //直接调用AQS的方法判断是否释放  
      sync.releaseShared(1);
    }
    
    //AQS的方法
    public final boolean releaseShared(int arg) {
        
      			//判断是否需要释放线程
      			//跳19
            if (tryReleaseShared(arg)) {
        
              //唤醒队列中的等待线程开始执行
                doReleaseShared();
                return true;
            }
            return false;
        }
    
    
     protected boolean tryReleaseShared(int releases) {
        
       //for循环做CAS自旋，对state-1
                for (;;) {
        
                  //拿到count
                    int c = getState();
             //如果count=0，就不需要释放，已经被释放过了
                    if (c == 0)
                        return false;
                  //存储count-1之后的结果
                    int nextc = c-1;
                  //CAS进行对count进行原子更新
                    if (compareAndSetState(c, nextc))
            //如果更新之后count=0，本次countDown就进行释放
                        return nextc == 0;
                }
            }

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

#### AQS在Semaphore的应用 ####

AQS在Semaphore中，state代表许可证的剩余数量；

##### acquire（公平和非公平情况） #####

public void acquire() throws InterruptedException {
        
        sync.acquireSharedInterruptibly(1);
    }
    
     public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
        
            if (Thread.interrupted())
                throw new InterruptedException();
       //tryAcquireShared分非公平和公平
       //非公平跳20
       //公平跳47
       //判读能否获取到许可证
            if (tryAcquireShared(arg) < 0)
               //让当前线程进入等待队列并阻塞
              //具体见CountDownLatch第24行
                doAcquireSharedInterruptibly(arg);
        }
    
    
    //非公平情况
    protected int tryAcquireShared(int acquires) {
        
       //跳27
      return nonfairTryAcquireShared(acquires);
            }
    
    
     final int nonfairTryAcquireShared(int acquires) {
        
       //for循环CAS自旋        
       for (;;) {
        
         //available存储剩余许可证即AQS的state
       int available = getState();
         //remaining存储state-1后的剩余许可证
     int remaining = available - acquires;
         //判断remaining是否<0：
         //如果remaining < 0，由于短路也不会执行CAS，返回remaining为负数
         //跳13符合条件，当前线程入队等待
         
         //如果remaining > 0，执行CAS，如果设置成功，就说明成功获取许可证，就会返回正数remaining
         //如果CAS失败就会继续自旋尝试CAS修改
         //跳13不符合条件，放行当前线程
    if (remaining < 0 || compareAndSetState(available, remaining))
      //返回剩余的许可证数量
                      return remaining;
                }
            }
    
    
    //公平情况
    protected int tryAcquireShared(int acquires) {
        
      //for循环CAS自旋
                for (;;) {
        
    //判断等待队列是否为null并且队列的第一个节点（节点存储线程）是否是当前线程
     							//如果当前线程之前还有线程等待，那么就返回-1
    							//跳13符合条件，当前线程入队等待
                    if (hasQueuedPredecessors())
                        return -1;
                  //与非公平逻辑相同
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

##### release #####

//释放许可证
    public void release() {
        
      //调用AQS方法判断释放许可证
        sync.releaseShared(1);
    }
    
     public final boolean releaseShared(int arg) {
        
       //tryReleaseShared方法跳18
            if (tryReleaseShared(arg)) {
        
                //唤醒队列中的等待线程开始执行
                doReleaseShared();
                return true;
            }
            return false;
        }
    
    
    protected final boolean tryReleaseShared(int releases) {
        
      //for循环CAS自旋
                for (;;) {
        
                  //获取当前剩余许可证即AQS的state
                    int current = getState();
                  //next暂存释放后的许可证数量
                    int next = current + releases;
                  //判断许可证数量释放溢出
                    if (next < current) // overflow
                        throw new Error("Maximum permit count exceeded");
                  //能否CAS成功，如果成功就返回true唤醒其他线程；CAS失败就继续自旋CAS修改
                    if (compareAndSetState(current, next))
                        return true;
                }
            }

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

#### AQS在ReentrantLock的应用 ####

state表示是锁的占有情况，包括可重入次数；

初始是0，表示没有线程占有；如果是1，表示有线程持有锁；如果锁被多次获取就可以对state进行累加，释放锁也进行减操作；当state值为0的时候，标识Lock不被任何锁占有。

##### unlock #####

public void unlock() {
        
        sync.release(1);
    }
    
    
    public final boolean release(int arg) {
        
      //tryRelease跳18
      //如果返回true，说明该锁已经释放，唤醒等待队列中的第一个线程
      //返回false，不做操作，队列线程继续等待
            if (tryRelease(arg)) {
        
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    
    
    protected final boolean tryRelease(int releases) {
        
      //定义c暂存state-1
                int c = getState() - releases;
      //当前线程不持有锁，解锁抛出异常
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
      //默认返回false
                boolean free = false;
      //只有当前重入次数c=0，说明当前没有线程获取该锁
                if (c == 0) {
        
                  //返回true
                    free = true;
                  //设置当前持有这把锁的线程为null
                    setExclusiveOwnerThread(null);
                }
      //存储新的重入次数
                setState(c);
                return free;
            }

##### lock（公平和非公平情况） #####

NonfairSync、FairSync都继承了Sync，Sync继承AQS；

public void lock() {
        
        sync.lock();
    }
    
    //非公平情况下获得锁
     final void lock() {
        
       //进行CAS，查看state是不是0：
       //如果是0就说明该锁没有人获取，将state修改为1，即拿到锁；
       //如果不是0，说明该锁已经被占有，无法获取锁
       if (compareAndSetState(0, 1))
    //如果CAS成功，当前线程设置为持有锁的线程                
         setExclusiveOwnerThread(Thread.currentThread());
                else
     //CAS失败跳18
                    acquire(1);
            }
    
     public final void acquire(int arg) {
        
       //tryAcquire跳27（非公平），67（公平）
       //tryAcquire(arg)返回false说明获取锁失败，然后执行&&后面
            if (!tryAcquire(arg) &&
                 //当前线程进入等待队列
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
              //出现异常进行中断
                selfInterrupt();
    }
    //非公平情况
      protected final boolean tryAcquire(int acquires) {
        
        //nonfairTryAcquire跳32
                return nonfairTryAcquire(acquires);
     }
    
    final boolean nonfairTryAcquire(int acquires) {
        
      //获取当前线程
                final Thread current = Thread.currentThread();
      //获取state也就锁的重入次数
                int c = getState();
      //如果c=0也就是当前锁没有线程占有
                if (c == 0) {
        
                  //CAS获取锁
                    if (compareAndSetState(0, acquires)) {
        
                      //将当前线程设置为持有锁
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
      //如果c！=0，并且当前线程是锁的持有者，说明现在是重入锁操作
                else if (current == getExclusiveOwnerThread()) {
        
                  //暂存重入次数c+1
                    int nextc = c + acquires;
                  //判断是否溢出
                    if (nextc < 0) // overflow
                        throw new Error("Maximum lock count exceeded");
                  //设置新的重入次数
                    setState(nextc);
                    return true;
                }
      //c！=0并且当前线程也没占有锁，说明锁被其他线程占有，接下来进入等待队列等待锁
                return false;
            }
    
    //公平情况获得锁
    final void lock() {
        
      //acquire跳18
                acquire(1);
            }
    
    //公平情况
     protected final boolean tryAcquire(int acquires) {
        
       	//获取当前线程
                final Thread current = Thread.currentThread();
       //获取state，锁的重入次数
                int c = getState();
       //如果c=0，说明当前锁没有被其他线程占有
                if (c == 0) {
        
                  //先判断当前线程之前的等待队列中是否有排队的线程，如果没有再进行CAS操作
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
        
                      //将当前线程设置为占有锁
                        setExclusiveOwnerThread(current);
                        return true;
                    }
                }
       //如果当前线程是该锁的占有者
                else if (current == getExclusiveOwnerThread()) {
        
                  //重入次数c+1
                    int nextc = c + acquires;
                  //重入次数溢出
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                  //设置新的重入次数
                    setState(nextc);
                    return true;
                }
         //c！=0并且当前线程也没占有锁，说明锁被其他线程占有，接下来进入等待队列等待锁
                return false;
            }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70]: /images/20221124/8ab8c92e5420468a95b20ca49bb00820.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 1]: /images/20221124/f5959cd87a094f0cb097e6f42f976673.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200808191537708.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 3]: /images/20221124/18ad8acc5ac94dceb0e78b19b109cc90.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 4]: /images/20221124/41b1728fbfc748caa5b1d651ceac927d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 5]: /images/20221124/df46fe47954c4ba9951cbdfc55bdc395.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDE5MDExMw_size_16_color_FFFFFF_t_70 6]: /images/20221124/c882f92681674774bd50a10d9e88dadf.png