CountDownLunch(闭锁)、CyclicBarrier(栅栏锁)、Semaphore(信号量)的区别

BUG弄潮儿

发布于 2024-06-08 08:42:50

710

发布于 2024-06-08 08:42:50

文章被收录于专栏：JAVA乐园JAVA乐园

CountDownLunch

countDownLunch，又叫闭锁。它有三个关键的api：

new CountDownLatch(count); 创建一个闭锁，并声明count的值
countDownLatch.await();如果countDownLunch的count不是0，则阻塞当前线程直到count等0
countDownLatch.countDown();将countDownLunch中的count减一

代码样例:

Logger logger = LoggerFactory.getLogger(this.getClass());
//创建一个count=1的闭锁
CountDownLatch countDownLatch = new CountDownLatch(1);
List<Thread> threads = new ArrayList<>();
//创建5个线程
for (int i = 0; i < 5; i++) {
 Thread thread = new Thread(() -> {
  logger.info("[{}]在等待发令枪", Thread.currentThread().getName());
  try {
   //等待闭锁的count=0
   countDownLatch.await();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
  logger.info("枪响了,[{}]跑!", Thread.currentThread().getName());
 }, "t" + (i + 1));
 thread.start();
 threads.add(thread);
 try {
  //让出CPU
  TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}
logger.info("开枪，开跑！");
//将count--
countDownLatch.countDown();
try {
 //让出CPU
 TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
 e.printStackTrace();
}
//循环等待所有线程结束
for (Thread thread : threads) {
 try {
  thread.join();
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}

运行结果：

从结果中可以看出来，t1~t5是同时开跑的。需要注意的是，countDownLatch.await();会在count的值等于0时，唤醒被阻塞的线程，但是被唤醒的线程是否马上就可执行，这个要看CPU的调度，不一定被唤醒后，马上就可以执行。

上面是多等一的用法，下面来一个一等多的用法：

public static void main(String[] args) {
 int count = 5;
 CountDownLatch countDownLatch = new CountDownLatch(count);
 String[] list = new String[count];
 Random random = new Random();
 for (int i = 0; i < count; i++) {
     int finalI = i;
     Thread thread = new Thread(() -> {
         for (int j = 0; j <= 100; j++) {
             try {
                 TimeUnit.MILLISECONDS.sleep(random.nextInt(200));
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
             list[finalI]= Thread.currentThread().getName()+"("+j+"%)";
             System.out.print("\r"+ Arrays.toString(list));
         }
         countDownLatch.countDown();
     }, "t" + (i + 1));
     thread.start();
 }
 try {
     countDownLatch.await();
 } catch (InterruptedException e) {
     e.printStackTrace();
 }
 System.out.print("\n结束了");
}

结果如下：

与join相比相同点：

都可以在某种程度上等待线程执行完毕

与join相比不同点：

join是Thread的方法，需要持有Thread的引用，但是现在很多时候都是像线程池中提交任务的，很难拿到这个Thread的引用。但是CountDownLunch是可以作为全局变量的，这个引用比较好拿到。
join是一定要等待线程结束的，但是CountDownLunch还是比较灵活的，可以在任意时刻countDown。

CyclicBarrier

CyclicBarrier，又叫栅栏锁

Logger logger = LoggerFactory.getLogger(this.getClass());
CyclicBarrier cyclicBarrier = new CyclicBarrier(2, ()->{
 logger.info("cyclicBarrier被置为0了,{}",Thread.currentThread().getName());
});
logger.info("cyclicBarrier初始化为2,{}",Thread.currentThread().getName());
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < 4; i++) {
 Thread thread = new Thread(() -> {
  logger.info("cyclicBarrier -1,[{}]",Thread.currentThread().getName());
  try {
   cyclicBarrier.await();
  } catch (InterruptedException | BrokenBarrierException e) {
   e.printStackTrace();
  }
  logger.info("cyclicBarrier 0了,[{}]",Thread.currentThread().getName());
 }, "t" + (i + 1));
 thread.start();
 threads.add(thread);
 try {
  TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}
//循环等待所有线程结束
for (Thread thread : threads) {
 try {
  thread.join();
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}

执行结果：

从执行结果可以看出来，cyclicBarrier是可以循环使用的，当cyclicBarrier的值=0时，会调用CyclicBarrier构造器中的runnable同时会将count重置为一开始设定的值。与countDownLunch相比：

相同点：

在count不等于0时，调用await的线程也是会阻塞的。

不同点：

cyclicBarrier可以循环使用，countDownLunch是一次性的
cyclicBarrier只要调用await就会使count-1，但是countDownLunch需要手动调用countDown方法

Semaphore

Semaphore，又叫信号量

信号量是用来限制一个时间点下，使用某资源的最大线程数。信号量限制的是线程数，而不是资源数。

Semaphore semaphore = new Semaphore(3);
Logger logger = LoggerFactory.getLogger(this.getClass());
logger.info("初始化了一个大小为3的信号量。");
List<Thread> threads = new ArrayList<>();
for (int i = 0; i < 5; i++) {
 Thread thread = new Thread(() -> {
  logger.info("[{}]尝试申请一个资源",Thread.currentThread().getName());
  try {
   semaphore.acquire();
   logger.info("[{}]申请到了资源",Thread.currentThread().getName());
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
  try {
   TimeUnit.SECONDS.sleep(5);
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
  logger.info("[{}]释放了一个资源",Thread.currentThread().getName());
  semaphore.release();
 }, "t" + (i + 1));
 thread.start();
 threads.add(thread);
 try {
  //让出CPU
  TimeUnit.SECONDS.sleep(1);
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}
//循环等待所有线程结束
for (Thread thread : threads) {
 try {
  thread.join();
 } catch (InterruptedException e) {
  e.printStackTrace();
 }
}

执行结果：