Beispielanalyse der Java-Parallelschleifenbarriere CyclicBarrier

CyclicBarrier

Der zuvor eingeführte CountDownLatch wurde im Vergleich zur Join-Methode zum Aufrufen von Threads erheblich optimiert, um die Synchronisation mehrerer Threads zu lösen. Der Zähler von CountDownLatch ist jedoch einmalig, dh nachdem der Zählerwert 0 erreicht hat, kehren die Aufrufe der Warte- und Countdown-Methoden von CountDownLatch sofort zurück, wodurch der Effekt der Thread-Synchronisierung nicht erzielt wird. Um den Bedarf an einem Zurücksetzen des Zählers zu decken, stellt das JDK-Entwicklungsteam daher die CyclicBarrier-Klasse bereit, und die Funktionen der CyclicBarrier-Klasse sind nicht auf die Funktionen von CountDownLatch beschränkt. Im wörtlichen Sinne bedeutet CyclicBarrier eine Schleifenbarriere, die es einer Gruppe von Threads ermöglicht, alle einen Zustand zu erreichen und sie dann alle gleichzeitig auszuführen. Der Grund, warum es hier Loopback genannt wird, besteht darin, dass es wiederverwendet werden kann, nachdem alle wartenden Threads die Ausführung abgeschlossen haben und der Status von CyclicBarrier zurückgesetzt wurde. Dies wird als Barriere bezeichnet, da der Thread nach dem Aufruf der Wait-Methode blockiert wird. Dieser Blockierungspunkt wird Barrierepunkt genannt. Nachdem alle Threads die Wait-Methode aufgerufen haben, durchbrechen die Threads die Barriere und laufen weiter nach unten. Bevor wir die Prinzipien vorstellen, werden wir einige Beispiele vorstellen, um unser Verständnis zu vertiefen. Im folgenden Beispiel möchten wir zwei Threads verwenden, um eine zerlegte Aufgabe A auszuführen, und dann ihre Ergebnisse zusammenfassen, nachdem die beiden Threads ihre Aufgaben abgeschlossen haben.

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

   }
}

Der obige Code erstellt ein CyclicBarrier-Objekt, dessen erster Parameter der Anfangswert des Zählers ist und dessen zweite Zahl Runable die Aufgabe ist, die ausgeführt werden muss, wenn der Zählwert 0 ist. In der Hauptfunktion wird zunächst ein Thread-Pool der Größe 2 erstellt. Fügen Sie dem Thread-Pool zwei Unteraufgaben hinzu, und jede Unteraufgabe ruft diese Methode auf, nachdem sie ihre eigene Logik abgeschlossen hat. Zu Beginn ist der Zählerwert 2. Wenn der erste Thread die Wartemethode aufruft, wird der Zählerwert auf 1 dekrementiert. Da der Zählerwert zu diesem Zeitpunkt nicht 0 ist, erreicht der aktuelle Thread den Barrierepunkt und wird blockiert. Wenn dann der zweite Thread aufruft, tritt er in die Barriere ein und der Zählerwert wird ebenfalls verringert. Zu diesem Zeitpunkt wird die Aufgabe im CyclicBarrier-Konstruktor ausgeführt verlässt den Barrierepunkt und weckt den blockierten zweiten Thread auf. Zu diesem Zeitpunkt verlässt auch der erste Thread den Barrierepunkt und läuft weiter nach unten.

Das obige Beispiel zeigt, dass mehrere Threads aufeinander warten. Wenn der N1-Thread die Wait-Methode aufruft, werden sie blockiert, wenn der N-te Thread den Wartepunkt erreicht Wenn der Zählerwert 0 erreicht, gibt der N-te Thread erst dann eine Benachrichtigung aus, um die vorherigen N1-Threads aufzuwecken. Das heißt, wenn alle Threads den Barrierepunkt erreichen, können sie gemeinsam weiter nach unten ausgeführt werden. In diesem Beispiel kann mithilfe von CountDownLatch eine ähnliche Ausgabe erzielt werden. Lassen Sie uns ein weiteres Beispiel geben, um die Wiederverwendbarkeit von CyclicBarrier zu veranschaulichen.

Angenommen, eine Aufgabe besteht aus Phase 1, Phase 2 und Phase 3. Jeder Thread muss Phase 1, Phase 2 und Phase 3 nacheinander ausführen. Wenn mehrere Threads die Aufgabe ausführen, muss die Ausführung von Phase 1 aller Threads gewährleistet sein Die Ausführung von Phase 3 kann erst eingegeben werden, nachdem alle Threads abgeschlossen sind, und die Ausführung von Phase 3 kann erst eingegeben werden, nachdem Phase 2 aller Threads abgeschlossen ist. CyclicBarrier wird im Folgenden verwendet, um diese Anforderung zu erfüllen.

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

    }
}

Im obigen Code ruft jeder Sub-Thread nach der Ausführung von Phase 1 auf. Er wartet, bis alle Threads den Barrierepunkt erreichen, bevor er gemeinsam ausgeführt wird. Dadurch wird sichergestellt, dass alle Threads die Phase 2 abgeschlossen haben erst nach dem 1. ausgeführt werden.

Das obige ist der detaillierte Inhalt vonBeispielanalyse der Java-Parallelschleifenbarriere CyclicBarrier. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!