并发编程-AQS

概念

AQS是AbstactQueuedSynchronizer的简称，它是一个Java提供的底层同步工具类，用一个int类型的变量表示同步状态，并提供了一系列的CAS操作来管理这个同步状态。AQS的主要作用是为Java中的并发同步组件提供统一的底层支持，例如ReentrantLock，CountdowLatch就是基于AQS实现的，用法是通过继承AQS实现其模版方法，然后将子类作为同步组件的内部类。

同步队列

一个双端队列，遵循FIFO原则，主要作用是用来存放在锁上阻塞的线程，当一个线程尝试获取锁时，如果已经被占用，那么当前线程就会被构造成一个Node节点加到同步队列的尾部，队列的头节点是成功获取锁的节点，当头节点线程释放锁时，会唤醒后面的节点并释放当前头节点的引用。

独占锁的获取和释放流程

######获取

1. 调用入口方法acquire(arg)
2. 调用模版方法tryAcquire(arg)尝试获取锁，若成功则返回，若失败则走下一步
3. 将当前线程构造成一个Node节点，并利用CAS将其加入到同步队列到尾部，然后该节点对应的线程进入自旋状态。
自旋时，首先判断其前驱节点是否为头节点&是否成功获取同步状态，两个条件都成立，则将当前线程的节点设置为头节点，如果不是，则利用LockSupport.park(this)将当前线程挂起 ,等待被前驱节点唤醒

######释放

1. 调用入口方法release(arg)

2. 调用模版方法tryRelease(arg)释放同步状态

3. 获取当前节点的下一个节点

4. 利用LockSupport.unpark(currentNode.next.thread)唤醒后继节点（接获取的第四步）

共享锁的获取和释放流程

######获取锁

1. 调用acquireShared(arg)入口方法
2. 进入tryAcquireShared(arg)模版方法获取同步状态，如果返返回值>=0，则说明同步状态(state)有剩余，获取锁成功直接返回
如果tryAcquireShared(arg)返回值<0，说明获取同步状态失败，向队列尾部添加一个共享类型的Node节点，随即该节点进入自旋状态
自旋时，首先检查前驱节点释放为头节点&tryAcquireShared()是否>=0(即成功获取同步状态)
如果是，则说明当前节点可执行，同时把当前节点设置为头节点，并且唤醒所有后继节点
如果否，则利用LockSupport.unpark(this)挂起当前线程，等待被前驱节点唤醒

######释放锁

调用releaseShared(arg)模版方法释放同步状态
如果释放成，则遍历整个队列，利用LockSupport.unpark(nextNode.thread)唤醒所有后继节点

独占锁和共享锁在实现上的区别

独占锁的同步状态值为1，即同一时刻只能有一个线程成功获取同步状态
共享锁的同步状态>1，取值由上层同步组件确定
独占锁队列中头节点运行完成后释放它的直接后继节点
共享锁队列中头节点运行完成后释放它后面的所有节点
共享锁中会出现多个线程（即同步队列中的节点）同时成功获取同步状态的情况

重入锁

重入锁指的是当前线成功获取锁后，如果再次访问该临界区，则不会对自己产生互斥行为。Java中对ReentrantLock和synchronized都是可重入锁，synchronized由jvm实现可重入即使，ReentrantLock都可重入性基于AQS实现。

同时，ReentrantLock还提供公平锁和非公平锁两种模式。

######ReentrantLock重入锁
重入锁的基本原理是判断上次获取锁的线程是否为当前线程，如果是则可再次进入临界区，如果不是，则阻塞。

由于ReentrantLock是基于AQS实现的，底层通过操作同步状态来获取锁，下面看一下非公平锁的实现逻辑：

final boolean nonfairTryAcquire(int acquires) {
            //获取当前线程
            final Thread current = Thread.currentThread();
            //通过AQS获取同步状态
            int c = getState();
            //同步状态为0，说明临界区处于无锁状态，
            if (c == 0) {
                //修改同步状态，即加锁
                if (compareAndSetState(0, acquires)) {
                    //将当前线程设置为锁的owner
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
            //如果临界区处于锁定状态，且上次获取锁的线程为当前线程
            else if (current == getExclusiveOwnerThread()) {
                 //则递增同步状态
                int nextc = c + acquires;
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc);
                return true;
            }
            return false;
        }

######非公平锁
非公平锁是指当锁状态为可用时，不管在当前锁上是否有其他线程在等待，新近线程都有机会抢占锁。

上述代码即为非公平锁和核心实现，可以看到只要同步状态为0，任何调用lock的线程都有可能获取到锁，而不是按照锁请求的FIFO原则来进行的。

######公平锁
公平锁是指当多个线程尝试获取锁时，成功获取锁的顺序与请求获取锁的顺序相同，下面看一个ReentrantLock的实现：

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

从上面的代码中可以看出，公平锁与非公平锁的区别仅在于是否判断当前节点是否存在前驱节点!hasQueuedPredecessors() &&，由AQS可知，如果当前线程获取锁失败就会被加入到AQS同步队列中，那么，如果同步队列中的节点存在前驱节点，也就表明存在线程比当前节点线程更早的获取锁，故只有等待前面的线程释放锁后才能获取锁。

#读写锁
Java提供了一个基于AQS到读写锁实现ReentrantReadWriteLock，该读写锁到实现原理是：将同步变量state按照高16位和低16位进行拆分，高16位表示读锁，低16位表示写锁。

######写锁的获取与释放
写锁是一个独占锁，所以我们看一下ReentrantReadWriteLock中tryAcquire(arg)的实现：

protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            int c = getState();
            int w = exclusiveCount(c);
            if (c != 0) {
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                setState(c + acquires);
                return true;
            }
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }

上述代码的处理流程已经非常清晰：

获取同步状态，并从中分离出低16为的写锁状态
如果同步状态不为0，说明存在读锁或写锁
如果存在读锁（c ！=0 && w == 0），则不能获取写锁（保证写对读的可见性）
如果当前线程不是上次获取写锁的线程，则不能获取写锁（写锁为独占锁）
如果以上判断均通过，则在低16为写锁同步状态上利用CAS进行修改（增加写锁同步状态，实现可重入）
将当前线程设置为写锁的获取线程
写锁的释放过程与独占锁基本相同：

protected final boolean tryRelease(int releases) {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            int nextc = getState() - releases;
            boolean free = exclusiveCount(nextc) == 0;
            if (free)
                setExclusiveOwnerThread(null);
            setState(nextc);
            return free;
        }

在释放的过程中，不断减少读锁同步状态，只为同步状态为0时，写锁完全释放。

######读锁的获取与释放
读锁是一个共享锁，获取读锁的步骤如下：

获取当前同步状态
计算高16为读锁状态+1后的值
如果大于能够获取到的读锁的最大值，则抛出异常
如果存在写锁并且当前线程不是写锁的获取者，则获取读锁失败
如果上述判断都通过，则利用CAS重新设置读锁的同步状态
读锁的获取步骤与写锁类似，即不断的释放写锁状态，直到为0时，表示没有线程获取读锁。

在JDK1.6以后，读锁的实现比上述过程更加复杂，有兴趣的同学可以看一下最新的后去读锁的源码。

用AQS写一个实例：

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class MyLock implements Lock {

    private Helper helper=new Helper();

    private class Helper extends AbstractQueuedSynchronizer{
        //获取锁
        @Override
        protected boolean tryAcquire(int arg) {
            int state=getState();
            if(state==0){
                //利用CAS原理修改state
                if(compareAndSetState(0,arg)){
                    //设置当前线程占有资源
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
            }else if(getExclusiveOwnerThread()==Thread.currentThread()){
                setState(getState()+arg);
                return true;
            }
            return false;
        }

        //释放锁
        @Override
        protected boolean tryRelease(int arg) {
            int state=getState()-arg;
            boolean flag=false;
            //判断释放后是否为0
            if(state==0){
                setExclusiveOwnerThread(null);
                setState(state);
                return true;
            }
            setState(state);//存在线程安全吗？重入性的问题，当前已经独占了资源()state
            return false;
        }

        public Condition newConditionObjecct(){
            return new ConditionObject();
        }
    }
    @Override
    public void lock() {
        helper.acquire(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        helper.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        return helper.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return helper.tryAcquireNanos(1,unit.toNanos(time));
    }

    @Override
    public void unlock() {
        helper.release(1);
    }

    @Override
    public Condition newCondition() {
        return helper.newConditionObjecct();
    }
}

public class Demo01 {
    private MyLock lock=new MyLock();
    private int m=0;
    public int next(){
       lock.lock();
        try {
            return m++;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        Demo01 demo=new Demo01();
        Thread[] th=new Thread[20];
        for (int i = 0; i < 20; i++) {
            th[i]=new Thread(()->{
                System.out.println(demo.next());
            });
            th[i].start();
        }
    }
}

public class Demo02 { //可重入
    private MyLock lock=new MyLock();
    private int m=0;
    public void a(){
        lock.lock();
        System.out.println("a");
        b();
        lock.unlock();
    }
    public void b(){
        lock.lock();
        System.out.println("b");
        lock.unlock();
    }

    public static void main(String[] args) {
        Demo02 demo=new Demo02();
        new Thread(()->{
            demo.a();
        }).start();
    }
}

import java.util.concurrent.locks.ReentrantLock;

public class Demo03 {
    private ReentrantLock lock=new ReentrantLock();
    private int m=0;
    public void a(){
        lock.lock();
        System.out.println("a");
        b();
        lock.unlock();
    }
    public void b(){
        lock.lock();
        System.out.println("b");
        lock.unlock();
    }

    public static void main(String[] args) {
        Demo03 demo=new Demo03();
        new Thread(()->{
            demo.a();
        }).start();
    }
}

CountDownLatch:

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class FightQueryDemo {

    private static List<String> company= Arrays.asList("东方航空","南方航空","海南航空");
    private static List<String> fightList=new ArrayList<>();

    public static void main(String[] args) throws InterruptedException{
        String origin="BJ";
        String dest="SH";
        Thread[] threads=new Thread[company.size()];
        CountDownLatch latch=new CountDownLatch(company.size());

        for (int i = 0; i < threads.length; i++) {
            String name=company.get(i);
            threads[i]=new Thread(()->{
                System.out.printf("%s 查询从%s到%s的机票
",name,origin,dest);
                //随机产生票数
                int val=new Random().nextInt(10);
                try {
                    TimeUnit.SECONDS.sleep(val);
                    fightList.add(name+"--"+val);
                    System.out.printf("%s公司查询成功！
",name);
                    latch.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            threads[i].start();
        }
        latch.await();
        System.out.println("==============查询结果如下：================");
        fightList.forEach(System.out::println);
    }
}

CyclicBarrier

import java.util.Random;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;

public class RaceDemo {

    public static void main(String[] args) {
        CyclicBarrier barrier=new CyclicBarrier(8);
        Thread[] play=new Thread[8];
        for (int i = 0; i < 8; i++) {
            play[i]=new Thread(()->{
                try {
                    TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                    System.out.println(Thread.currentThread().getName()+"准备好了");
                    barrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                System.out.println("选手"+Thread.currentThread().getName()+"起跑");
            },"play["+i+"]");
            play[i].start();
        }
    }
}