Java并发之CountDownLatch、CyclicBarrier和Semaphore

时间:2022-11-05 20:49:26

这次说一下 JUC 中的同步器三个主要的成员:CountDownLatch、CyclicBarrier 和Semaphore。这三个是 JUC 中较为常用的同步器,通过它们可以方便地实现很多线程之间协作的功能。

  • CountDownLatch

CountDownLatch一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。
用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法,所以在当前计数到达零之前,await 方法会一直受阻塞。之后,会释放所有等待的线程,await
的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数,请考虑使用 CyclicBarrier。
CountDownLatch 是一个通用同步工具,它有很多用途。将计数 1 初始化的 CountDownLatch
用作一个简单的开/关锁存器,或入口:在通过调用 countDown() 的线程打开入口前,所有调用 await 的线程都一直在入口处等待。用
N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待,或者使其在某项操作完成 N
次之前一直等待。 CountDownLatch 的一个有用特性是,它不要求调用 countDown
方法的线程等到计数到达零时才继续,而在所有线程都能通过之前,它只是阻止任何线程继续通过一个 await。

示例

package chapter3;  

import java.util.concurrent.CountDownLatch;

public class Conference implements Runnable{

private final CountDownLatch controller;

public Conference(int number){

controller = new CountDownLatch(number);
}

public void sign(String name){

System.out.println("Waiting for "+controller.getCount());
System.out.println(name+" has arrived.");
//递减锁存器的计数,如果计数到达零,则释放所有等待的线程。
controller.countDown();
}

@Override
public void run() {

System.out.println("共:"+controller.getCount());
try {
//使当前线程在锁存器倒计数至零之前一直等待,除非线程被中断
controller.await();
System.out.printf("all arrived,we can starting...\n");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
package chapter3;  

import java.util.concurrent.TimeUnit;

public class SignIn implements Runnable{

private String name;
private Conference conference;

public SignIn(Conference conference,String name){

this.conference = conference;
this.name = name;
}

@Override
public void run() {

long duration = (long)(Math.random()*10);
try {
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
conference.sign(name);
}
}
package chapter3;  

public class Test {

public static void main(String[] args) {

Conference conference = new Conference(10);
Thread conferenceThread = new Thread(conference);
conferenceThread.start();
for(int i=0;i<10;i++){
SignIn signIn = new SignIn(conference, "user"+i);
Thread thread = new Thread(signIn);
thread.start();
}
}
}
  • CyclicBarrier

CyclicBarrier一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点 (common barrier
point)。在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待,此时 CyclicBarrier 很有用。因为该 barrier
在释放等待线程后可以重用,所以称它为循环 的 barrier。
CyclicBarrier 支持一个可选的 Runnable 命令,在一组线程中的最后一个线程到达之后(但在释放所有线程之前),该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态,此屏障操作
很有用。

CyclicBarrier 它主要的方法就是一个:await()。await()
方法没被调用一次,计数便会减少1,并阻塞住当前线程。当计数减至0时,阻塞解除,所有在此
CyclicBarrier上面阻塞的线程开始运行。在这之后,如果再次调用 await() 方法,计数就又会变成
N-1,新一轮重新开始,这便是 Cyclic的含义所在。
CyclicBarrier.await() 方法带有返回值,用来表示当前线程是第几个到达这个 Barrier 的线程。 和 CountDownLatch 一样,CyclicBarrier 同样可以在构造函数中设定总计数值。与 CountDownLatch
不同的是,CyclicBarrier 的构造函数还可以接受一个 Runnable,会在 CyclicBarrier 被释放时执行。

示例

import java.util.concurrent.CyclicBarrier;

public class TestCyclicBarrier {

private static final int THREAD_NUM = 5;

public static class WorkerThread implements Runnable{

CyclicBarrier barrier;

public WorkerThread(CyclicBarrier b){
this.barrier = b;
}

@Override
public void run() {
// TODO Auto-generated method stub
try{
System.out.println("Worker's waiting");
//线程在这里等待,直到所有线程都到达barrier。
barrier.await();
System.out.println("ID:"+Thread.currentThread().getId()+" Working");
}catch(Exception e){
e.printStackTrace();
}
}

}

/**
* @param args
*/

public static void main(String[] args) {
// TODO Auto-generated method stub
CyclicBarrier cb = new CyclicBarrier(THREAD_NUM, new Runnable() {
//当所有线程到达barrier时执行
@Override
public void run() {
// TODO Auto-generated method stub
System.out.println("Inside Barrier");
}
});

for(int i=0;i<THREAD_NUM;i++){
new Thread(new WorkerThread(cb)).start();
}
}
}

CyclicBarrier 和 CountDownLatch 在用法上的不同: CountDownLatch
适用于一组线程和另一个主线程之间的工作协作。一个主线程等待一组工作线程的任务完毕才继续它的执行是使用 CountDownLatch
的主要场景;CyclicBarrier
用于一组或几组线程,比如一组线程需要在一个时间点上达成一致,例如同时开始一个工作。另外,CyclicBarrier的循环特性和构造函数所接受的
Runnable 参数也是 CountDownLatch 所不具备的。

  • Semaphore

一个计数信号量。从概念上讲,信号量维护了一个许可集。如有必要,在许可可用前会阻塞每一个 acquire(),然后再获取该许可。每个
release() 添加一个许可,从而可能释放一个正在阻塞的获取者。但是,不使用实际的许可对象,Semaphore
只对可用许可的号码进行计数,并采取相应的行动。 Semaphore 通常用于限制可以访问某些资源(物理或逻辑的)的线程数目。

示例:

package chapter2;
import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Semaphore;

public class TestSemaphore {

public static void main(String[] args) {

ExecutorService exec = Executors.newCachedThreadPool();
// 只能5个线程同时访问
final Semaphore semaphore = new Semaphore(5);
// 模拟20个客户端访问
for (int index = 0; index < 20; index++) {
final int num = index;
Runnable run = new Runnable() {
public void run() {
try {
// 获取许可
semaphore.acquire();
System.out.println(num+" start...");
Thread.sleep((long) (Math.random() * 10000));
System.out.println(num+" end...");
// 访问完后,释放
semaphore.release();
System.out.println("当前可用:"+semaphore.availablePermits());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
exec.execute(run);
}
// 退出线程池
exec.shutdown();
}
}

CountDownLatch 是能使一组线程等另一组线程都跑完了再继续跑;CyclicBarrier能够使一组线程在一个时间点上达到同步,可以是一起开始执行全部任务或者一部分任务。同时,它是可以循环使用的;Semaphore是只允许一定数量的线程同时执行一段任务。