JUC 中提供了几个比较常用的并发工具类，比如 CountDownLatch、CyclicBarrier、Semaphore，他们都是基于AQS和Condition 来实现的

CountDownLatch

countdownlatch 是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完毕再执行。从命名可以解读到 countdown 是倒数的意思，类似于我们倒计时的概念。

countdownlatch 提供了两个方法，一个是 countDown，一个是 await， countdownlatch 初始化的时候需要传入一个整数，在这个整数倒数到 0 之前，调用了 await 方法的程序都必须要等待，然后通过 countDown 来倒数。

   public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(3);
        new Thread(() -> {
            System.out.println("" + Thread.currentThread().getName() + "-执行中");
            countDownLatch.countDown();
            System.out.println("" + Thread.currentThread().getName() + "-执行完毕");
        }, "t1").start();
        new Thread(() -> {
            System.out.println("" + Thread.currentThread().getName() + "-执行中");
            countDownLatch.countDown();
            System.out.println("" + Thread.currentThread().getName() + "-执行完毕");
        }, "t2").start();
        new Thread(() -> {
            System.out.println("" + Thread.currentThread().getName() + "-执行中");
            countDownLatch.countDown();
            System.out.println("" + Thread.currentThread().getName() + "-执行完毕");
        }, "t3").start();
        countDownLatch.await();
        System.out.println("所有线程执行完毕");
    }

从代码的实现来看，有点类似 join 的功能，但是比 join 更加灵活。CountDownLatch 构造函数会接收一个 int 类型的参数作为计数器的初始值，当调用 CountDownLatch 的 countDown方法时，这个计数器就会减一。
通过 await 方法去阻塞去阻塞主流程

image.png

模拟高并发场景

public class CountDownLatchDemo extends Thread{

    static CountDownLatch countDownLatch=new CountDownLatch(1);

    public static void main(String[] args) {
        for(int i=0;i<1000;i++){
            new CountDownLatchDemo().start();
        }
        countDownLatch.countDown();//释放
    }

    @Override
    public void run() {
        try {
            countDownLatch.await(); //阻塞  1000个线程 Thread.currentThread
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //TODO
        System.out.println("ThreadName:"+Thread.currentThread().getName());
    }
}

CountDownLatch 源码分析

对于 CountDownLatch，我们仅仅需要关心两个方法，一个是 countDown() 方法，另一个是 await() 方法。
countDown() 方法每次调用都会将 state 减 1，直到 state 的值为 0；而 await 是一个阻塞方法，当 state 减为 0 的时候，await 方法才会返回。await 可以被多个线程调用，大家在这个时候脑子里要有个图：所有调用了 await 方法的线程阻塞在 AQS 的阻塞队列中，等待条件满足（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);
    }

countdownlatch 也用到了 AQS，在 CountDownLatch 内部写了一个 Sync 并且继承了 AQS这个抽象类重写了 AQS 中的共享锁方法。这块代码主要是判断当前线程是否获取到了共享锁;（在 CountDownLatch 中，使用的是共享锁机制，因为CountDownLatch 并不需要实现互斥的特性）

看看doAcquireSharedInterruptibly方法

  private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
        final Node node = addWaiter(Node.SHARED); //创建一个共享模式的节点添加到队列中
        boolean failed = true;
        try {
            for (; ; ) {//被唤醒的线程进入下一次循环继续判断
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);//就判断尝试获取锁
                    if (r >= 0) {//r>=0 表示获取到了执行权限，这个时候因为 state!=0，所以不会执行这段代码
                        setHeadAndPropagate(node, r);
                        p.next = null; // 把当前节点移除 aqs 队列
                        failed = false;
                        return;
                    }
                }
                //阻塞线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

1. addWaiter 设置为 shared 模式。
2. tryAcquire 和 tryAcquireShared 的返回值不同，因此会多出一个判断过程
3. 在判断前驱节点是头节点后，调用了 setHeadAndPropagate 方法，而不是简单的更新一下头节点。

假如这个时候有 3 个线程调用了 await 方法，由于这个时候 state 的值还不为 0，所以这三
个线程都会加入到 AQS 队列中。并且三个线程都处于阻塞状态

image.png

接下来看看唤醒方法CountDownLatch.countDown

由于线程被 await 方法阻塞了，所以只有等到 countdown 方法使得 state=0 的时候才会被唤醒，我们来看看 countdown 做了什么

1. 只有当 state 减为 0 的时候，tryReleaseShared 才返回 true, 否则只是简单的 state = state - 1
2. 如果 state=0, 则调用 doReleaseShared 唤醒处于 await 状态下的线程

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

   public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {//子类实现
            doReleaseShared();
            return true;
        }
        return false;
    }

//用自旋的方法实现 state 减 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;
            }
        }
    }

 private void doReleaseShared() {
        for (; ; ) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue; // loop to recheck cases
                    unparkSuccessor(h);
                }
                // 这个 CAS 失败的场景是：执行到这里的时候，刚好有一个 节点入队，入队会将这个 ws 设置为 -1
                else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue; // loop on failed CAS
            }
            // 如果到这里的时候，前面唤醒的线程已经占领了 head，那么再循环
            // 通过检查头节点是否改变了，如果改变了就继续循环
            if (h == head) // loop if head changed
            break;
        }
    }

共享锁的释放和独占锁的释放有一定的差别
前面唤醒锁的逻辑和独占锁是一样，先判断头结点是不是 SIGNAL 状态，如果是，则修改为0，并且唤醒头结点的下一个节点
PROPAGATE： 标识为 PROPAGATE 状态的节点，是共享锁模式下的节点状态，处于这个状态下的节点，会对线程的唤醒进行传播

h == head：说明头节点还没有被刚刚用 unparkSuccessor 唤醒的线程（这里可以理解为ThreadB）占有，此时 break 退出循环。
h != head：头节点被刚刚唤醒的线程（这里可以理解为 ThreadB）占有，那么这里重新进入下一轮循环，唤醒下一个节点（这里是 ThreadB ）。我们知道，等到 ThreadB 被唤醒后，其实是会主动唤醒 ThreadC.
一旦 ThreadA 被唤醒，代码又会继续回到 doAcquireSharedInterruptibly 中来执行。如果当前 state 满足=0 的条件，则会执行 setHeadAndPropagate 方法

   private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

这个方法的主要作用是把被唤醒的节点，设置成 head 节点。 然后继续唤醒队列中的其他线程。
由于现在队列中有 3 个线程处于阻塞状态，一旦 ThreadA 被唤醒，并且设置为 head 之后，会继续唤醒后续的 ThreadB

image.png

Semaphore

semaphore 也就是我们常说的信号灯，semaphore 可以控制同时访问的线程个数，通过acquire 获取一个许可，如果没有就等待，通过 release 释放一个许可。有点类似限流的作用。
叫信号灯的原因也和他的用处有关，比如某商场就 5 个停车位，每个停车位只能停一辆车，如果这个时候来了 10 辆车，必须要等前面有空的车位才能进入。

     class Car extends  Thread{
        private int num;
        private Semaphore semaphore;

        public Car(int num, Semaphore semaphore) {
            this.num = num;
            this.semaphore = semaphore;
        }
        public void run(){
            try {
                semaphore.acquire(); //获得一个令牌, 如果拿不到令牌，就会阻塞
                System.out.println("第"+num+" 抢占一个车位");
                Thread.sleep(2000);
                System.out.println("第"+num+" 开走喽");
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

 public static void main(String[] args) {
        Semaphore semaphore=new Semaphore(5);
        for(int i=0;i<10;i++){
            new Car(i,semaphore).start();
        }
    }

从 Semaphore 的功能来看，我们基本能猜测到它的底层实现一定是基于 AQS 的共享锁

创建 Semaphore 实例的时候，需要一个参数 permits，这个基本上可以确定是设置给 AQS
的 state 的，然后每个线程调用 acquire 的时候，执行 state = state - 1，release 的时候执
行 state = state + 1，当然，acquire 的时候，如果 state = 0，说明没有资源了，需要等待
其他线程 release。

Semaphore 分公平策略和非公平策略

  static final class FairSync extends Sync {
        private static final long serialVersionUID =
                2014338818796000944L;
        FairSync(int permits) {
            super(permits);
        }
        protected int tryAcquireShared(int acquires) {
            for (;;) {
                // 区别就在于是不是会先判断是否有线程在排队，然后才进行 CAS 减操作
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                return remaining;
            }
        }
    }

    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;
        NonfairSync(int permits) {
            super(permits);
        }
        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }
    final int nonfairTryAcquireShared(int acquires) {
        for (;;) {
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                return remaining;
        }
    }

通过对比发现公平和非公平的区别就在于是否多了一个 hasQueuedPredecessors 的判断,由于后面的代码和 CountDownLatch 的是完全一样，都是基于共享锁的实现，所以大家可以自行分析源码

CyclicBarrier

CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会 开 门 ， 所有被 屏 障 拦 截的线 程 才 会 继续工 作 。 CyclicBarrier 默 认 的构造 方 法 是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用 await 方法告诉CyclicBarrier 当前线程已经到达了屏障，然后当前线程被阻塞

使用场景
当存在需要所有的子任务都完成时，才执行主任务，这个时候就可以选择使用 CyclicBarrier

public class CycliBarrierDemo extends Thread {
    @Override
    public void run() {
        System.out.println("开始进行数据分析");
    }

    //循环屏障
    //可以使得一组线程达到一个同步点之前阻塞.
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, new CycliBarrierDemo());
        new Thread(new DataImportThread(cyclicBarrier, "file1")).start();
        new Thread(new DataImportThread(cyclicBarrier, "file2")).start();
        new Thread(new DataImportThread(cyclicBarrier, "file3")).start();

    }
}

注意点
1）对于指定计数值 parties，若由于某种原因，没有足够的线程调用 CyclicBarrier 的 await，则所有调用 await 的线程都会被阻塞；
2）同样的 CyclicBarrier 也可以调用 await(timeout, unit)，设置超时时间，在设定时间内，如果没有足够线程到达，则解除阻塞状态，继续工作；
3）通过 reset 重置计数，会使得进入 await 的线程出现 BrokenBarrierException；
4 ）如果采用是 CyclicBarrier(int parties, Runnable barrierAction) 构 造 方 法 ， 执 行barrierAction 操作的是最后一个到达的线程

实现原理
CyclicBarrier 相比 CountDownLatch 来说，要简单很多，源码实现是基于 ReentrantLock 和 Condition 的组合使用。CyclicBarrier 和 CountDownLatch 很像，只是 CyclicBarrier 可以有不止一个栅栏，因为它的栅栏（Barrier）可以重复使用（Cyclic）

——学自咕泡学院