AbstractQueuedSynchronizer的简单介绍

Posted 2021-02-08 qlqwjy

　　AbstractQueuedSynchronizer简称为AQS。大多数开发者不会直接使用AQS，标准同步器类的集合能够满足绝大多数情况的需求。

1.AbstractQueuedSynchronizer简介

　　在基于AQS构建的同步容器类中，最基本的操作包括各种形式的获取和释放操作。获取操作是一种依赖状态的操作，并且通常会阻塞。当使用锁或信号量时，“获取”操作的含义就很直观，即获取的是锁或者许可，并且调用者可能会一直等待同步器类处于可被获取的状态。在使用CountDownLatch时，获取操作意味着"等待并直到闭锁到达结束状态"，而在使用FutureTask时，则意味着等待并直到任务已经完成。"释放"并不是一个可阻塞的操作，当执行"释放"操作时，所有在请求时被阻塞的线程都会开始执行。

　　AQS负责同步管理器类中的状态，它管理了一个整数状态信息，可以通过getState、setState以及compareAndSetState三个protected类型方法进行操作。这个整数可以表示任意状态。例如:ReentrantLock用它来表示所有者线程已经重复获取该锁的次数(重入的层数)，Semaphore用于表示剩余的许可数量，FutureTask用它表示任务的状态(尚未开始、正在运行、已完成以及取消)。在同步器类还可以管理一些额外的状态变量，例如:ReentrantLock保存了锁的当前持有者的信息，这样就能区分某个获取操作是重入的还是竞争的。

　　根据同步器的不同，获取操作可以看做是一种独占操作--Exclusive(例如ReentrantLock),也可以是一个非独占操作--共享shared(例如Semaphore和CountDownLatch)。

　　自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法：

isHeldExclusively()：该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。
tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。
tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。

　　

    boolean acquire(){
        while(当前状态不运行获取操作){
            if(需要阻塞获取请求){
                如果当前线程不在队列种，则将其插入队列
                阻塞当前线程
            }else {
                return false;
            }
        }
        可能更新同步容器的状态
        如果线程位于队列中，则将其移除队列
        return true;
    }
    boolean release(){
        更新同步容器的状态
        if(新的状态允许某个被阻塞的线程获取成功){
            借出队列中一个或多个线程的阻塞状态
        }
    }

　　上面是AQS的获取操作与释放操作的形式。

　　一个获取操作包括两部分。首先，同步器判断当前状态是否允许获得操作，如果是，则允许线程执行，否则获取操作将阻塞或失败。这种判断是同步器的语义决定的。例如:对于锁来说，如果它没有被某个线程持有，那么就能被成功地获取；而对于闭锁来说，如果它处于结束状态，那么也能成功地获取。其次，就是更新同步容器的状态，获取同步的某个线程可能对其他线程是否也获取该同步器造成影响。例如，当获取一个锁后，锁的状态从"未被持有"变为"已被持有"，而从Semaphore获取一个许可后，将把剩余的许可数量减一。然而，当一个线程获取闭锁时，并不会影响其他线程是否获取它，即获取闭锁的操作不会改变闭锁的状态。

　　如果某个同步器支持独占的获取操作，那么需要实现一些保护方法，包括tryAcquire、tryRelease和isHeldExclusively，而对于支持共享获取的同步器，则应该实现traAcquireShared和tryReleaseShared等方法。AQS的acquire和acquireShared、release和releaseShared等方法都将调用这些方法在子类中带有前缀try的版本来判断某个操作是否能执行。在同步器的子类中，可以通过getState、setState以及compareAndSetState来检查以及更新状态，并通过返回的值来告知基类获取或释放的操作是否成功。例如:如果tryAcquireShared返回一个负数表示操作获取失败，返回0值表示同步器通过独占方式被获取，返回正值则表示同步器通过非独占方式被获取。对于tryRelease和tryReleaseShared方法来说，如果释放操作使得所有在获取同步器时被阻塞的线程恢复执行，那么这两个方法应该返回true。

　　为了使支持条件队列的锁(ReentrantLock)实现起来更简单，AQS还提供了一些机制来构造与同步器相关联的条件变量。

 　　

例如一个简单的二元闭锁

package cn.qlq.thread.ttwo;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class OneShotLatch {
    private final Sync sync = new Sync();

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

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    private class Sync extends AbstractQueuedSynchronizer {
        @Override
        protected int tryAcquireShared(int arg) {
            // 如果闭锁是开的(state == 1),那么这个操作将成功，否则会失败
            return getState() == 1 ? 1 : -1;
        }

        @Override
        protected boolean tryReleaseShared(int arg) {
            setState(1);// 打开锁
            return true;// 返回成功
        }
    }
}

 

　　在上面代码中，AQS状态用来表示闭锁状态-关闭(0)与打开(1)。

　　await方法调用acquireSharedInterruptibly方法，acquireSharedInterruptibly是AbstractQueuedSynchronizer类的方法，acquireSharedInterruptibly方法中调用被Sync重写过的tryAcquireShared方法。

　　signal方法调用releaseShared方法，releaseShared是AbstractQueuedSynchronizer类的方法，releaseShared方法中调用被Sync重写的tryReleaseShared方法。

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

测试上面的二元闭锁:

    public static void main(String[] args) throws InterruptedException {
        final OneShotLatch oneShotLatch = new OneShotLatch();
        for (int i = 0; i < 3; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    log.info("threadname " + Thread.currentThread().getName() + "	 start");
                    try {
                        oneShotLatch.await();
                        log.info("threadname " + Thread.currentThread().getName() + "	 end");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }

        Thread.sleep(5 * 1000);
        oneShotLatch.signal();
    }

结果:

13:20:27 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-1 start
13:20:27 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-0 start
13:20:27 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-2 start
13:20:32 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-0 end
13:20:32 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-1 end
13:20:32 [cn.qlq.thread.ttwo.OneShotLatch]-[INFO] threadname Thread-2 end

 

2.java.util.concurrent包中的AQS

 　　java.util.concurrent包中有很多可阻塞类，例如ReentrantLock可重入锁、Semaphore信号量、ReentrantReadWriteLock读写锁、CountDownLatch闭锁、SynchronousQueue同步队列和FutureTask等，都是基于AQS构建的。

2.1 ReentrantLock中使用AQS

　　ReentrantLock只支持独占的方式获取操作，因此它实现了tryAcquire、tryRelease和isHeldExclusively。ReentrantLock将同步状态用于保存锁的获取次数，并且还维护一个owner变量保存当前所有者线程的标识符，只有在当前线程刚刚获取到锁或者正要释放锁的时候才会修改这个变量。在tryRelease中检查owner，从而确保当前线程在执行unlock操作之前已经获得了锁；在tryAcquire中将使用这个域来区分获取操作是竞争还是重入。

下面是公平锁的tryAcquire:

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

 

下面是ReentrantLock的静态内部类Sync的tryRelease的代码

        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.lock是调用了Sync.lock,然后调用对应公平锁和非公平锁的lock，进而调用acquire(1)--继承自AbstractQueuedSynchronizer

public class ReentrantLock implements Lock, java.io.Serializable {
private final Sync sync;
    public void lock() {
        sync.lock();
    }
...
}

abstract static class Sync extends AbstractQueuedSynchronizer {
　　abstract void lock();
}

static final class FairSync extends Sync {final void lock() {
            acquire(1);
        }
}

 

ReentrantLock还利用了AQS多个条件变量和多个等待线程集的内置支持。Lock.newCondition将返回一个新的ConditionObject实例，这是AQS的一个内部类。

2.2 Semaphore与CountDownLatch中使用AQS

　　Semaphore将AQS的状态用于保存当前许可的数量。tryAcquireShare方法首先计算许可剩余数量，如果没有剩余数量会返回一个值表示失败。如果有剩余的许可，那么会通过cpmpareAndSetState以原子方式来降低许可的数量。如果这个操作成功，那么将返回一个值表示获取成功。在返回值中还包含了表示其他共享操作能否成功的信息，如果成功，那么其他等待的线程通用会解除阻塞。

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

        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (next < current) // overflow
                    throw new Error("Maximum permit count exceeded");
                if (compareAndSetState(current, next))
                    return true;
            }
        }

 

　　CountDown与AQS使用非常类似。在同步状态保存的是当前的计数值。countDown方法调用release计数器递减，并且当计数器为0解除阻塞；await调用acquire，当计数器为0时acquire会立即返回，否则将阻塞。

2.3 FutureTask中使用AQS

　　Future.get语义上非常类似于闭锁的语义---如果发生了某个事件(由FutureTask表示的任务执行完成或被取消)，那么线程就可以恢复执行，否则这些线程将停留在队列中并直到该事件发生。

　　在FutureTask中，AQS同步状态保存的是任务的状态。例如:正在运行、已完成或者已取消。FutureTask还维护一些额外的状态变量，用于保存的计算结果或者抛出的异常。此外，它还维护了一个引用，指向正在执行计算任务的线程（处于执行状态），因而 如果任务取消，该线程就会中断。