CountDownLatch和CyclicBarrier

CountDownLatch

线程同步协作，await等待其他所有线程完成倒计时后，恢复运行
state的初始值为count
内部维护了一个AQS同步器，每次countDown后，会进行CAS修改state减1，修改后state为0，则唤醒被阻塞的线程

CountDownLatch是共享锁的一种实现
调用await的时候，如果state不为0，就证明任务还没有执行完毕，await会一直阻塞，await方法之后的语句不会被执行。
然后，CountDownLatch会自旋CAS判断state是否为0，如果state为0，就会释放所有等待的线程，await之后的语句得到执行

java">CountDownLatch latch = new CountDownLatch(3); // 计数值一般和创建的线程数相同
new Thread(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    latch.countDown(); // 计数减一
}).start();
new Thread(() -> {
    try {
        TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    latch.countDown();
}).start();
new Thread(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    latch.countDown();
}).start();

log.debug("waiting...");
latch.await(); // 阻塞, 等待计数归0(也即3个线程执行完毕), 主线程被唤醒
log.debug("wait end...");

java">CountDownLatch latch = new CountDownLatch(3);
ExecutorService pool = Executors.newFixedThreadPool(4);
for (int i = 0; i < 3; i++) {
    pool.submit(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.debug("{}: 计数减一",Thread.currentThread().getName());
        latch.countDown();
    });
}
pool.submit(() -> {
    log.debug("waiting...");
    try {
        latch.await(); // 等待上面三个线程跑完任务
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    log.debug("wait end");
});

java">ExecutorService pool = Executors.newFixedThreadPool(10); // 10名玩家
CountDownLatch latch = new CountDownLatch(10); // 计数
Random random = new Random();
String[] all = new String[10];

for (int i = 0; i < 10; i++) {
    int d = i;
    pool.submit(() -> {
        for (int j = 0; j <= 100; j++) {
            try {
                Thread.sleep(random.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            all[d] = j + "%";
            System.out.print("\r" + Arrays.toString(all));
        }
        latch.countDown();
    });
}

latch.await(); // 等待10个玩家都加载到"100%"
System.out.println("\n游戏开始");

CountDownLatch是一次性的，计算器的值只能在构造器中设置，之后不可以再次设置，CountDownLatch使用完毕后，不能再次被使用。

CyclicBarrier

线程池的线程数要和计数值一致
CountDownLatch的实现是基于AQS，CycliBarrier是基于ReentrantLock(ReentrantLock也属于AQS同步器)和Condition。

CyclicBarrier内部维护count变量作为计数器，count的初始值为parties的值。
每个线程到了栅栏这里，将count减1，减1后不是0，就到Condition中阻塞。
count减1后变为0了，表示当前线程是这一代栅栏等待的最后一个线程，当前线程执行构造器中传的任务，然后唤醒之前被阻塞的所有线程。
然后重置count，开启下一代。

java">// 场景:有一个栅栏(或者称为障碍物),假设构造时parties传了3
// 前2个线程调用await被阻塞住,第3个线程来了,栅栏才会放行,3个线程才可以继续执行后面的代码

// 第1个和第2个线程调用了await, count值减1, 减完之后不是0, 所以这两个线程都会到Condition休息室中等待
// 第3个线程调用await时,count值减1之后变为0,(如果构造CyclicBarrier时传了task,则第3个线程先会执行那个task)
// 然后第3个线程唤醒Condition中的所有线程,并且将count重置为3,然后第3个线程继续执行await后面的代码
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;// 一代中栅栏需要等待的线程
    this.count = parties;// 计数器(每次调用await, count减1)
    this.barrierCommand = barrierAction; // 第parties个线程调用await后,会执行这个task,然后唤醒Condition中的线程
}
public CyclicBarrier(int parties) {
    this(parties, null);
}

java">private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll(); // 唤醒Condition中的所有线程
    // set up next generation
    count = parties; // 重置count
    generation = new Generation();
}

int index = --count;
if (index == 0) {  // tripped
    boolean ranAction = false;
    try {
        final Runnable command = barrierCommand;
        if (command != null)
            command.run(); // 是由第parties个线程执行
        ranAction = true;
        nextGeneration();
        return 0;
    } finally {
        if (!ranAction)
            breakBarrier();
    }
}

应用：

java">ExecutorService pool = Executors.newFixedThreadPool(2);
// 内部维护的count为0后, (一代中最后一个进入barrier的线程)还会将其置为2的
// 构造器传的task, 也是由(一代中最后一个进入barrier的线程)来执行
CyclicBarrier barrier = new CyclicBarrier(2,() -> log.debug("所有任务执行完毕"));
for (int i = 0; i < 3; i++) {
    pool.submit(() -> {
        log.debug("task1 begin ====>");
        try {
            barrier.await(); // 每次await,count会减1
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    });
    pool.submit(() -> {
        log.debug("task2 begin ====>");
        try {
            barrier.await();
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    });
}

02:51:54.950 [pool-1-thread-1] DEBUG cn.study.CountDown - task1 begin ====>
02:51:54.950 [pool-1-thread-2] DEBUG cn.study.CountDown - task2 begin ====>
02:51:54.952 [pool-1-thread-2] DEBUG cn.study.CountDown - 所有任务执行完毕// 这个task是由(这一代中最后一个进入barrier的线程)运行的
02:51:54.953 [pool-1-thread-2] DEBUG cn.study.CountDown - task2 begin ====>
02:51:54.953 [pool-1-thread-1] DEBUG cn.study.CountDown - task1 begin ====>
02:51:54.953 [pool-1-thread-1] DEBUG cn.study.CountDown - 所有任务执行完毕// 这个task是由(这一代中最后一个进入barrier的线程)运行的
02:51:54.953 [pool-1-thread-1] DEBUG cn.study.CountDown - task1 begin ====>
02:51:54.953 [pool-1-thread-2] DEBUG cn.study.CountDown - task2 begin ====>
02:51:54.953 [pool-1-thread-2] DEBUG cn.study.CountDown - 所有任务执行完毕// 这个task是由(这一代中最后一个进入barrier的线程)运行的