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();
}
}
}