Exemple d'analyse de la barrière de boucle de programmation simultanée Java CyclicBarrier

CyclicBarrier

Le CountDownLatch introduit précédemment a été beaucoup optimisé pour résoudre la synchronisation de plusieurs threads par rapport à la méthode de jointure d'appel des threads. Cependant, le compteur de CountDownLatch est unique, c'est-à-dire qu'une fois que la valeur du compteur atteint 0, l'appel des méthodes d'attente et de compte à rebours de CountDownLatch reviendra immédiatement, ce qui n'obtiendra pas l'effet de synchronisation des threads. Par conséquent, afin de répondre au besoin de réinitialisation du compteur, l'équipe de développement du JDK fournit la classe CyclicBarrier, et les fonctions de la classe CyclicBarrier ne se limitent pas aux fonctions de CountDownLatch. Littéralement compris, CyclicBarrier signifie une barrière de boucle, qui permet à un groupe de threads d'atteindre tous un état puis de les exécuter tous en même temps. La raison pour laquelle on l'appelle bouclage est qu'il peut être réutilisé une fois que tous les threads en attente ont fini de s'exécuter et que l'état du CyclicBarrier est réinitialisé. C'est ce qu'on appelle une barrière car le thread sera bloqué après avoir appelé la méthode wait. Ce point de blocage est appelé le point barrière. Une fois que tous les threads auront appelé la méthode wait, les threads franchiront la barrière et continueront à descendre. Avant de présenter les principes, nous présenterons quelques exemples pour approfondir notre compréhension. Dans l'exemple suivant, ce que nous voulons réaliser est d'utiliser deux threads pour exécuter une tâche décomposée A, puis de résumer leurs résultats une fois que les deux threads ont terminé leurs tâches.

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest {

   
   //创建一个线程数固定为2的线程池
   private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
       @Override
       public void run() {
           System.out.println(Thread.currentThread() + " task1 merge result");
       }
   });

   public static void main(String[] args) throws InterruptedException{
       ExecutorService executorService = Executors.newFixedThreadPool(2);

       
       //添加线程A到线程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task1");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //添加线程B到线程池
       executorService.submit(new Runnable() {
           @Override
           public void run() {
               try {
                   System.out.println(Thread.currentThread() + "task2");
                   System.out.println(Thread.currentThread() + "enter in  barrier");
                   cyclicBarrier.await();
                   System.out.println(Thread.currentThread() + "enter out  barrier");
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       });

       //关闭线程池
       executorService.shutdown();

   }
}

Le code ci-dessus crée un objet CyclicBarrier, dont le premier paramètre est la valeur initiale du compteur, et le deuxième nombre Runable est la tâche qui doit être effectuée lorsque la valeur du compteur est 0. Dans la fonction principale, un pool de threads de taille 2 est d'abord créé. Ajoutez deux sous-tâches au pool de threads et chaque sous-tâche appellera cette méthode après avoir terminé sa propre logique. Au début, la valeur du compteur est 2. Lorsque le premier thread appelle la méthode wait, la valeur du compteur sera décrémentée à 1. Puisque la valeur du compteur n'est pas 0 à ce moment, le thread actuel atteint le point de barrière et est bloqué. Ensuite, lorsque le deuxième thread appelle wait, il entrera dans la barrière et la valeur du compteur sera également décrémentée. Désormais, la valeur du compteur est 0. À ce moment, il exécutera la tâche dans le constructeur CyclicBarrier. Une fois l'exécution terminée, il exécutera la tâche dans le constructeur CyclicBarrier. quittera le point de barrière et réveillera le deuxième thread bloqué. À ce moment-là, le premier thread quittera également le point de barrière et continuera à descendre.

L'exemple ci-dessus montre que plusieurs threads s'attendent. Si la valeur du compteur est N, alors les N1 threads qui appellent ensuite la méthode wait seront bloqués car ils atteignent le point de barrière lorsque le Nième thread appelle wait. La valeur du compteur atteint 0, alors seulement le Nième thread émettra une notification pour réveiller les N1 threads précédents. C'est-à-dire que lorsque tous les threads atteignent le point de barrière, ils peuvent continuer à s'exécuter ensemble vers le bas. Cet exemple peut obtenir un résultat similaire en utilisant CountDownLatch. Donnons un autre exemple pour illustrer la réutilisabilité de CyclicBarrier.

Supposons qu'une tâche se compose de la phase 1, de la phase 2 et de la phase 3. Chaque thread doit exécuter la phase 1, la phase 2 et la phase 3 en série. Lorsque plusieurs threads exécutent la tâche, l'exécution de la phase 1 de tous les threads doit être garantie. ne peut être entré qu'une fois que tous sont terminés, et l'exécution de la phase 3 ne peut être entrée qu'une fois la phase 2 de tous les threads terminée. CyclicBarrier est utilisé ci-dessous pour répondre à cette exigence.

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CycleBarrierTest1 {

    //创建一个线程数固定为2的线程池
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);

    public static void main(String[] args) throws InterruptedException{
        ExecutorService executorService = Executors.newFixedThreadPool(2);


        //添加线程A到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //添加线程B到线程池
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println(Thread.currentThread() + "step1");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step2");
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread() + "step3");
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //关闭线程池
        executorService.shutdown();

    }
}

Dans le code ci-dessus, chaque sous-thread appelle la méthode wait après avoir exécuté la phase 1. Il attend que tous les threads atteignent le point de barrière avant de s'exécuter ensemble. Cela garantit que tous les threads auront terminé la phase 2. être exécuté seulement après 1.

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!