Java中CyclicBarrier使用指南

梦里梦外; 2022-10-12 01:20 27阅读 0赞

### 文章目录 ###

*  1.介绍
 *  2.Java 并发——同步器
 *  3. CyclicBarrier
 *  4.使用
 *  5.实现
 *  6.结果

# 1.介绍 #

CyclicBarriers 是在 Java 5 中作为 java.util.concurrent 包的一部分引入的同步结构。

# 2.Java 并发——同步器 #

java.util.concurrent 包包含几个类，这些类有助于管理一组相互协作的线程。 其中一些包括：

*  *CyclicBarrier*
 *  *Phaser*
 *  *CountDownLatch*
 *  *Exchanger*
 *  *Semaphore*
 *  *SynchronousQueue*

这些类为线程之间的常见交互模式提供了开箱即用的功能。

如果有一组相互通信的线程并且类似于其中一个常见模式，可以简单地重用适当的库类（也称为同步器），而不是尝试使用一组锁和条件来提出自定义方案 对象和同步关键字。

# 3. CyclicBarrier #

CyclicBarrier 是一个同步器，它允许一组线程相互等待到达一个共同的执行点，也称为屏障。

> CyclicBarriers 用于程序中，其中我们有固定数量的线程，这些线程在继续执行之前必须等待彼此到达公共点。

屏障被称为循环的，因为它可以在等待线程被释放后重新使用。

# 4.使用 #

CyclicBarrier 的构造函数很简单。 它需要一个整数来表示需要调用屏障实例上的 await() 方法以表示到达公共执行点的线程数：

public CyclicBarrier(int parties)

需要同步执行的线程也称为参与方，调用 await() 方法是可以注册某个线程已到达屏障点的方式。

此调用是同步的，调用此方法的线程会暂停执行，直到指定数量的线程在屏障上调用了相同的方法。 所需数量的线程调用了 await() 的这种情况称为跳闸。

或者，可以将第二个参数传递给构造函数，它是一个 Runnable 实例。 这具有由最后一个触发屏障的线程运行的逻辑：

public CyclicBarrier(int parties, Runnable barrierAction)

# 5.实现 #

要查看 CyclicBarrier 的实际效果，考虑以下场景：

有一个操作，固定数量的线程执行并将相应的结果存储在列表中。 当所有线程完成它们的操作后，其中一个（通常是最后一个触发屏障的线程）开始处理每个线程获取的数据。

实现所有动作发生的主类：

public class CyclicBarrierDemo { 
    
        private CyclicBarrier cyclicBarrier;
        private List<List<Integer>> partialResults
         = Collections.synchronizedList(new ArrayList<>());
        private Random random = new Random();
        private int NUM_PARTIAL_RESULTS;
        private int NUM_WORKERS;
    
        // ...
    }

这个类非常简单——NUM\_WORKERS 是将要执行的线程数，NUM\_PARTIAL\_RESULTS 是每个工作线程将要产生的结果数。

最后，有 partialResults，它是一个列表，用于存储每个工作线程的结果。 请注意，这个列表是一个 SynchronizedList，因为多个线程将同时写入它，并且 add() 方法在普通 ArrayList 上不是线程安全的。

现在来实现每个工作线程的逻辑：

public class CyclicBarrierDemo { 
    
        // ...
    
         class NumberCruncherThread implements Runnable { 
    
            @Override
            public void run() { 
                String thisThreadName = Thread.currentThread().getName();
                List<Integer> partialResult = new ArrayList<>();
    
         
                for (int i = 0; i < NUM_PARTIAL_RESULTS; i++) { 
                    Integer num = random.nextInt(10);
                    System.out.println(thisThreadName
                            + ": 处理一些数字！ 最后结果 [" + num + "]");
                    partialResult.add(num);
                }
    
                partialResults.add(partialResult);
                try { 
                    System.out.println(thisThreadName
                            + " 等待其他线程到达屏障.");
                    cyclicBarrier.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                } catch (BrokenBarrierException e) { 
                    e.printStackTrace();
                }
            }
        }
    
    }

现在实现障碍物被触发时运行的逻辑。

为了简单起见，将部分结果列表中的所有数字相加：

ublic class CyclicBarrierDemo { 
    
        // ...
        
          class AggregatorThread implements Runnable { 
    
            @Override
            public void run() { 
    
                String thisThreadName = Thread.currentThread().getName();
    
                System.out.println(
                        thisThreadName + ": 计算" + NUM_WORKERS
                                + " 个工作线程, 每个工作线程 " + NUM_PARTIAL_RESULTS + " 组结果的总和.");
                int sum = 0;
    
                for (List<Integer> threadResult : partialResults) { 
                    System.out.print("Adding ");
                    for (Integer partialResult : threadResult) { 
                        System.out.print(partialResult + " ");
                        sum += partialResult;
                    }
                    System.out.println();
                }
                System.out.println(thisThreadName + ": 最终结果为 = " + sum);
            }
        }
    
    }

最后一步是构建 CyclicBarrier 并使用 main() 方法启动：

public class CyclicBarrierDemo { 
    
        // Previous code
     
         public void runSimulation(int numWorkers, int numberOfPartialResults) { 
            NUM_PARTIAL_RESULTS = numberOfPartialResults;
            NUM_WORKERS = numWorkers;
            //AggregatorThread 回调执行函数
            cyclicBarrier = new CyclicBarrier(NUM_WORKERS, new AggregatorThread());
    
            System.out.println("产生 " + NUM_WORKERS
                    + " 工作线程计算，每个线程产生"
                    + NUM_PARTIAL_RESULTS + "组数据的结果");
    
            for (int i = 0; i < NUM_WORKERS; i++) { 
                Thread worker = new Thread(new NumberCruncherThread());
                worker.setName("Thread " + i);
                worker.start();
            }
        }
    
        public static void main(String[] args) { 
            CyclicBarrierDemo demo = new CyclicBarrierDemo();
            demo.runSimulation(5, 3);
        }
        }
    }

在上面的代码中，用 5 个线程初始化了循环屏障，每个线程产生 3 个整数作为计算的一部分，并将它们存储在结果列表中。

一旦屏障被触发，最后一个触发屏障的线程将执行 AggregatorThread 中指定的逻辑，即 - 将线程产生的所有数字相加。

# 6.结果 #

这是上述程序一次执行的输出——每次执行可能会产生不同的结果，因为线程可以以不同的顺序产生：

> 产生 5 工作线程计算，每个线程产生3组数据的结果  
> Thread 1: 处理一些数字！ 最后结果 \[9\]  
> Thread 1: 处理一些数字！ 最后结果 \[5\]  
> Thread 1: 处理一些数字！ 最后结果 \[7\]  
> Thread 1 等待其他线程到达屏障.  
> Thread 0: 处理一些数字！ 最后结果 \[8\]  
> Thread 0: 处理一些数字！ 最后结果 \[0\]  
> Thread 0: 处理一些数字！ 最后结果 \[6\]  
> Thread 0 等待其他线程到达屏障.  
> Thread 3: 处理一些数字！ 最后结果 \[9\]  
> Thread 3: 处理一些数字！ 最后结果 \[6\]  
> Thread 3: 处理一些数字！ 最后结果 \[8\]  
> Thread 3 等待其他线程到达屏障.  
> Thread 2: 处理一些数字！ 最后结果 \[8\]  
> Thread 2: 处理一些数字！ 最后结果 \[7\]  
> Thread 2: 处理一些数字！ 最后结果 \[9\]  
> Thread 2 等待其他线程到达屏障.  
> Thread 4: 处理一些数字！ 最后结果 \[7\]  
> Thread 4: 处理一些数字！ 最后结果 \[5\]  
> Thread 4: 处理一些数字！ 最后结果 \[2\]  
> Thread 4 等待其他线程到达屏障.  
> Thread 4: 计算5 个工作线程, 每个工作线程 3 组结果的总和.  
> Adding 9 5 7  
> Adding 8 0 6  
> Adding 9 6 8  
> Adding 8 7 9  
> Adding 7 5 2  
> Thread 4: 最终结果为 = 96

如上面的输出所示，线程 4 是触发屏障并执行最终聚合逻辑的线程。 线程实际上也没有必要按照它们启动的顺序运行，如上例所示。