Java并发工具之AQS原理

灰太狼 2023-07-15 11:26 35阅读 0赞

## 1. 为什么需要AQS？ ##

*  AQS:AbstractQueuedSynchronizer，抽象队列同步器；
 *  锁和协作类的共同点：闸门；
 *  ReentrantLock、Semaphore、CountDownLatch、ReentrantReadWriteLock都有类似的协作功能，因为都用了一个共同的基类：AQS；
 *  就ReentrantLock和Semaphore而言可以屏蔽很多细节，有许多共同的部分，将其抽象出来就是AQS；

## 2. Semaphore和AQS的关系？ ##

*  Semaphore内部有一个Sync类，Sync类继承了AQS；
 *  CountDownLatch也是这样；
 *  AQS是个工具，把它放到Semaphore、CountDownLatch、ReentrantLock的内部，就可以使用；

## 3. AQS的比喻 ##

*  Semaphore：限制资源的数量，一个人面试完了之后，后一个人才能拿到许可证进来面试；
 *  CountDownLatch：群面，等待人到齐；
 *  Semaphore、CountDownLatch这些顶层工具，要做的就是定制规则，剩下的交给AQS来做；

## 4. AQS的作用？ ##

*  同步状态的原子管理；
 *  线程的阻塞与解除阻塞；
 *  队列的管理；
 *  AQS是一个用于构建锁、同步器、协作工具类的工具类；
 *  有了AQS，构建线程写作类就容易多了；

## 5. AQS内部原理解析 ##

AQS由一下三大部分组成：

1.  state状态，在不同的类中有不同的含义；
     2. 控制线程抢锁和配合的FIFO队列，即存放等待的线程；
     3. 期望协作工具类去实现的获取/释放等重要方法；

### 5.1 state状态 ###

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

*  state的具体含义，会根据具体实现类的不同而不同，比如在Semaphore里，它表示“剩余许可证的数量”，在CountDownLatch里，表示“还需要倒数的数量”，在ReentrantLock中，表示锁的占用情况，0表示没有被占有，1表示被持有了，如果有2、3、4…等表示某线程重入了几次；
 *  state是volatile修饰的，会被并发地修改，所以所有的state的方法需要保证线程安全，比如getState、setState以及compareAndSetState（通过硬件级别的原子性保证了应用层操作的原子性）操作来读取和更新这个状态。这些方法都依赖于juc.atomic包的支持；

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

*  这个队列用来存放“等待的线程”，AQS就是“排队管理器”，当多个线程争用同一把锁时，必须有排队机制将那些没能拿到锁的线程串在一起。当锁释放时，锁管理器就会挑选一个合适的线程占用这个刚刚释放的锁；
 *  AQS会维护一个等待的线程队列，把线程都放在这个队列里；

### 5.3 协作工具类去实现的获取/释放等 ###

*  这里的获取和释放方法，是利用AQS的协作工具类里最重要的方法，是由协作类自己去实现的，并且含义各不相同；
 *  获取方法：依赖于State变量，经常会阻塞（比如获取不到锁的情况）；
 *  在Semaphore中，获取就是acquire方法，作用是获取一个许可证；
 *  在CountDownLatch里面，获取就是await方法，直到其他线程把state减成0之后，当前线程才能被唤醒；
 *  释放方法：释放操作不会阻塞；
 *  在Semaphore中，释放就是release方法，作用是释放一个许可证；
 *  CountDownLatch里面，获取就是countDown方法，作用是“倒数一个数”；

需要线程协作类自己去实现的方法：

*  Semaphore里面的静态内部类Sync中的getPermits、nonfairTryAcquireShared、tryReleaseShared等；
 *  CountDownLatch里面的静态内部类Sync中的getCount、tryAcquireShared、tryReleaseShared等；

## 6. AQS应用实例、源码解析 ##

### 6.1 AQS用法 ###

1.  写一个类，想好协作的逻辑，实现获取/释放方法；
2.  内部写一个Sync类继承AQS；
3.  根据是否独占来重写，如果是独占的就重写tryAcquire/tryRelease；如果是共享的就重写tryAcquireShared(int acquires)和tryReleaseShared(int releases)等方法，在步骤1写的获取/释放方法中调用AQS;

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

1.  首先创建CountDownLatch并传入倒数的数值5；

CountDownLatch countDownLatch = new CountDownLatch(5);

1.  在CountDownLatch的构造函数中创建Sync对象并传入数值

public CountDownLatch(int count) { 
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }

1.  在Sync的构造函数中调用setState(count)方法设置state

Sync(int count) { 
                setState(count);
            }

1.  在CountDownLatch里面有个getCount()方法

public long getCount() { 
            return sync.getCount();
        }

*  间接调用了sync.getCount方法

int getCount() { 
                return getState();
            }

*  sync的getCount方法调用了getState方法

protected final int getState() { 
            return state;
        }

AQS在CountDownLatch的总结：

*  调用CountDownLatch的await方法时，便会尝试获取共享锁，不过一开始是获取不到该锁的，于是线程被阻塞；
 *  而“共享锁”可获取到的条件，就是“锁计数器”归0；
 *  而“锁计数器”的初始值为count，每当一个线程调用该CountDownLatch对象的countDown方法时，才将“锁计数器”-1；
 *  count个线程调用了countDown之后，锁计数器才为0，而前面提高的等待获取共享锁的线程才能继续运行；

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

*  在Semaphore中，state就是“剩余许可证的数量”；
 *  看tryAcquire方法，判断nonfairTryAcquireShared大于等于0的话，就表示成功；

final int nonfairTryAcquireShared(int acquires) { 
                for (;;) { 
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

*  这里会先检查剩余许可证数量remaining 够不够，如果remaining 小于0，那么直接返回负数，表示失败；如果够了，就自旋加CAS改变state的值，直到改变成功就返回整数；如果期间被其他人修改了导致剩余数量不够了（因为多个线程竞争导致CAS操作也有可能失败，因为for循环自旋的缘故，再次循环到remaining和CAS，都不满足，就进入到下一个循环中，直到满足CAS操作或者remaining 小于0，表示线程获取到remaining ，线程继续执行），那也返回负数表示失败。

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

*  unLock方法， 调用了sync的release方法

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

release方法调用了AQS类中的release方法

public final boolean release(int arg) { 
            if (tryRelease(arg)) { 
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

tryRelease方法如下，会先判断当前这个线程是不是持有锁的线程，如果是的话，如果state - 1的结果c等于0就把free改为true并返回，同时把线程持有的锁设置为null，表示没有任何线程持有这把锁，如果不等于0，等于减少了重入的次数，把state的值修改为c

protected final boolean tryRelease(int releases) { 
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) { 
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }

*  lock方法， 调用了sync的lock方法，可以看出Sync并没有实现lock方法，是因为lock方法有公平和非公平实现

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

abstract void lock();

非公平的实现方法如下，CAS操作表示当前state为0的时候才能成功，表示没有线程持有这把锁的时候才能让当前线程持有，一旦CAS失败就会调用acquire方法，进入阻塞状态或者重入

final void lock() { 
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

*  分析释放锁的方法tryRelease：
    
     *  由于是可重入的，所以state代表重入的次数，每次释放锁先判断是不是当前持有锁的线程释放的，如果不是就抛异常，如果是的话，重入次数就减一，如果减到了0，就说明完全释放了，于是free就是true，并且把state设置为0
 *  分析加锁的方法：
    
     *  判断当前state是不是等于0，如果是则让当前线程持有锁，如果不是就判断是不是当前线程又来重入，如果不是当前线程来重入，就进入阻塞状态；

## 7. AQS实战一个简化版CountDownLatch ##

/** * 一次性门闩，倒数一次，其他线程就开始释放，arg为0表示关闭，为1表示打开 */
    package tools.aqs;
    
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    
    /** * 一次性门闩，倒数一次，其他线程就开始释放，arg为0表示关闭，为1表示打开 */
    public class OneShotLatch { 
    
        private final Sync sync = new Sync();
    
        public void countDown() { 
            sync.releaseShared(0);
        }
    
        public void await() { 
            sync.acquireShared(0);
        }
    
        private static class Sync extends AbstractQueuedSynchronizer { 
    
            @Override
            protected int tryAcquireShared(int arg) { 
                return getState() == 1 ? 1: -1;//getState等于0，表示门关闭，返回-1，执行await的线程进入阻塞状态，
                                            // 在阻塞状态下不停的自旋，直到tryAcquireShared >= 0，表示10个子线程获取到锁
            }
    
            @Override
            protected boolean tryReleaseShared(int arg) { 
                setState(1);//让10个自旋的线程tryAcquireShared>0，可以获取到锁继续运行
                return true;//释放主线程继续运行
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            OneShotLatch oneShotLatch = new OneShotLatch();
            for (int i = 0; i < 10; i++) { 
                new Thread(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(Thread.currentThread().getName() + "尝试获取Latch...");
                        oneShotLatch.await();
                        System.out.println(Thread.currentThread().getName() + "门闩打开，继续运行...");
                    }
                }).start();
            }
            Thread.sleep(1000);
            oneShotLatch.countDown();
    
        }
    }

Thread-0尝试获取Latch...
    Thread-1尝试获取Latch...
    Thread-5尝试获取Latch...
    Thread-6尝试获取Latch...
    Thread-7尝试获取Latch...
    Thread-3尝试获取Latch...
    Thread-2尝试获取Latch...
    Thread-8尝试获取Latch...
    Thread-4尝试获取Latch...
    Thread-9尝试获取Latch...
    Thread-0门闩打开，继续运行...
    Thread-1门闩打开，继续运行...
    Thread-5门闩打开，继续运行...
    Thread-6门闩打开，继续运行...
    Thread-7门闩打开，继续运行...
    Thread-3门闩打开，继续运行...
    Thread-2门闩打开，继续运行...
    Thread-8门闩打开，继续运行...
    Thread-9门闩打开，继续运行...
    Thread-4门闩打开，继续运行...