Comment appliquer la barrière de cycle CyclicBarrier en Java

1. Introduction

CyclicBarrier signifie littéralement barrière de boucle (barrière cyclique). Il peut faire attendre un groupe de threads pour un certain état (point de barrière) puis les exécuter tous en même temps. C'est ce qu'on appelle le bouclage car CyclicBarrier peut être réutilisé une fois que tous les threads en attente sont libérés.

La fonction de CyclicBarrier est de faire attendre un groupe de threads lorsqu'un point commun est atteint, tous les threads précédemment en attente continueront à s'exécuter et la fonction CyclicBarrier pourra être réutilisée.

2. Utilisation de CyclicBarrier

Méthode de construction :

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

Méthodes importantes :

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

Scénarios d'application de CyclicBarrier

• L'utilisation de CyclicBarrier peut être utilisée pour plusieurs calcul threadé des données, et enfin fusionner les résultats du calcul.

• Utilisant les fonctionnalités de CyclicBarrier selon lesquelles le compteur peut être réinitialisé et la barrière peut être réutilisée, il peut prendre en charge des scénarios similaires à "le train est plein de monde"

Simuler des scénarios de calcul fusionnés

Utilisez CyclicBarrier pour calculer des données dans plusieurs threads, et enfin des scénarios de fusion des résultats de calcul.

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();
    }
}

Simulez la scène du "le train est plein de monde"

En utilisant les fonctionnalités de CyclicBarrier qui permettent de réinitialiser le compteur et de réutiliser la barrière, il peut prendre en charge la scène du "le train est plein de monde"

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();
            }
        }
    }

}

Résultat de sortie :

3 Joueur n°1 est en position, prêt à partager : 78ms 0
1 joueur est en position, prêt à partager : 395ms 1
5 joueur est en position, prêt à partager : 733ms 2
2 joueur est en position, prêt à partager : 776ms3
Joueur n°4 est en position, prêt à partager : 807ms4
Arbitre : Le jeu a commencé~~
Joueur n°4 est en position, prêt à partager : 131ms0
Le joueur n°3 est en position, prêt à partager : 256ms1
Joueur n°2 déjà en place, prêt à partager : 291ms2
Le joueur n°1 est en place, prêt à partager : 588ms3
Le joueur n°5 est en place , prêt à partager : 763ms4
Arbitre : Le jeu a commencé~~

3. Analyse du code source de CyclicBarrier

Processus CyclicBarrier

Le processus principal est :

• Obtenez le verrou et entrez le blocage si vous comptez ! = 0;

• Avant d'entrer dans le blocage, vous devez d'abord entrer dans la file d'attente des conditions, puis libérer le verrou, et enfin bloquer

• Si count != 0, un réveil sera effectué et tous les nœuds dans la condition ; la file d'attente sera convertie en files d'attente de blocage ;

• Après avoir été réveillé, le verrou sera acquis. Si l'acquisition du verrou échoue, il entrera dans la file d'attente de blocage du verrou ;

• Si le verrou est acquis avec succès, le le verrou est libéré et les threads de la file d'attente de synchronisation sont réveillés.

Ce qui suit est un organigramme simple :

Ce qui suit est le processus spécifique de certains appels de code :

Quelques questions courantes ?

• 1. Un groupe de threads s'attendent avant de déclencher la barrière. Comment la logique de réveil est-elle mise en œuvre une fois que le dernier thread atteint la barrière ? file d’attente des conditions. java.util.concurrent.locks.Condition#signalAll

• 2. Comment le cycle de suppression de colonnes est-il implémenté ? En fait, la file d'attente de conditions et la file d'attente de blocage d'un mutex ReentrantLock sont converties.

• 3. Logique de mise en œuvre de la conversion de la file d'attente de conditions en file d'attente de synchronisation ? Pendant le processus de conversion, tous les threads bloqués dans la file d'attente de conditions seront d'abord réveillés, puis le verrou sera acquis si l'acquisition échoue, il entrera. la file d'attente de synchronisation.

La différence entre CyclicBarrier et CountDownLatch

• Le compteur de CountDownLatch ne peut être utilisé qu'une seule fois, tandis que le compteur de CyclicBarrier peut être réinitialisé à l'aide de la méthode reset(). Par conséquent, CyclicBarrier peut gérer des scénarios métiers plus complexes. Par exemple, si une erreur de calcul se produit, le compteur peut être réinitialisé et les threads peuvent être réexécutés. CyclicBarrier fournit également getNumberWaiting (pour obtenir le nombre de threads bloqués par CyclicBarrier), isBroken. (en utilisant pour savoir si le thread bloqué a été interrompu) et d'autres méthodes.

• CountDownLatch bloquera le thread principal, CyclicBarrier ne bloquera pas le thread principal, seulement le thread enfant.

• CountDownLatch et CyclicBarrier peuvent implémenter l'attente entre les threads, mais ils ont des objectifs différents. CountDownLatch est généralement utilisé pour qu'un ou plusieurs threads attendent que d'autres threads terminent leurs tâches avant de les exécuter. CyclicBarrier est généralement utilisé pour qu'un groupe de threads s'attende les uns les autres pour atteindre un certain état, puis le groupe de threads s'exécute en même temps.

• CyclicBarrier peut également fournir une barrièreAction pour fusionner les résultats de calcul multithread.

• CyclicBarrier implémente le blocage et le réveil d'un groupe de threads via le "verrou exclusif" et Conditon de ReentrantLock, tandis que CountDownLatch est implémenté via le "verrou partagé" d'AQS

  # 🎜🎜#

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!