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表示屏障攔截的線程數(shù)量，每個線程調(diào)用 await 方法告訴 CyclicBarrier 我已經(jīng)到達了屏障，然后當前線程被阻塞。
 public CyclicBarrier(int parties)
 // 用于在線程到達屏障時，優(yōu)先執(zhí)行 barrierAction，方便處理更復雜的業(yè)務場景(該線程的執(zhí)行時機是在到達屏障之后再執(zhí)行)

Important method:

//屏障 指定數(shù)量的線程全部調(diào)用await()方法時，這些線程不再阻塞
// BrokenBarrierException 表示柵欄已經(jīng)被破壞，破壞的原因可能是其中一個線程 await() 時被中斷或者超時
public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException
//循環(huán)  通過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("三人平均成績?yōu)?"+(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()
                            +"同學的平均成績?yōu)椋?quot;+score);
                    try {
                        //執(zhí)行完運行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

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.

#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!