多线程 - CyclicBarrier

小咪咪 2024-03-22 13:31 49阅读 0赞

CyclicBarrier是Java中的一个同步工具类，它允许一组线程在某个点上互相等待，直到所有线程都到达该点，然后再继续执行。  
它的工作方式类似于CountDownLatch，但它可以重复使用。

具体来说，当CyclicBarrier被创建时，需要指定一个参与同步的线程数量以及一个当所有线程都到达该点时要执行的任务。  
在执行过程中，每个线程在到达该点时会调用await()方法来等待其他线程的到达。  
当所有线程都到达时，CyclicBarrier会自动解除等待状态，并执行指定的任务。  
此时，所有线程都可以继续执行下一步操作，而CyclicBarrier也会重置状态，以便于下一轮的同步。

CyclicBarrier的一个常见应用场景是分治算法，其中多个线程分别执行某个子任务，然后再通过CyclicBarrier同步结果，以便于继续执行下一轮的子任务。  
另外，CyclicBarrier还可以用于实现类似于多阶段计算的任务，其中每个阶段需要等待所有线程都完成前一阶段的计算结果，然后再继续进行下一阶段的计算。

三者对比

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>CyclicBarrier</th> 
   <th>Semaphore</th> 
   <th>CountDownLatch</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>功能</td> 
   <td>一组线程互相等待，直到所有线程都到达某个点之后再同时继续执行。</td> 
   <td>控制同时访问某个资源的线程数量。</td> 
   <td>让一个或多个线程等待其他线程完成操作后再执行。</td> 
  </tr> 
  <tr> 
   <td>使用场景</td> 
   <td>任务分解成多个子任务，子任务完成后再进行下一步操作。</td> 
   <td>线程间的互斥或者协作控制，如连接池、文件池等。</td> 
   <td>等待某个事件完成或者等待某些资源准备好后再进行操作。</td> 
  </tr> 
  <tr> 
   <td>构造方法</td> 
   <td><code>CyclicBarrier(int parties, Runnable barrierAction)</code></td> 
   <td><code>Semaphore(int permits)</code></td> 
   <td><code>CountDownLatch(int count)</code></td> 
  </tr> 
  <tr> 
   <td>核心方法</td> 
   <td><code>await()</code></td> 
   <td><code>acquire()</code>和<code>release()</code></td> 
   <td><code>await()</code>和<code>countDown()</code></td> 
  </tr> 
  <tr> 
   <td>计数器</td> 
   <td>有，每个线程到达屏障时计数器减1，直到计数器为0时，屏障打开，线程继续执行。</td> 
   <td>有，表示可用的许可数量。线程获取许可时许可数量减1，释放许可时许可数量加1。</td> 
   <td>有，初始计数器为给定的值，每个线程完成任务时计数器减1，直到计数器为0时，等待的线程继续执行。</td> 
  </tr> 
  <tr> 
   <td>重复使用</td> 
   <td>可以，每次到达屏障时计数器会重置。</td> 
   <td>可以，<code>acquire()</code>方法可以被重复调用，每次调用会消耗一个许可。</td> 
   <td>不可以，一旦计数器为0就无法再次使用。</td> 
  </tr> 
 </tbody> 
</table>

## 简单使用CyclicBarrier ##

package com.lfsun.highconcurrency000.juc.cyclicbarrier;
    
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * 简单使用CyclicBarrier
     */
    public class MySimpleCyclicBarrierExample {
        
        public static void main(String[] args) throws Exception {
        
            // 创建一个CyclicBarrier，需要指定参与同步的线程数和要执行的任务
            CyclicBarrier barrier = new CyclicBarrier(3, () -> {
        
                System.out.println("所有线程已到达屏障，开始执行任务...");
            });
    
            // 创建3个线程并启动
            for (int i = 0; i < 3; i++) {
        
                new Thread(() -> {
        
                    try {
        
                        System.out.println(Thread.currentThread().getName() + " 到达屏障...");
                        // 调用await()方法等待其他线程到达
                        barrier.await();
                        System.out.println(Thread.currentThread().getName() + " 执行任务...");
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }).start();
            }
        }
    }

## 循环使用CyclicBarrier ##

import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * 循环使用CyclicBarrier
     */
    public class MyLoopCyclicBarrierExample {
        
        public static void main(String[] args) {
        
            int THREAD_NUM = 3; // 线程数
            int LOOP_NUM = 3; // 循环次数
    
            CyclicBarrier barrier = new CyclicBarrier(THREAD_NUM, () -> {
        
                System.out.println("所有线程已到达屏障，开始执行任务...");
            });
    
            for (int i = 0; i < THREAD_NUM; i++) {
        
                new Thread(() -> {
        
                    for (int j = 0; j < LOOP_NUM; j++) {
        
                        try {
        
                            System.out.println(Thread.currentThread().getName() + " 到达屏障...");
                            barrier.await();
                            System.out.println(Thread.currentThread().getName() + " 执行任务...");
                        } catch (InterruptedException | BrokenBarrierException e) {
        
                            e.printStackTrace();
                        }
                    }
                }).start();
            }
        }
    }