CyclicBarrier是什么?
CyclicBarrier
也叫同步屏障,在JDK1.5被引入,可以让一组线程达到一个屏障时被阻塞,直到最后一个线程达到屏障
时,屏障才会开门
,所有
被阻塞的线程才会继续执行
。
他的主要用途是控制多个线程互相等待
,只有当多个线程都到达时,这些线程才会继续执行。
CyclicBarrier好比一扇门
,默认情况下关闭状态
,堵住了线程执行的道路,直到所有线程都就位,门才打开,让所有线程一起通过。
CyclicBarrier如何使用和工作?
API
CyclicBarrier有两个构造函数
public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)
第一个参数,其参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier已经到达屏障位置,线程被阻塞。
第二个参数,表示线程都处于barrier时,一起执行之前,其中barrierAction任务会在所有线程到达屏障后执行。
让线程处于barrier状态的方法await()
public int await()
public int await(long timeout, TimeUnit unit)
第一个默认方法,表示要等到所有的线程都处于barrier状态,才一起执行
第二个方法,指定了等待的时间,当所有线程没有都处于barrier状态,又到了指定的时间,所在的线程就继续执行了。
其它的一些方法
获取当前有多少个线程阻塞等待在临界点上
int getNumberWaiting()
用于查询阻塞等待的线程是否被中断
boolean isBroken()
CyclicBarrier是通过维护计数器来实现的。线程执行 await() 方法之后计数器会减 1
(而不是很多人认为的加1)
,并进行等待,直到计数器为 0,所有调用 await() 方法而在等待的线程才能继续执行。
CyclicBarrier的底层原理
CyclicBarrier类是concurrent并发包下的一工具类。CyclicBarrier实现主要基于ReentrantLock
。
线程间同步阻塞是使用的是ReentrantLock
,可重入锁
线程间通信使用的是Condition
,Condition 将 Object 监视器方法(wait、notify 和 notifyAll)分解成截然不同的对象,以便通过将这些对象与任意 Lock 实现组合使用。
public class CyclicBarrier {
private static class Generation {
boolean broken = false;
}
/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped */
private final Condition trip = lock.newCondition();
/** The number of parties */
private final int parties;
/* The command to run when tripped */
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();
...省略后面代码
}
其中Generation用来控制屏障的循环使用,如果generation.broken为true的话,说明这个屏障已经损坏,当某个线程await的时候,直接抛出异常.
await实现
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
int index = --count;
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// We're about to finish waiting even if we had not
// been interrupted, so this interrupt is deemed to
// "belong" to subsequent execution.
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
-
每当线程执行await,内部变量count减1,如果count!= 0,说明有线程还未到屏障处,则在锁条件变量trip上等待。
当count == 0时,说明所有线程都已经到屏障处,执行条件变量的signalAll方法唤醒等待的线程。 -
其中 nextGeneration方法可以实现屏障的循环使用:
- 重新生成Generation对象
- 恢复count值
CountDownLatch与CyclicBarrier的比较
CountDownLatch
与CyclicBarrier
都是用于控制并发的工具类,都可以理解成维护的就是一个计数器
,但是这两者还是各有不同侧重点的:
-
CountDownLatch
一般用于某个线程A等待若干个其他线程执行完任务之后,它才执行;而CyclicBarrier一般用于一组线程互相等待至某个状态,然后这一组线程再同时执行;CountDownLatch
强调一个线程等多个线程完成某件事情
。CyclicBarrier
是多个线程互等,等大家都完成,再携手共进。 -
调用
CountDownLatch
的countDown()
方法后,当前线程并不会阻塞
,会继续往下执行;而调用CyclicBarrier
的await()
方法,会阻塞当前线程,直到CyclicBarrier
指定的线程全部都到达了指定点的时候,才能继续往下执行; -
CountDownLatch
方法比较少,操作比较简单,而CyclicBarrier
提供的方法更多,比如能够通过getNumberWaiting()
,isBroken()
这些方法获取当前多个线程的状态,并且CyclicBarrier
的构造方法可以传入barrierAction
,指定当所有线程都到达时执行的业务功能; -
CountDownLatch
是不能复用的,而CyclicLatch
是可以复用的。 -
和
CountdownLatch
相似,都是通过维护计数器来实现的。线程执行 await() 方法之后计数器会减 1,并进行等待,直到计数器为 0,所有调用 await() 方法而在等待的线程才能继续执行。 -
CyclicBarrier
和CountdownLatch
的一个区别是,CyclicBarrier
的计数器通过调用reset()
方法可以循环使用,所以它才叫做循环屏障。
CyclicBarrier示例程序
package com.bestksl.example.aqs;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample3 {
static final int clientTotal = 500;
static final int threadTotal = 10;
static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> { // 可以选择传入一个Runnable 对象 完成更复杂的逻辑
log.info("runnable");
});
public static void main(String[] args) throws InterruptedException {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int n = i;
Thread.sleep(1000);
executorService.execute(() -> {
try {
p(n);
} catch (InterruptedException e) {
log.info(e.toString());
}
});
}
executorService.shutdown();
}
private static void p(int n) throws InterruptedException {
log.info(n + "ready..");
try {
cyclicBarrier.await(20000, TimeUnit.MILLISECONDS); // 如果超出等待时间(可选参数), 则会抛出BrokenBarrierException & TimeoutException
} catch (BrokenBarrierException | TimeoutException e) {
log.info("warn!! {}", e.toString());
}
log.info(n + "resume");
}
}
执行结果
/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/bin/java "-javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=56252:/Applications/IntelliJ IDEA.app/Contents/bin" -Dfile.encoding=UTF-8 -classpath /Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/cldrdata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/jaccess.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/nashorn.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/ext/zipfs.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jce.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jfxswt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/packager.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_241.jdk/Contents/Home/lib/tools.jar:/Users/haoxuanli/Documents/GitHub/Concurrency/target/classes:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter-web/2.2.5.RELEASE/spring-boot-starter-web-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter/2.2.5.RELEASE/spring-boot-starter-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot/2.2.5.RELEASE/spring-boot-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-autoconfigure/2.2.5.RELEASE/spring-boot-autoconfigure-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter-logging/2.2.5.RELEASE/spring-boot-starter-logging-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/ch/qos/logback/logback-classic/1.2.3/logback-classic-1.2.3.jar:/Users/haoxuanli/.m2/repository/ch/qos/logback/logback-core/1.2.3/logback-core-1.2.3.jar:/Users/haoxuanli/.m2/repository/org/slf4j/slf4j-api/1.7.25/slf4j-api-1.7.25.jar:/Users/haoxuanli/.m2/repository/org/apache/logging/log4j/log4j-to-slf4j/2.12.1/log4j-to-slf4j-2.12.1.jar:/Users/haoxuanli/.m2/repository/org/apache/logging/log4j/log4j-api/2.12.1/log4j-api-2.12.1.jar:/Users/haoxuanli/.m2/repository/org/slf4j/jul-to-slf4j/1.7.30/jul-to-slf4j-1.7.30.jar:/Users/haoxuanli/.m2/repository/jakarta/annotation/jakarta.annotation-api/1.3.5/jakarta.annotation-api-1.3.5.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-core/5.2.4.RELEASE/spring-core-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-jcl/5.2.4.RELEASE/spring-jcl-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/yaml/snakeyaml/1.25/snakeyaml-1.25.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter-json/2.2.5.RELEASE/spring-boot-starter-json-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/core/jackson-databind/2.10.2/jackson-databind-2.10.2.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/core/jackson-annotations/2.10.2/jackson-annotations-2.10.2.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/core/jackson-core/2.10.2/jackson-core-2.10.2.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/datatype/jackson-datatype-jdk8/2.10.2/jackson-datatype-jdk8-2.10.2.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/datatype/jackson-datatype-jsr310/2.10.2/jackson-datatype-jsr310-2.10.2.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/jackson/module/jackson-module-parameter-names/2.10.2/jackson-module-parameter-names-2.10.2.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter-tomcat/2.2.5.RELEASE/spring-boot-starter-tomcat-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/apache/tomcat/embed/tomcat-embed-core/9.0.31/tomcat-embed-core-9.0.31.jar:/Users/haoxuanli/.m2/repository/org/apache/tomcat/embed/tomcat-embed-el/9.0.31/tomcat-embed-el-9.0.31.jar:/Users/haoxuanli/.m2/repository/org/apache/tomcat/embed/tomcat-embed-websocket/9.0.31/tomcat-embed-websocket-9.0.31.jar:/Users/haoxuanli/.m2/repository/org/springframework/boot/spring-boot-starter-validation/2.2.5.RELEASE/spring-boot-starter-validation-2.2.5.RELEASE.jar:/Users/haoxuanli/.m2/repository/jakarta/validation/jakarta.validation-api/2.0.2/jakarta.validation-api-2.0.2.jar:/Users/haoxuanli/.m2/repository/org/hibernate/validator/hibernate-validator/6.0.18.Final/hibernate-validator-6.0.18.Final.jar:/Users/haoxuanli/.m2/repository/org/jboss/logging/jboss-logging/3.3.2.Final/jboss-logging-3.3.2.Final.jar:/Users/haoxuanli/.m2/repository/com/fasterxml/classmate/1.3.4/classmate-1.3.4.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-web/5.2.4.RELEASE/spring-web-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-beans/5.2.4.RELEASE/spring-beans-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-webmvc/5.2.4.RELEASE/spring-webmvc-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-aop/5.2.4.RELEASE/spring-aop-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-context/5.2.4.RELEASE/spring-context-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/springframework/spring-expression/5.2.4.RELEASE/spring-expression-5.2.4.RELEASE.jar:/Users/haoxuanli/.m2/repository/org/projectlombok/lombok/1.18.0/lombok-1.18.0.jar:/Users/haoxuanli/.m2/repository/joda-time/joda-time/2.10.5/joda-time-2.10.5.jar com.bestksl.example.aqs.CyclicBarrierExample3
20:28:26.711 [pool-1-thread-1] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 0ready..
20:28:27.715 [pool-1-thread-2] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 1ready..
20:28:28.716 [pool-1-thread-3] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 2ready..
20:28:29.719 [pool-1-thread-4] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 3ready..
20:28:30.724 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 4ready..
20:28:30.724 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - runnable
20:28:30.724 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 4resume
20:28:30.725 [pool-1-thread-3] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 2resume
20:28:30.725 [pool-1-thread-4] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 3resume
20:28:30.725 [pool-1-thread-1] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 0resume
20:28:30.725 [pool-1-thread-2] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 1resume
20:28:31.725 [pool-1-thread-2] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 5ready..
20:28:32.726 [pool-1-thread-1] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 6ready..
20:28:33.729 [pool-1-thread-4] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 7ready..
20:28:34.734 [pool-1-thread-3] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 8ready..
20:28:35.738 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 9ready..
20:28:35.738 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - runnable
20:28:35.738 [pool-1-thread-5] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 9resume
20:28:35.738 [pool-1-thread-2] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 5resume
20:28:35.738 [pool-1-thread-1] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 6resume
20:28:35.738 [pool-1-thread-4] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 7resume
20:28:35.738 [pool-1-thread-3] INFO com.bestksl.example.aqs.CyclicBarrierExample3 - 8resume
Process finished with exit code 0