1.CyclicBarrier
CyclicBarrier初始化时规定一个数目,然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时,所有进入等待状态的线程被唤醒并继续。
功能:可以使一定数量的参与方反复在指定的地方(就是调用await()的地方)汇集(阻塞自己) ,只有所有参与方都调用了await(),栅栏就会打开,所有线程阻塞才会解除
适用场景:CyclicBarrier可以用在所有子线程之间互相等待多次的情形。如并行迭代算法,这种算法通常将一个问题拆分成一系列相互独立的子问题
code
public static void main(String[] args){ //CyclicBarrier 是几个线程导了一个点后等其他线程到了一起出发 哪个地方等就在哪个地方调用await()
ExecutorService threadpool=Executors.newCachedThreadPool();
final CyclicBarrier cb=new CyclicBarrier(3);
for(int i=0;i<3;i++){
Runnable runnable=new Runnable() {
@Override
public void run() {
try {
Thread.sleep((long)Math.random()*10000);
System.out.println("线程" + Thread.currentThread().getName() + "即将到达地点1,当前已有"+(cb.getNumberWaiting()+1 )+"个线程");
//到达出发点
cb.await();
Thread.sleep((long)Math.random()*10000);
System.out.println("线程"+Thread.currentThread().getName()+"即将到达集合地点2,当前已有"+(cb.getNumberWaiting()+1)+"个线程");
cb.await();
Thread.sleep((long)Math.random()*10000);
System.out.println("线程"+Thread.currentThread().getName()+"即将到达集合地点3,当前已有"+(cb.getNumberWaiting()+1)+"个线程");
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
threadpool.execute(runnable);
}
}
2.CountDownLatch(闭锁)
CountDownLatch先初始化规定一个计数器,没调用一次countDwon()计数器减一,如果数量变成0就达到结束状态,闭锁打开(await()方法阻塞解除)
闭锁的作用相当于一扇门,闭锁的状态没有结束时这扇门一直视关闭的,闭锁达到结束状态时将允许所有线程通过
适用场景:CountDownLatch可以应用于主线程等待所有子线程结束后再继续执行的情况。闭锁可以用来确定某些活动直到其他活动都完成了才继续执行
如确保某个服务在其他服务启动之后才能启动;在多玩家的游戏中确保所有玩家就绪后才执行
Code
public static void main(String[] args){
ExecutorService threadpool = Executors.newCachedThreadPool();
final CountDownLatch cdorder=new CountDownLatch(1);
final CountDownLatch cdanswer=new CountDownLatch(3);
for(int i=0;i<3;i++){
Runnable runnable=new Runnable() {
@Override
public void run() {
try {
System.out.println("线程"+Thread.currentThread().getName()+"正准备接受命令");
cdorder.await();
System.out.println("线程"+Thread.currentThread().getName()+"已接受命令");
Thread.sleep((long)Math.random()*10000);
System.out.println("线程"+Thread.currentThread().getName()+"回应命令处理结果");
cdanswer.countDown();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
};
threadpool.execute(runnable);
}
try {
Thread.sleep((long)Math.random()*10000);
System.out.println("线程"+Thread.currentThread().getName()+"即将发布命令");
cdorder.countDown();
System.out.println("线程"+Thread.currentThread().getName()+"已发送命令,正在等待结果");
cdanswer.await();
System.out.println("线程"+Thread.currentThread().getName()+"已收到所有响应结果");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
threadpool.shutdown();
}
CountDownLatch与CyclicBarrier区别
1.CountDownLatch减计数方式
CyclicBarrier 加计数方式
2.CountDownLatch计算为0时释放所有等待的线程
CyclicBarrier 计数达到指定值时释放所有等待线程
3.CountDownLatch计数为0时,无法重置
CyclicBarrier 计数达到指定值时,计数置为0重新开始
4.CountDownLatch调用countDown()方法计数减一,调用await()方法只进行阻塞,对计数没任何影响
CyclicBarrier 调用await()方法计数加1,若加1 的值不等于构造方法的值,则线程阻塞
5.CountDownLatch不可重复利用
CyclicBarrier可重复利用
6.线程在countDown()之后,会继续执行自己的任务,而CyclicBarrier会在所有线程任务结束之后,才会进行后续任务
3.Semaphore(信号量)
计数信号量用来控制同时访问某个特定资源的操作数量,信号量维护了一个许可集。如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个 release() 添加一个许可,从而可能释放一个正在阻塞的获取者。但是,不使用实际的许可对象,Semaphore 只对可用许可的号码进行计数,并采取相应的行动。拿到信号量的线程可以进入代码,否则就等待。通过acquire()和release()获取和释放访问许可。初始值为1的Semaphore二值信号量可以用作互斥体
适用场景:Semaphore可以用于做流量控制,特别公用资源有限的应用场景,比如数据库连接。假如有一个需求,要读取几万个文件的数据,因为都是IO密集型任务,我们可以启动几十个线程并发的读取,但是如果读到内存后,还需要存储到数据库中,而数据库的连接数只有10个,这时我们必须控制只有十个线程同时获取数据库连接保存数据,否则会报错无法获取数据库连接。这个时候,我们就可以使用Semaphore来做流控
Code
public static void main(String[] args){
ExecutorService threadpool = Executors.newCachedThreadPool();
final Semaphore sp=new Semaphore(3);
for(int i=0;i<10;i++){
Runnable runnable=new Runnable() {
@Override
public void run() {
try {
sp.acquire();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("线程" + Thread.currentThread().getName() +" 进入,当前有" + (3- sp.availablePermits())+ " 个线程并发");
try {
Thread.sleep((long)(Math.random()*10000));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程"+Thread.currentThread().getName()+"即将离开");
sp.release();
System.out.println("线程"+Thread.currentThread().getName()+"已离开,当前已有"+(3-sp.availablePermits())+"个线程并发");
}
};
threadpool.execute(runnable);
}
}
控制任务提交速度
class BoundedExecutor{//使用Semaphore控制任务的提交速率
private final Executor exec;
private final Semaphore semaphore;
public BoundedExecutor(Executor exec,Semaphore semaphore){
this.exec=exec;
this.semaphore=semaphore;
}
public void submitTask(final Runnable command) throws InterruptedException{
semaphore.acquire();
try{
exec.execute(new Runnable() {
@Override
public void run() {
try{
command.run();
}finally{
semaphore.release();
}
}
});
}catch(RejectedExecutionException e){
semaphore.release();
}
}
}