How to apply CyclicBarrier cycle barrier in Java

1. Introduction

CyclicBarrier literally means loop barrier (cyclic barrier). It can make a group of threads wait for a certain state (barrier point) and then execute them all at the same time. It is called loopback because CyclicBarrier can be reused after all waiting threads are released.

The function of CyclicBarrier is to make a group of threads wait for each other. When a common point is reached, all previously waiting threads will continue to execute, and the CyclicBarrier function can be reused.

2. Use of CyclicBarrier

Construction method:

 // parties表示屏障拦截的线程数量，每个线程调用 await 方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。
 public CyclicBarrier(int parties)
 // 用于在线程到达屏障时，优先执行 barrierAction，方便处理更复杂的业务场景(该线程的执行时机是在到达屏障之后再执行)

Important method:

//屏障 指定数量的线程全部调用await()方法时，这些线程不再阻塞
// BrokenBarrierException 表示栅栏已经被破坏，破坏的原因可能是其中一个线程 await() 时被中断或者超时
public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException
//循环  通过reset()方法可以进行重置

CyclicBarrier Application Scenario

• Using CyclicBarrier can be used in scenarios where multi-threaded data is calculated and the calculation results are finally merged.

• Using the feature of CyclicBarrier that the counter can be reset and the barrier can be reused, it can support scenarios similar to "full departure"

Simulation Merged calculation scenario

Using CyclicBarrier can be used to calculate data in multiple threads and finally merge the calculation results.

public class CyclicBarrierTest2 {
    //保存每个学生的平均成绩
    private Conc urrentHashMap<String, Integer> map=new ConcurrentHashMap<String,Integer>();
    private ExecutorService threadPool= Executors.newFixedThreadPool(3);
    private CyclicBarrier cb=new CyclicBarrier(3,()->{
        int result=0;
        Set<String> set = map.keySet();
        for(String s:set){
            result+=map.get(s);
        }
        System.out.println("三人平均成绩为:"+(result/3)+"分");
    });
    public void count(){
        for(int i=0;i<3;i++){
            threadPool.execute(new Runnable(){

                @Override
                public void run() {
                    //获取学生平均成绩
                    int score=(int)(Math.random()*40+60);
                    map.put(Thread.currentThread().getName(), score);
                    System.out.println(Thread.currentThread().getName()
                            +"同学的平均成绩为："+score);
                    try {
                        //执行完运行await(),等待所有学生平均成绩都计算完毕
                        cb.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }

            });
        }
    }
    public static void main(String[] args) {
        CyclicBarrierTest2 cb=new CyclicBarrierTest2();
        cb.count();
    }
}

Simulate the scene of "the bus is full of people"

Using the characteristics of CyclicBarrier's counter can be reset and the barrier can be reused, it can support scenes similar to "the bus is full of people"

public class CyclicBarrierTest3 {
    public static void main(String[] args) {
        AtomicInteger counter = new AtomicInteger();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                5, 5, 1000, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100),
                (r) -> new Thread(r, counter.addAndGet(1) + " 号 "),
                new ThreadPoolExecutor.AbortPolicy());

        CyclicBarrier cyclicBarrier = new CyclicBarrier(5,
                () -> System.out.println("裁判：比赛开始~~"));

        for (int i = 0; i < 10; i++) {
            threadPoolExecutor.submit(new Runner(cyclicBarrier));
        }

    }
    static class Runner extends Thread{
        private CyclicBarrier cyclicBarrier;
        public Runner (CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }
        @Override
        public void run() {
            try {
                int sleepMills = ThreadLocalRandom.current().nextInt(1000);
                Thread.sleep(sleepMills);
                System.out.println(Thread.currentThread().getName() + " 选手已就位, 准备共用时： " + sleepMills + "ms" + cyclicBarrier.getNumberWaiting());
                cyclicBarrier.await();

            } catch (InterruptedException e) {
                e.printStackTrace();
            }catch(BrokenBarrierException e){
                e.printStackTrace();
            }
        }
    }

}

Output result:

Player No. 3 is in place, ready to share: 78ms0
Player No. 1 is in place, ready to share: 395ms1## Player No. 5 is in position, ready to share: 733ms2 Player No. 2 is in position, ready to share: 776ms Player No. 3
is in position, ready to share: 807ms4
Referee: The game has started ~~
4 players are in place, ready to share: 131ms0
3 players are in place, ready to share: 256ms1
2 players are in place, ready to share: 291ms2
Player No. 1 is in place, ready to share: 588ms3
Player No. 5 is in place, ready to share: 763ms4
Referee: The game begins~~

3. CyclicBarrier source code analysis

CyclicBarrier process

The main process is:

Acquire the lock and enter blocking if count != 0;
Before entering blocking, you first need to enter the condition queue, then release the lock, and finally block;
If count != 0, a wake-up will be performed. All nodes in the condition queue are converted into blocking queues;
After being awakened, the lock will be acquired. If the lock acquisition fails, it will enter the lock blocking queue;
If the lock is acquired successfully, the lock is released, and the threads in the synchronization queue are awakened.

The following is a simple flow chart:

The following are some specific codes Calling process:

#A few common questions?

1. A group of threads wait for each other before triggering the barrier. How is the wake-up logic implemented after the last thread reaches the barrier. The wake-up process is by calling
• java.util. concurrent.locks.Condition#signalAll

  Wake up all nodes on the condition queue.

2. How is column deletion cycle implemented? In fact, the condition queue and blocking queue of a mutex ReentrantLock are converted.
3. Implementation logic of conversion from condition queue to synchronization queue? During the conversion process, all blocked threads in the condition queue will first be awakened, and then the lock will be acquired. If the acquisition fails, Enter the synchronization queue.
The difference between CyclicBarrier and CountDownLatch

The counter of CountDownLatch can only be used once, while the counter of CyclicBarrier can be reset using the reset() method. Therefore, CyclicBarrier can handle more complex business scenarios. For example, if a calculation error occurs, the counter can be reset and the threads can be re-executed.
CyclicBarrier also provides getNumberWaiting (which can obtain CyclicBarrier blocking the number of threads), isBroken (used to know whether the blocked thread has been interrupted) and other methods.
CountDownLatch will block the main thread, CyclicBarrier will not block the main thread, only the child thread.
Both CountDownLatch and CyclicBarrier can realize waiting between threads, but they have different focuses. CountDownLatch is generally used for one or more threads to wait for other threads to complete their tasks before executing them. CyclicBarrier is generally used for a group of threads to wait for each other to reach a certain state, and then the group of threads execute at the same time.
CyclicBarrier can also provide a barrierAction to merge multi-threaded calculation results.

• CyclicBarrier implements the blocking and awakening of a group of threads through ReentrantLock's "exclusive lock" and Conditon, while CountDownLatch is implemented through the "shared lock" of AQS

The above is the detailed content of How to apply CyclicBarrier cycle barrier in Java. For more information, please follow other related articles on the PHP Chinese website!