CyclicBarrier概述
- 允许多个线程相互等待，直到到达某个公共屏障点（Common Barrier Point），只有当每个线程都准备就绪后，才能各自往下执行后面的操作；
- 其和CountDownLatch相似的地方是都是通过计数器实现；
- 当某个现场调用await()后，该线程就进入等待状态，而且计数器执行加1操作，当计数器达到了设置的初始值，调用await()方法进行等待的线程都会被唤醒，继续await()之后的操作；
- 由于CyclicBarrier在释放等待线程后可以重用，所以称其为循环屏障；
应用场景
- 多线程计算数据，最后合并计算结果；
- 如果发生错误，可以重置计数器，让线程重新执行1次；
CyclicBarrier vs CountDownLatch
- CountDownLatch的计数器只能使用1次；CyclicBarrier的计数器可以使用reset方法重置；
- CountDownLatch主要实现1个或多个线程，需要等待其他线程完成某项操作后，才能继续往下执行；CyclicBarrier主要实现了多个线程之间相互等待，直到所有的线程都满足了条件之后，才能继续执行后续操作；
- CyclicBarrier的getNumberWaiting()方法，获得线程的阻塞数量；
- CyclicBarrier的isBroken知道阻塞的线程是否已经被中断了；
CyclicBarrier 示例（一）
- 初始化的时候给定一个值，表示有多少个线程要相互等待；
- 每个线程在到达公共屏障点的时候要调用await()方法；
- 当所有的线程都到达了公共屏障点后，所有线程之后的代码就可以执行了；
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample1 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}
CyclicBarrier 示例（二） 等不及先走了
- barrier.await(2000, TimeUnit.MILLISECONDS)在等待的时候可以设置一个最大等待实现，如果在在这个时间内，所有的线程都到达了，那么一起往下走；
- 如果在这个时间还有线程没有到，就不等了，自己往前走；
- 但是注意，时间到还没来全时，会抛出一个异常，如果在异常抛出后后续代码还想执行，那么就要catch住这个异常；
- 如果在这个时间内，都到齐了，那么一起走，没问题；
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CyclicBarrierExample2 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try {
barrier.await(2000, TimeUnit.MILLISECONDS);
} catch (Exception e) {
log.warn("{} occur BarrierException", threadNum);
}
log.info("{} continue", threadNum);
}
}
CyclicBarrier 示例（三） CyclicBarrier可以设置“回调函数”
- CyclicBarrier在初始化的时候，可以设置一个Runnable作为回调函数，其意义是：当都到达了屏障后，先于各个线程的后续方法执行回调函数；
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample3 {
private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
log.info("callback is running");
});
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}
输出：
19:58:59.623 [pool-1-thread-1] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 0 is ready
19:59:00.620 [pool-1-thread-2] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 1 is ready
19:59:01.621 [pool-1-thread-3] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 2 is ready
19:59:02.620 [pool-1-thread-4] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 3 is ready
19:59:03.621 [pool-1-thread-5] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 4 is ready
19:59:03.621 [pool-1-thread-5] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - callback is running
19:59:03.621 [pool-1-thread-5] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 4 continue
19:59:03.621 [pool-1-thread-1] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 0 continue
19:59:03.621 [pool-1-thread-2] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 1 continue
19:59:03.621 [pool-1-thread-3] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 2 continue
19:59:03.621 [pool-1-thread-4] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 3 continue
19:59:04.621 [pool-1-thread-6] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 5 is ready
19:59:05.622 [pool-1-thread-4] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 6 is ready
19:59:06.623 [pool-1-thread-3] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 7 is ready
19:59:07.624 [pool-1-thread-1] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 8 is ready
19:59:08.624 [pool-1-thread-2] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 9 is ready
19:59:08.624 [pool-1-thread-2] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - callback is running
19:59:08.624 [pool-1-thread-2] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 9 continue
19:59:08.624 [pool-1-thread-6] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 5 continue
19:59:08.624 [pool-1-thread-4] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 6 continue
19:59:08.624 [pool-1-thread-3] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 7 continue
19:59:08.624 [pool-1-thread-1] INFO com.example.concurrency.example.aqs.CyclicBarrierExample3 - 8 continue
