11.AQS及相关实现类

Bertha 。 2023-07-04 15:57 9阅读 0赞

**AQS**

AQS是一款抽象的队列式同步器，他定义了一套多线程访问共享资源的同步框架。

当请求得资源空闲那么他将当前请求资源设置为有效的工作线程，并将共享变量设置为锁定状态，如果请求的资源锁定，那么就进入CLH阻塞队列中。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70][]

AQS定义两种资源共享方式：Exclusve（独占，只有一个线程执行，如Reentrantlock）和Share(共享，多个线程同时执行，Semaphore/CountDownLatch)

1.  调用自定义同步器的tryAcquire()尝试直接去获取资源，如果成功则直接返回；
2.  没成功，则addWaiter()将该线程加入等待队列的尾部，并标记为独占模式；
3.  acquireQueued()使线程在等待队列中休息，有机会时（轮到自己，会被unpark()）会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过，则返回true，否则返回false。
4.  如果线程在等待过程中被中断过，它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()，将中断补上。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70 1][]

这也就是ReetrantLock.lock流程。

**ReentrantLock**

这是一种可重入锁，支持公平锁和非公平锁，在获得所得同步块的时候，不需要再一次去获得锁。

公平锁：每一次的tryAcquire都会检查CLH队列中是否仍有前驱的元素，如果有的话那么会继续等待。这种锁保证先来先服务的的原则。

非公平锁：每一次都会检查并设置锁的状态，这种情况就是队列中即使有等待的线程，那么这个节点也会跟头节点进行争抢，这样就不保证先来先服务，但是对等待的线程保证先来先服务

公平锁源代码分析：

**lock方法：**

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

这事实上式调用AQS的方法。

尝试去争取锁：

protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) {
    // c=0 说明没有其他线程占有锁
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
    // 队列中没有其他线程在等待锁，而且CAS把state设置成入参的值成功，这里是1（这里的CAS就是我
    // 们前文提的并发竞争机制），则当前线程获取锁成功并将owner线程设置为当前线程
                        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;
            }

如果争取到了就直接返回，如果失败了就进入队列:

private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
    
    // Try the fast path of enq; backup to full enq on failure
    // 上面这个官方注释很直白，其实下面的enq方法里也执行了这段代码，但是这里先直接试一下看能
    //  否插入成功
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
    // CAS把tail设置成当前节点，如果成功的话就说明插入成功，直接返回node，失败说明有其他线程也
    // 在尝试插入而且其他线程成功,如果是这样就继续执行enq方法
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
        }

enq方法：

private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                if (t == null) { // Must initialize
    // 最开始head和tail都是空的，需要通过CAS做初始化，如果CAS失败，则循环重新检查tail
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
    // head和tail不是空的，说明已经完成初始化，和addWaiter方法的上半段一样，CAS修改
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }

如果在请求失败以后还执行了acquireQueue方法，因为我们执行插入队列之后还没有阻塞当前线程呢。

final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
    /*
    * 如果前置节点是head，说明当前节点是队列第一个等待的节点，这时去尝试获取锁，如果成功了则
    * 获取锁成功。这里有的同学可能没看懂，不是刚尝试失败并插入队列了吗，咋又尝试获取锁？ 其实这*
    * 里是个循环，其他刚被唤醒的线程也会执行到这个代码
    */
                    if (p == head && tryAcquire(arg)) {
    // 队首且获取锁成功，把当前节点设置成head，下一个节点成了等待队列的队首
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
    /* shouldParkAfterFailedAcquire方法判断如果获取锁失败是否需要阻塞，如果需要的话就执行
    *  parkAndCheckInterrupt方法，如果不需要就继续循环
    */
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取pred前置节点的等待状态
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
    /* 前置节点状态是signal，那当前节点可以安全阻塞，因为前置节点承诺执行完之后会通知唤醒当前
    * 节点
    */
                return true;
            if (ws > 0) {
    
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
    // 前置节点如果已经被取消了，则一直往前遍历直到前置节点不是取消状态，与此同时会修改链表关系
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                pred.next = node;
            } else {
                /*
                 * waitStatus must be 0 or PROPAGATE.  Indicate that we
                 * need a signal, but don't park yet.  Caller will need to
                 * retry to make sure it cannot acquire before parking.
                 */
    // 前置节点是0或者propagate状态，这里通过CAS把前置节点状态改成signal
    // 这里不返回true让当前节点阻塞，而是返回false，目的是让调用者再check一下当前线程是否能
    // 成功获取锁，失败的话再阻塞，这里说实话我也不是特别理解这么做的原因
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }

加入前面一步返回true需要阻塞，则调用下面方法：

private final boolean parkAndCheckInterrupt() {
    // 阻塞当前线程，监事是当前sync对象
            LockSupport.park(this);
    // 阻塞返回后，返回当前线程是否被中断
            return Thread.interrupted();
        }

到这里一次lock的调用就结束了。

*  调用tryAcquire方法尝试获取锁，获取成功的话修改state并直接返回true，获取失败的话把当前线程加到等待队列中。
 *  加到等待队列中之后先检查前置节点是不是signal,如果是的话直接阻塞当前线程并等待唤醒，如果不是的话判断是不是cancel状态，是cancel状态就往前遍历并把cancel状态的节点从队列中删除，如果状态是0就改成signal。
 *  阻塞被唤醒之后如果是队首并且尝试获取锁成功就返回true，否则就继续执行上一步代码进入阻塞。

unlock方法：

unlock方式就是AQS中的release方法。

public final boolean release(int arg) {
    /*
     尝试释放锁如果失败，直接返回失败，如果成功并且head的状态不等于0就唤醒后面等待的节点
    */
            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的状态值
                int c = getState() - releases;
    // 如果持有锁的线程不是当前线程，直接抛出异常
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
                if (c == 0) {
    // 如果c==0，说明所有持有锁都释放完了，其他线程可以请求获取锁
                    free = true;
                    setExclusiveOwnerThread(null);
                }
    // 这里只会有一个线程执行到这，不存在竞争，因此不需要CAS
                setState(c);
                return free;
            }
    
    private void unparkSuccessor(Node node) {
            /*
             * If status is negative (i.e., possibly needing signal) try
             * to clear in anticipation of signalling.  It is OK if this
             * fails or if status is changed by waiting thread.
             */
            int ws = node.waitStatus;
            if (ws < 0)
    /*
    如果状态小于0，把状态改成0，0是空的状态，因为node这个节点的线程释放了锁后续不需要做任何
    操作，不需要这个标志位，即便CAS修改失败了也没关系，其实这里如果只是对于锁来说根本不需要CAS，因为这个方法只会被释放锁的线程访问，只不过unparkSuccessor这个方法是AQS里的方法就必须考虑到多个线程同时访问的情况（可能共享锁或者信号量这种）
    */
                compareAndSetWaitStatus(node, ws, 0);
    
            /*
             * Thread to unpark is held in successor, which is normally
             * just the next node.  But if cancelled or apparently null,
             * traverse backwards from tail to find the actual
             * non-cancelled successor.
             */
            Node s = node.next;
    // 这段代码的作用是如果下一个节点为空或者下一个节点的状态>0（目前大于0就是取消状态）
    // 则从tail节点开始遍历找到离当前节点最近的且waitStatus<=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);
        }

解锁的流程：

*  修改状态位
 *  唤醒排队的节点
 *  结合lock()方法，被唤醒的节点会自动替换为当前节点为head

参考文章：

[https://blog.csdn.net/m47838704/article/details/80013056][https_blog.csdn.net_m47838704_article_details_80013056]

[http://ifeve.com/juc-aqs-reentrantlock/][http_ifeve.com_juc-aqs-reentrantlock]

**CountDownLatch**

是通过一个计数器来实现，计数器初始的值是线程的数量，每当一个线程执行完毕，计数器的值就减一，当计数器的值为0时，就代表所有线程执行完，然后在闭锁等待得线程就可以恢复工作了。

**源码分析**：

构造函数

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    // Sync 就是继承了一个AQS
    this.sync = new Sync(count);
    }

看一下内部得Sync

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
    
        Sync(int count) {
            //设置了AQS得state
            setState(count);
        }
    
        int getCount() {
            return getState();
        }
    
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
    
        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

计算器减一得方法：

public void countDown() {
        sync.releaseShared(1);
    }

通过sync类得代码可以看出来，countDown释放1个，然后sync中通过cas修改state的值。

await()方法：

//让线程等待知道state=0。
    //如果state大于0则禁用当前线程，并处于休眠状态，知道state=0或者其他线程打断。
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    //判断线程的个数
    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

**CyclicBarrier**

他的作用是让所有线程都等待完成以后才会进行下一步操作。

举个例子就是说当朋友聚餐，等待所有朋友到期以后才上菜。

源码分析：

/** 这是一把可重入锁 */
    private final ReentrantLock lock = new ReentrantLock();
    /** 一个信号量通知，类似于wait/notify那个机制 */
    private final Condition trip = lock.newCondition();
    /** 参入的线程个数 */
    private final int parties;
    /* 当线程执行完以后要执行的线程 */
    private final Runnable barrierCommand;
    /** 当前代 */
    private Generation generation = new Generation();
    //计数器
    private int count;
    //静态内部类
    private static class Generation {
        boolean broken = false;
    }
    
    //构造方法
    public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }
    public CyclicBarrier(int parties) {
        this(parties, null);
    }
    
    //await方法
    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
    public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }
    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;
            //检查栅栏是否被打翻
            if (g.broken)
                throw new BrokenBarrierException();
            //检查线程有没有被中断    
            if (Thread.interrupted()) {
                //打分栅栏，唤醒所有线程，抛出终端异常。
                breakBarrier();
                throw new InterruptedException();
            }
            //每次都将计数器的值减一
            int index = --count;
            //计算器的值位0时，则唤醒所有的线程到下一代
            if (index == 0) {  // tripped
                boolean ranAction = false;
                try {
                    //唤醒所有线程先执行当前任务
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    //跳转到下一代
                    nextGeneration();
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }
    
            // 如果不为0则执行该循环
            for (;;) {
                try {
                    //判断是否是定时等待
                    if (!timed)
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    //若当前线程被打断则打翻栅栏，唤醒其他线程
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // 若在捕获中断异常前已经完成栅栏上的操作，直接中断。
                        Thread.currentThread().interrupt();
                    }
                }
    
                if (g.broken)
                    throw new BrokenBarrierException();
    
                if (g != generation)
                    return index;
    
                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

上面贴出的代码中注释都比较详细，我们只挑一些重要的来讲。可以看到在dowait方法中每次都将count减1，减完后立马进行判断看看是否等于0，如果等于0的话就会先去执行之前指定好的任务，执行完之后再调用nextGeneration方法将栅栏转到下一代，在该方法中会将所有线程唤醒，将计数器的值重新设为parties，最后会重新设置栅栏代次，在执行完nextGeneration方法之后就意味着游戏进入下一局。如果计数器此时还不等于0的话就进入for循环，根据参数来决定是调用trip.awaitNanos(nanos)还是trip.await()方法，这两方法对应着定时和非定时等待。如果在等待过程中当前线程被中断就会执行breakBarrier方法，该方法叫做打破栅栏，意味着游戏在中途被掐断，设置generation的broken状态为true并唤醒所有线程。同时这也说明在等待过程中有一个线程被中断整盘游戏就结束，所有之前被阻塞的线程都会被唤醒。线程醒来后会执行下面三个判断，看看是否因为调用breakBarrier方法而被唤醒，如果是则抛出异常；看看是否是正常的换代操作而被唤醒，如果是则返回计数器的值；看看是否因为超时而被唤醒，如果是的话就调用breakBarrier打破栅栏并抛出异常。这里还需要注意的是，如果其中有一个线程因为等待超时而退出，那么整盘游戏也会结束，其他线程都会被唤醒。下面贴出nextGeneration方法和breakBarrier方法的具体代码。

private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }
    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

CyclicBarrier与CountDownLatch区别：

CyclicBarrier由的计数器由自己控制，并且可以循环执行。

CountDownLatch的计数器有其他线程控制，单次循环。

Semaphore

semaphore是synchronized的加强版，作用是控制线程的并发数量。

是一个计数信号量，限制访问个数，可以用作限流。

他底层也是通过AQS实现的。

看一下他重要方法的源码：

//获取许可证
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        //如果线程被打断        
        if (Thread.interrupted())
            throw new InterruptedException();
        //如果不足以满足申请个数    
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    //判断申请获取个数能否满足
    protected int tryAcquireShared(int acquires) {
        for (;;) {
            if (hasQueuedPredecessors())
                return -1;
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

**ReentrantReadWriteLock**

读写锁是一种特殊的自旋锁，他把对共享资源的访问者划分成了读者和写者，读者对共享资源进行访问，写者则是对共享资源进行写操作，有点像数据库中的读写锁，读锁可以存在多个，写锁稚只能存在一个。他继承了ReetrantLock.主要通过sync实现

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70 2][]

读锁用前16位，表示持有读锁的线程数。写锁用后16位，表示写锁的重入次数。

源码分析（主要看一下读写锁得实现）：

//读锁
    private final ReentrantReadWriteLock.ReadLock readerLock;
    //写锁
    private final ReentrantReadWriteLock.WriteLock writerLock;
    
    //构造方法
    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
    //默认使用非公平锁
    public ReentrantReadWriteLock() {
        this(false);
    }
    public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
    public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }
    //读锁
    protected ReadLock(ReentrantReadWriteLock lock) {
        sync = lock.sync;
    }
    public void lock() {
        sync.acquireShared(1);
    }
    protected final int tryAcquireShared(int unused) {
        Thread current = Thread.currentThread();
        //获得当前资源状态
        int c = getState();
        //获取写锁得个数，判断有无写锁
        if (exclusiveCount(c) != 0 &&
        //有写锁的话判断是不是当前线程持有
            getExclusiveOwnerThread() != current)
            //加写锁失败
            return -1;
        //获取读写个数    
        int r = sharedCount(c);
        //判断是否应该阻塞，如果是公平锁则阻塞，非公平锁和头节点竞争
        if (!readerShouldBlock() &&
            //判断所得个数是否小于最大个数
            r < MAX_COUNT &&
            //cas修改获得锁的个数
            compareAndSetState(c, c + SHARED_UNIT)) {
            //判断当前锁得个数是否为0
            if (r == 0) {
                //设置锁得持有线程为当前线程
                firstReader = current;
                //记录所得个数
                firstReaderHoldCount = 1;
              //如果当前个数不为0，并且头节点为  
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                //读锁重入计数器更新
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
            }
            return 1;
        }
        return fullTryAcquireShared(current);
    }

参考资源：

[https://blog.csdn.net/fxkcsdn/article/details/82217760][https_blog.csdn.net_fxkcsdn_article_details_82217760]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200213000350271.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200213000909591.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE=,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_m47838704_article_details_80013056]: https://blog.csdn.net/m47838704/article/details/80013056
[http_ifeve.com_juc-aqs-reentrantlock]: http://ifeve.com/juc-aqs-reentrantlock/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200213001638652.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1YWl4aWFvaGFpXzE=,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_fxkcsdn_article_details_82217760]: https://blog.csdn.net/fxkcsdn/article/details/82217760