Java处理并发编程工具集合(JUC)详解2-阿里云开发者社区

4 AQS

4.1 前言

如果要想真正的理解JUC下的并发工具的实现原理，我们必须要来学习AQS，因为它是JUC下很多类的基石。

在讲解AQS之前，如果老板让你自己写一个SDK层面的锁，给其他同事去使用，你会如何写呢？

1、搞一个状态标记，用来表示持有或未持有锁，但得是volatile类型的保证线程可见性。

2、编写一个lock，unlock函数用于抢锁和释放锁，就是对状态标记的修改操作

3、lock函数要保证并发下只能有一个线程能抢到锁，其他线程要等待获取锁（阻塞式），可以采用CAS+自旋的方式实现

初步实现如下：

public class MyLock {
    // 定义一个状态变量status：为1表示锁被持有，为0表示锁未被持有
    private volatile int status;

    private static final Unsafe unsafe = reflectGetUnsafe();
    private static final long valueOffset;

    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                    (MyLock.class.getDeclaredField("status"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    private static Unsafe reflectGetUnsafe() {
        try {
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            return (Unsafe) field.get(null);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 阻塞式获取锁
     * @return
     */
    public boolean lock() {
        while (!compareAndSet(0,1)) {

        }
        return true;
    }

    // cas 设置 status
    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }

    /**
     * 释放锁
     */
    public void unlock() {
        status = 0;
    }
}

问题：获取不到锁自旋时，是空转，浪费CPU

1、使用yield让出CPU执行权，等待调度

public boolean lock() {
    while (!compareAndSet(0,1)) {
        Thread.yield();//yield+自旋,尽可能的防止CPU空转,让出CPU资源
    }
    return true;
}

或者可以采用线程休眠的方式，但是休眠时间不太好确定，太长太短都不好。

2、采用等待唤醒机制，但是这里由于没有使用synchronized关键字，所以也无法使用wait/notify，但是我们可以使用park/unpark，获取不到锁的线程park并且去队列排队，释放锁时从队列拿出一个线程unpark

private static final Queue<Thread> QUEUE = new LinkedBlockingQueue<>();

public boolean lock() {
    while (!compareAndSet(0,1)) {
        QUEUE.offer(Thread.currentThread());
        LockSupport.park();//线程休眠
    }
    return true;
}

public void unlock() {
    status = 0;
    LockSupport.unpark(QUEUE.poll());
}

4.2 AQS概述

AQS（AbstractQueuedSynchronizer）：抽象队列同步器，定义了一套多线程访问共享资源的同步器框架，提供了SDK层面的锁机制，JUC中的很多类譬如：ReentrantLock/Semaphore/CountDownLatch…等都是基于它。

通过查阅作者的对于该类的文档注释可以得到如下核心信息：

1、AQS用一个volatile int state;属性表示锁状态，1表示锁被持有，0表示未被持有，具体的维护由子类去维护，但是提供了修改该属性的三个方法：getState()，setState(int newState)，compareAndSetState(int expect, int update)，其中CAS方法是核心。

2、框架内部维护了一个FIFO的等待队列，是用双向链表实现的，我们称之为CLH队列，

3、框架也内部也实现了条件变量Condition，用它来实现等待唤醒机制，并且支持多个条件变量

4、AQS支持两种资源共享的模式：独占模式（Exclusive）和共享模式（Share），所谓独占模式就是任意时刻只允许一个线程访问共享资源，譬如ReentrantLock；而共享模式指的就是允许多个线程同时访问共享资源，譬如Semaphore/CountDownLatch

5、使用者只需继承AbstractQueuedSynchronizer并重写指定的方法，在方法内完成对共享资源state的获取和释放，至于具体线程等待队列的维护，AQS已经在顶层实现好了，在那些final的模板方法里。

* <p>To use this class as the basis of a synchronizer, redefine the
 * following methods, as applicable, by inspecting and/or modifying
 * the synchronization state using {@link #getState}, {@link
 * #setState} and/or {@link #compareAndSetState}:
 *
 * <ul>
 * <li> {@link #tryAcquire}
 * <li> {@link #tryRelease}
 * <li> {@link #tryAcquireShared}
 * <li> {@link #tryReleaseShared}
 * <li> {@link #isHeldExclusively}
 * </ul>
 *
 * Each of these methods by default throws {@link
 * UnsupportedOperationException}.  Implementations of these methods
 * must be internally thread-safe, and should in general be short and
 * not block. Defining these methods is the <em>only</em> supported
 * means of using this class. All other methods are declared
 * {@code final} because they cannot be independently varied.

6、AQS底层使用了模板方法模式，给我们提供了许多模板方法，我们直接使用即可。

API	说明
final void acquire(int arg)	独占模式获取锁，AQS顶层已实现，内部调用了tryAcquire	模板方法

boolean tryAcquire(int arg)	独占模式尝试获取锁，AQS中未实现，由子类去实现，获取到锁返回true
final boolean release(int arg)	释放独占锁，AQS顶层已实现，内部调用了tryRelease	模板方法

boolean tryRelease(int arg)	尝试释放独占锁，AQS中未实现，由子类去实现，成功释放返回true
final void acquireShared(int arg)	共享模式获取锁，AQS顶层已实现，内部调用了tryAcquireShared	模板方法

int tryAcquireShared(int arg)

尝试获取共享锁，返回负数表示失败，0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源，AQS中未实现，由子类实现

final boolean releaseShared(int arg)

释放共享锁，返回true代表释放成功，AQS中已实现，内部调用了tryReleaseShared

模板方法

boolean tryReleaseShared(int arg)

尝试释放锁，释放后允许唤醒后续等待结点返回true，否则返回false，AQS中未实现，需要由子类实现

boolean isHeldExclusively()

共享资源是否被独占

4.3 基本使用

此时老板给你加了需求，要求你实现一个基于AQS的锁，那该怎么办呢？

在AbstractQueuedSynchronizer的类注释中给出了使用它的基本方法，我们按照它的写法尝试即可

/**
 * 基于 aqs实现锁
 */
public class MyLock implements Lock {

    //同步器
    private Syn syn = new Syn();

    @Override
    public void lock() {
        //调用模板方法
        syn.acquire(1);
    }

    @Override
    public void unlock() {
        //调用模板方法
        syn.release(0);
    }

    // 其他接口方法暂时先不实现 省略


    // 实现一个独占同步器
    class Syn extends AbstractQueuedSynchronizer{
        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0,arg)) {
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            setState(arg);
            return true;
        }

    }

}

4.4 原理解析

自己实现的锁在使用过程中发现一个问题，就是有时候有的线程特别容易抢到锁，而有的线程老是抢不到锁，虽说线程们抢锁确实看命，但能不能加入一种设计，让各个线程机会均等些，起码不要出现某几个线程总是特倒霉抢不到锁的情况吧！

这其实就是涉及到锁是否是公平的，那么什么是公平锁什么是非公平锁呢？

这我们就不得不深入我们使用的模板方法中看一眼了

/**
     * Acquires in exclusive mode, ignoring interrupts.  Implemented
     * by invoking at least once {@link #tryAcquire},
     * returning on success.  Otherwise the thread is queued, possibly
     * repeatedly blocking and unblocking, invoking {@link
     * #tryAcquire} until success.  This method can be used
     * to implement method {@link Lock#lock}.
     *
     * @param arg the acquire argument.  This value is conveyed to
     *        {@link #tryAcquire} but is otherwise uninterpreted and
     *        can represent anything you like.
     */
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

//结合我自己写的尝试获取锁的方法
protected boolean tryAcquire(int arg) {
    if (compareAndSetState(0,arg)) {
        return true;
    }
    return false;
}

这里大概描述如下：

1、线程一来首先调用tryAcquire，在tryAcquire中直接CAS获取锁，如果获取不成功通过addWaiter加入等待队列，然后走acquireQueued让队列中的某个等待线程去获取锁。

2、不公平就体现在这里，线程来了也不先看一下等待队列中是否有线程在等待，如果没有线程等待，那直接获取锁没什么问题，如果有线程等待就直接去获取锁不就相当于插队么？

那如何实现这种公平性呢？这就不得不探究一下AQS的内部的实现原理了，下面我们依次来看：

1、查看AbstractQueuedSynchronizer的类定义，虽然它里面代码很多，但重要的属性就那么几个，

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    private volatile int state;
    private transient volatile Node head;
    private transient volatile Node tail;
    
    static final class Node {
        //其他不重要的略
        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
    }
    
    public class ConditionObject implements Condition, java.io.Serializable {...}
}

结合前面讲的AQS的类文档注释不难猜到，内部类 Node以及其类型的变量 head 和 tail 就表示 AQS 内部的一个等待队列，而剩下的 state 变量就用来表示锁的状态。

等待队列应该就是线程获取锁失败时，需要临时存放的一个地方，用来等待被唤醒并尝试获取锁。再看 Node 的属性我们知道，Node 存放了当前线程的指针 thread，也即可以表示当前线程并对其进行某些操作，prev 和 next 说明它构成了一个双向链表，也就是为某些需要得到前驱或后继节点的算法提供便利。

2、AQS加锁最核心的代码就是如下，我们要来探究它的实现原理

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

它的原理及整个过程我们以图的形式说明如下：

3、原理搞懂了，那如何让自定义的锁是公平的呢？

其实导致不公平的原因就是线程每次调用acquire时，都会先去tryAcquire，而该方法目前的实现时直接去抢锁，也不看现在等待队列中有没有线程在排队，如果有线程在排队，那岂不是变成了插队，导致不公平。

所以现在的解决办法就是，在tryAcquire时先看一下等待队列中是否有在排队的，如果有那就乖乖去排队，不插队，如果没有则可以直接去获取锁。

那如何知道线程AQS等待队列中是否有线程排队呢？其实AQS顶层已经实现好了，它提供了一个hasQueuedPredecessors函数：如果在当前线程之前有一个排队的线程，则为True; 如果当前线程位于队列的头部(head.next)或队列为空，则为false。

protected boolean tryAcquire(int arg) {
    //先判断等待队列中是否有线程在排队 没有线程排队则直接去获取锁
    if (!hasQueuedPredecessors() && compareAndSetState(0,arg)) {
        return true;
    }
    return false;
}

4、现在已经有公平锁了，但是成年人的世界不是做选择题，而是都想要，自己编写的锁既能支持公平锁，也支持非公平锁，让使用者可以自由选择，怎么办？

其实只要稍微改造一下即可，

public class MyLock implements Lock {

    //同步器
    private Sync syn ;

    MyLock () {
        syn = new NoFairSync();
    }

    MyLock (boolean fair) {
        syn = fair ? new FairSync():new NoFairSync();
    }


    @Override
    public void lock() {
        //调用模板方法
        syn.acquire(1);
    }

    @Override
    public void unlock() {
        //调用模板方法
        syn.release(0);
    }

    // Lock接口其他方法暂时先不实现 略
    


    // 实现一个独占同步器
    class Sync extends AbstractQueuedSynchronizer{


        @Override
        protected boolean tryRelease(int arg) {
            setState(arg);
            return true;
        }

    }

    class FairSync extends Sync {
        @Override
        protected boolean tryAcquire(int arg) {
            //先判断等待队列中是否有线程在排队 没有线程排队则直接去获取锁
            if (!hasQueuedPredecessors() && compareAndSetState(0,arg)) {
                return true;
            }
            return false;
        }
    }

    class NoFairSync extends Sync {
        @Override
        protected boolean tryAcquire(int arg) {
            //直接去获取锁
            if (compareAndSetState(0,arg)) {
                return true;
            }
            return false;
        }
    }

}

5、现在锁的公平性问题解决了，但是老板又出了新的需求，要求我们的锁支持可重入，因为它写了如下一段代码，发现一直获取不到锁

static Lock lock = new MyLock();
static void test3() {
    lock.lock();
    try {
        System.out.println("test3 get lock,then do something ");
        test4();
    } finally {
        lock.unlock();
    }
}
static void test4() {
    lock.lock();
    try {
        System.out.println("test4 get lock,then do something ");
    } finally {
        lock.unlock();
    }
}

那如何让锁支持可重入呢？也就是说如果一个线程持有锁之后，还能继续获取锁，也就是说让锁只对不同线程互斥。

查看AbstractQueuedSynchronizer的定义我们发现，它还继承自另一个类：AbstractOwnableSynchronizer

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
    implements java.io.Serializable {...}

public abstract class AbstractOwnableSynchronizer
    implements java.io.Serializable {
    
    private transient Thread exclusiveOwnerThread;
    
    protected final void setExclusiveOwnerThread(Thread thread) {...}
    protected final Thread getExclusiveOwnerThread(){...}
}

看到这我们明白了，原来AQS中有个变量是可以保存当前持有独占锁的线程的。那好办了，当我们获取锁时，如果发现锁被持有不要着急放弃，先看看持有锁的线程是否时当前线程，如果是还能继续获取锁。

另外关于可重入锁，还要注意一点，锁的获取和释放操作是成对出现的，就像下面这样

lock
    lock
        lock
            lock
                ....
            unlock
        unlock
    unlock
unlock

所以对于重入锁不仅要能记录锁被持有，还要记录重入的次数，释放的时候也不是直接将锁真实的释放，而是先减少重入次数，能释放的时候在释放。

故此时状态变量state不在只有两个取值0,1，某线程获取到锁state=1，如果当前线程重入获取只需增加状态值state=2，依次同理，锁释放时释放一次状态值-1，当state=0时才真正释放，其他线程才能继续获取锁。

修改我们锁的代码如下：公平非公平在可重入上的逻辑是一样的

public class MyLock implements Lock {

    //同步器
    private Sync syn ;

    MyLock () {
        syn = new NoFairSync();
    }

    MyLock (boolean fair) {
        syn = fair ? new FairSync():new NoFairSync();
    }


    @Override
    public void lock() {
        //调用模板方法
        syn.acquire(1);
    }

    @Override
    public void unlock() {
        //调用模板方法
        syn.release(1);
    }

    // Lock接口其他方法暂时先不实现 略


    // 实现一个独占同步器
    class Sync extends AbstractQueuedSynchronizer{


        @Override
        protected boolean tryRelease(int arg) {
            if (Thread.currentThread() != getExclusiveOwnerThread()) {
                throw new IllegalMonitorStateException();
            }
            boolean realRelease = false;
            int nextState = getState() - arg;
            if (nextState == 0) {
                realRelease = true;
                setExclusiveOwnerThread(null);
            }
            setState(nextState);
            return realRelease;
        }

    }

    class FairSync extends Sync {
        @Override
        protected boolean tryAcquire(int arg) {
            final Thread currentThread = Thread.currentThread();
            int currentState = getState();
            if (currentState == 0 ) { // 可以获取锁
                //先判断等待队列中是否有线程在排队 没有线程排队则直接去获取锁
                if (!hasQueuedPredecessors() && compareAndSetState(0,arg)) {
                    setExclusiveOwnerThread(currentThread);
                    return true;
                }
            }else if (currentThread == getExclusiveOwnerThread()) {
                //重入逻辑 增加 state值
                int nextState = currentState + arg;
                if (nextState < 0) {
                    throw new Error("Maximum lock count exceeded");
                }
                setState(nextState);
                return true;
            }
            return false;
        }
    }

    class NoFairSync extends Sync {
        @Override
        protected boolean tryAcquire(int arg) {
            final Thread currentThread = Thread.currentThread();
            int currentState = getState();
            if (currentState ==0 ) { // 可以获取锁
                //直接去获取锁
                if (compareAndSetState(0,arg)) {
                    setExclusiveOwnerThread(currentThread);
                    return true;
                }
            }else if (currentThread == getExclusiveOwnerThread()) {
                //重入逻辑 增加 state值
                int nextState = currentState + arg;
                if (nextState < 0) {
                    throw new Error("Maximum lock count exceeded");
                }
                setState(nextState);
                return true;
            }
            return false;
        }
    }

}

好了至此我们已经掌握了AQS的核心原理以及它的一个经典实现ReentrantLock几乎全部的知识点，此时打开ReentrantLock的源码你会发现一切都很清爽！！！

5 并发容器

juc中还包含很多其他的并发容器（了解）

1.ConcurrentHashMap

对应：HashMap

目标：代替Hashtable、synchronizedMap，使用最多，源码篇会详细讲解

原理：JDK7中采用Segment分段锁，JDK8中采用CAS+synchronized

2.CopyOnWriteArrayList

对应：ArrayList

目标：代替Vector、synchronizedList

原理：高并发往往是读多写少的特性，读操作不加锁，而对写操作加Lock独享锁，先复制一份新的集合，在新的集合上面修改，然后将新集合赋值给旧的引用，并通过volatile 保证其可见性。

查看源码：volatile array，lock加锁，数组复制

3.CopyOnWriteArraySet

对应：HashSet

目标：代替synchronizedSet

原理：与CopyOnWriteArrayList实现原理类似。

4.ConcurrentSkipListMap

对应：TreeMap

目标：代替synchronizedSortedMap(TreeMap)

原理：基于Skip list（跳表）来代替平衡树，按照分层key上下链接指针来实现。

附加：跳表

5.ConcurrentSkipListSet

对应：TreeSet

目标：代替synchronizedSortedSet(TreeSet)

原理：内部基于ConcurrentSkipListMap实现，原理一致

6.ConcurrentLinkedQueue

对应：LinkedList

对应：无界线程安全队列

原理：通过队首队尾指针，以及Node类元素的next实现FIFO队列

7.BlockingQueue

对应：Queue

特点：拓展了Queue，增加了可阻塞的插入和获取等操作

原理：通过ReentrantLock实现线程安全，通过Condition实现阻塞和唤醒

实现类：

LinkedBlockingQueue：基于链表实现的可阻塞的FIFO队列
ArrayBlockingQueue：基于数组实现的可阻塞的FIFO队列
PriorityBlockingQueue：按优先级排序的队列