并发工具类之（CountDownLatch，CyclicBarr
2019-12-05 本文已影响0人
laod_wh
CountDownLatch
可以理解为线程的计数器，使一个线程等待一组线程执行完以后在执行，初始值为线程数的值，当一个线程执行完以后，计数器的值减1，当计数器为0时，表示计数器上所有线程数执行完成，开始执行后续的任务。且不支持计数器重置，当计数器为0后，当其他线程调await时，会直接通过。
先写个简单的例子：
@Override
public void run(ApplicationArguments args) throws Exception{
//创建固定线程数，来测试计数器的使用
ExecutorService executorService = Executors.newFixedThreadPool(5);
//初始化计数器
CountDownLatch countDownLatch = new CountDownLatch(5);
for (int i = 0; i < 5; i++) {
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
log.info("runnable start--> Threadname:{}",Thread.currentThread().getName());
Thread.sleep(1000);
log.info("runnable end--> Threadname:{}",Thread.currentThread().getName());
countDownLatch.countDown();//线程执行完减1
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
executorService.execute(runnable);
}
log.info("main --> 等待子线程执行完！！！");
//获取计数器值
log.info("current count -->{}",countDownLatch.getCount());
//阻塞，等待计数器执行完
//如果设置了超时时间，到时间会直接走下面的线程
//countDownLatch.await(100, TimeUnit.MILLISECONDS);
countDownLatch.await();
log.info("main 开始执行。。。");
log.info("current count -->{}",countDownLatch.getCount());
}
执行结果：
2019-12-05 20:14:31.302 INFO 1160 --- [pool-1-thread-5] com.laod.MemcachedApplication : runnable start--> Threadname:pool-1-thread-5
2019-12-05 20:14:31.301 INFO 1160 --- [pool-1-thread-3] com.laod.MemcachedApplication : runnable start--> Threadname:pool-1-thread-3
2019-12-05 20:14:31.301 INFO 1160 --- [pool-1-thread-4] com.laod.MemcachedApplication : runnable start--> Threadname:pool-1-thread-4
2019-12-05 20:14:31.302 INFO 1160 --- [ main] com.laod.MemcachedApplication : main --> 等待子线程执行完！！！
2019-12-05 20:14:31.300 INFO 1160 --- [pool-1-thread-1] com.laod.MemcachedApplication : runnable start--> Threadname:pool-1-thread-1
2019-12-05 20:14:31.300 INFO 1160 --- [pool-1-thread-2] com.laod.MemcachedApplication : runnable start--> Threadname:pool-1-thread-2
2019-12-05 20:14:31.302 INFO 1160 --- [ main] com.laod.MemcachedApplication : current count -->5
2019-12-05 20:14:32.306 INFO 1160 --- [pool-1-thread-5] com.laod.MemcachedApplication : runnable end--> Threadname:pool-1-thread-5
2019-12-05 20:14:32.306 INFO 1160 --- [pool-1-thread-2] com.laod.MemcachedApplication : runnable end--> Threadname:pool-1-thread-2
2019-12-05 20:14:32.306 INFO 1160 --- [pool-1-thread-4] com.laod.MemcachedApplication : runnable end--> Threadname:pool-1-thread-4
2019-12-05 20:14:32.306 INFO 1160 --- [pool-1-thread-1] com.laod.MemcachedApplication : runnable end--> Threadname:pool-1-thread-1
2019-12-05 20:14:32.306 INFO 1160 --- [pool-1-thread-3] com.laod.MemcachedApplication : runnable end--> Threadname:pool-1-thread-3
2019-12-05 20:14:32.307 INFO 1160 --- [ main] com.laod.MemcachedApplication : main 开始执行。。。
2019-12-05 20:14:32.307 INFO 1160 --- [ main] com.laod.MemcachedApplication : current count -->0
CyclicBarrier
上面的CountDownLatch，如果业务需两个线程步调一致，然后在通知其他线程，虽然CountDownLatch也可以做，定义计数器2，每个线程执行完后都减1，如果计数器大于0，就线程1或者线程2等待，如果计数器等于0，就唤醒线程1，线程2，在执行其他线程。
虽然这样也可以实现，两线程步调一致，但是较繁琐，Java并发包内有现成的可以使用：CyclicBarrier，与上面很打一个区别，这个有自动重置，回调方法的功能，当减到 0 的时候，会自动重置你设置的初始值，循环执行 直接上例子：
@Override
public void run(ApplicationArguments args) throws Exception{
// 业务1
Vector<String> pos = new Vector<String>();
// 业务2
Vector<String> dos = new Vector<String>();
//创建固定线程数，来测试计数器的使用
ExecutorService executor = Executors.newFixedThreadPool(2);
//从字面意思就能理解 可循环利用
CyclicBarrier barrier = new CyclicBarrier(2,() ->{
executor.execute(()->check(pos,dos));
});
//业务1
Thread T1 = new Thread(()->{
while (1==1){
//处理具体的业务
try {
log.info("模拟业务1一直有数据");
pos.add("1");
barrier.await();
log.info("numberwaiting -->{}",barrier.getNumberWaiting());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
T1.start();
//业务2
Thread T2 = new Thread(()->{
while (1==1){
//处理具体的业务
try {
log.info("模拟业务2一直有数据");
dos.add("2");
barrier.await();
log.info("numberwaiting -->{}",barrier.getNumberWaiting());
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
T2.start();
log.debug("-----");
}
private void check(Vector<String> pos, Vector<String> dos) {
pos.remove(0);
dos.remove(0);
log.info("--- 处理业务场景 步调一致");
}
执行结果如下，可以很清楚看到两线程业务处理完成后，在做其他处理：
2019-12-06 09:28:02.067 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:02.067 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:02.068 INFO 1698 --- [pool-1-thread-1] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:02.068 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:02.068 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:02.071 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:03.071 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:03.072 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:03.072 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:03.074 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:03.075 INFO 1698 --- [pool-1-thread-2] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:04.077 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:04.077 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:04.077 INFO 1698 --- [pool-1-thread-1] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:04.077 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:04.079 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:05.083 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:05.084 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:05.084 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:05.084 INFO 1698 --- [pool-1-thread-2] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:05.085 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:06.089 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:06.089 INFO 1698 --- [pool-1-thread-1] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:06.089 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:06.089 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:06.091 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:07.091 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:07.091 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:07.091 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:07.091 INFO 1698 --- [pool-1-thread-2] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:07.092 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
2019-12-06 09:28:08.096 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : 模拟业务1一直有数据
2019-12-06 09:28:08.096 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:08.096 INFO 1698 --- [ Thread-9] com.laod.MemcachedApplication : numberwaiting -->0
2019-12-06 09:28:08.096 INFO 1698 --- [pool-1-thread-1] com.laod.MemcachedApplication : --- 处理业务场景 步调一致
2019-12-06 09:28:08.098 INFO 1698 --- [ Thread-10] com.laod.MemcachedApplication : 模拟业务2一直有数据
总结：
CountDownLatch:
业务场景：一个线程等待多个线程的执行场景，不能循环利用，一旦计数器为0后，且有线程调用await()，会直接通过。
CyclicBarrier:
业务场景：一组线程之间相互等待，可循环利用，计数器为0后，会自动重置，而且还支持回调函数。
完
