Java-Parallelität: Countdown-Latch gegen zyklische Barriere

Ich habe die API java.util.concurrent gelesen und festgestellt, dass

• CountDownLatch: Eine Synchronisationshilfe, mit der ein oder mehrere Threads warten können, bis eine Reihe von Vorgängen in anderen Threads abgeschlossen ist.
• CyclicBarrier: Eine Synchronisationshilfe, mit der eine Reihe von Threads darauf warten können, dass sie einen gemeinsamen Barrierepunkt erreichen.

Für mich scheinen beide gleich zu sein, aber ich bin mir sicher, dass noch viel mehr dahinter steckt.

Zum Beispiel in CoundownLatch, the countdown value could not be reset, that can happen in the case of CyclicBarrier.

Gibt es einen anderen Unterschied zwischen den beiden?
Was sind die , use caseswo jemand den Wert des Countdowns zurücksetzen möchte?

Ein wesentlicher Unterschied besteht darin, dass CyclicBarrier eine (optionale) ausführbare Aufgabe ausführt, die ausgeführt wird, sobald die allgemeine Barrierebedingung erfüllt ist.

Außerdem können Sie die Anzahl der an der Barriere wartenden Clients und die zum Auslösen der Barriere erforderliche Anzahl ermitteln. Nach dem Auslösen wird die Barriere zurückgesetzt und kann wieder verwendet werden.

Für einfache Anwendungsfälle - Dienste starten usw. ... ist ein CountdownLatch in Ordnung. Ein CyclicBarrier ist nützlich für komplexere Koordinierungsaufgaben. Ein Beispiel für so etwas wäre die parallele Berechnung - bei der mehrere Unteraufgaben an der Berechnung beteiligt sind - ähnlich wie bei MapReduce .

Es gibt noch einen anderen Unterschied.

Bei Verwendung von a CyclicBarrierwird davon ausgegangen, dass Sie die Anzahl der wartenden Threads angeben, die die Barriere auslösen. Wenn Sie 5 angeben, müssen mindestens 5 Threads aufgerufen werden await().

Wenn Sie a verwenden CountDownLatch, geben Sie die Anzahl der Aufrufe an countDown(), die dazu führen , dass alle wartenden Threads freigegeben werden. Dies bedeutet, dass Sie a CountDownLatchmit nur einem Thread verwenden können.

"Warum sollten Sie das tun?", Können Sie sagen. Stellen Sie sich vor, Sie verwenden eine mysteriöse API, die von einer anderen Person codiert wurde, die Rückrufe ausführt. Sie möchten, dass einer Ihrer Threads wartet, bis ein bestimmter Rückruf mehrmals aufgerufen wurde. Sie haben keine Ahnung, auf welchen Threads der Rückruf aufgerufen wird. In diesem Fall ist a CountDownLatchperfekt, während ich mir keine Möglichkeit CyclicBarriervorstellen kann, dies mit a zu implementieren (eigentlich kann ich das, aber es beinhaltet Zeitüberschreitungen ... igitt!).

Ich wünschte nur, das CountDownLatchkönnte zurückgesetzt werden!

Ein Punkt, den noch niemand erwähnt hat, ist, dass in einem CyclicBarrier, wenn ein Thread ein Problem hat (Timeout, unterbrochen ...), alle anderen, die erreicht haben await(), eine Ausnahme bekommen. Siehe Javadoc:

Der CyclicBarrier verwendet ein All-or-None-Bruchmodell für fehlgeschlagene Synchronisationsversuche: Wenn ein Thread aufgrund von Unterbrechung, Ausfall oder Zeitüberschreitung einen Sperrpunkt vorzeitig verlässt, werden alle anderen an diesem Sperrpunkt wartenden Threads auch abnormal über BrokenBarrierException (oder InterruptedException) verlassen wenn auch sie ungefähr zur gleichen Zeit unterbrochen wurden).

Ich denke, dass der JavaDoc die Unterschiede explizit erklärt hat. Die meisten Leute wissen, dass CountDownLatch nicht zurückgesetzt werden kann, CyclicBarrier jedoch. Dies ist jedoch nicht der einzige Unterschied, oder der CyclicBarrier könnte in ResetbleCountDownLatch umbenannt werden. Wir sollten die Unterschiede aus der Perspektive ihrer Ziele erkennen, die in JavaDoc beschrieben sind

CountDownLatch: Eine Synchronisationshilfe, mit der ein oder mehrere Threads warten können, bis eine Reihe von Vorgängen in anderen Threads abgeschlossen ist.

CyclicBarrier: Eine Synchronisationshilfe, mit der eine Reihe von Threads darauf warten können, dass sie einen gemeinsamen Barrierepunkt erreichen.

In countDownLatch gibt es einen oder mehrere Threads, die darauf warten, dass eine Reihe anderer Threads abgeschlossen wird. In dieser Situation gibt es zwei Arten von Threads: Ein Typ wartet, ein anderer Typ tut etwas. Nach Abschluss ihrer Aufgaben können sie warten oder einfach beendet werden.

In CyclicBarrier gibt es nur einen Thread-Typ, sie warten aufeinander, sie sind gleich.

Der Hauptunterschied ist direkt in den Javadocs für CountdownLatch dokumentiert. Nämlich:

Ein CountDownLatch wird mit einer bestimmten Anzahl initialisiert. Die Wartemethoden werden blockiert, bis der aktuelle Zählerstand aufgrund von Aufrufen der countDown () -Methode Null erreicht. Danach werden alle wartenden Threads freigegeben und alle nachfolgenden Aufrufe von Wartezeiten werden sofort zurückgegeben. Dies ist ein One-Shot-Phänomen - die Zählung kann nicht zurückgesetzt werden. Wenn Sie eine Version benötigen, die die Anzahl zurücksetzt, sollten Sie einen CyclicBarrier verwenden.

Quelle 1.6 Javadoc

Ein CountDownLatch wird für die einmalige Synchronisation verwendet. Bei Verwendung eines CountDownLatch darf jeder Thread countDown () so oft aufrufen, wie er möchte. Threads, die await () aufgerufen haben, werden blockiert, bis die Anzahl Null erreicht, da andere nicht blockierte Threads countDown () aufrufen. Das Javadoc für CountDownLatch lautet:

Die Wartemethoden werden blockiert, bis der aktuelle Zählerstand aufgrund von Aufrufen der countDown () -Methode Null erreicht. Danach werden alle wartenden Threads freigegeben und alle nachfolgenden Aufrufe von Wartezeiten werden sofort zurückgegeben. ...

Eine andere typische Verwendung wäre, ein Problem in N Teile zu unterteilen, jeden Teil mit einem Runnable zu beschreiben, der diesen Teil ausführt und auf den Latch herunterzählt, und alle Runnables einem Executor in die Warteschlange zu stellen. Wenn alle Unterteile vollständig sind, kann der koordinierende Thread warten warten. (Wenn Threads auf diese Weise wiederholt heruntergezählt werden müssen, verwenden Sie stattdessen einen CyclicBarrier.)

Im Gegensatz dazu wird die zyklische Barriere für mehrere Synchronisationspunkte verwendet, z. B. wenn eine Reihe von Threads eine Schleifen- / Phasenberechnung ausführt und vor dem Start der nächsten Iteration / Phase synchronisiert werden muss. Gemäß dem Javadoc für CyclicBarrier :

Die Barriere wird als zyklisch bezeichnet, da sie nach dem Lösen der wartenden Threads wiederverwendet werden kann.

Im Gegensatz zum CountDownLatch gehört jeder Aufruf von await () zu einer bestimmten Phase und kann dazu führen, dass der Thread blockiert wird, bis alle zu dieser Phase gehörenden Parteien await () aufgerufen haben. Es gibt keine explizite countDown () -Operation, die vom CyclicBarrier unterstützt wird.

Diese Frage wurde bereits angemessen beantwortet, aber ich denke, ich kann durch das Posten eines Codes einen kleinen Mehrwert schaffen.

Um das Verhalten der zyklischen Barriere zu veranschaulichen, habe ich einen Beispielcode erstellt. Sobald die Barriere gekippt wird, wird sie automatisch zurückgesetzt, damit sie wieder verwendet werden kann (daher ist sie "zyklisch"). Beachten Sie beim Ausführen des Programms, dass die Ausdrucke "Lass uns spielen" erst ausgelöst werden, nachdem die Barriere gekippt wurde.

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierCycles {

    static CyclicBarrier barrier;

    public static void main(String[] args) throws InterruptedException {
        barrier = new CyclicBarrier(3); 

        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);

        System.out.println("Barrier automatically resets.");

        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
    }

}


class Worker extends Thread {
    @Override
    public void run() {
        try {
            CyclicBarrierCycles.barrier.await();
            System.out.println("Let's play.");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

Als ich über Latches und Cyclicbarriers studierte, kam ich auf diese Metaphern. Cyclicbarriers : Stellen Sie sich vor, ein Unternehmen hat einen Besprechungsraum. Um das Meeting zu starten, muss eine bestimmte Anzahl von Meeting-Teilnehmern zum Meeting kommen (um es offiziell zu machen). Das Folgende ist der Code eines normalen Besprechungsteilnehmers (eines Mitarbeiters).

class MeetingAtendee implements Runnable {

CyclicBarrier myMeetingQuorumBarrier;

public MeetingAtendee(CyclicBarrier myMileStoneBarrier) {
    this.myMeetingQuorumBarrier = myMileStoneBarrier;
}

@Override
public void run() {
    try {
        System.out.println(Thread.currentThread().getName() + " i joined the meeting ...");
        myMeetingQuorumBarrier.await();
        System.out.println(Thread.currentThread().getName()+" finally meeting stared ...");
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (BrokenBarrierException e) {
        System.out.println("Meeting canceled! every body dance <by chic band!>");
    }
 }
}

Der Mitarbeiter nimmt an der Besprechung teil und wartet darauf, dass andere Personen zur Besprechung kommen. Außerdem wird er verlassen, wenn das Meeting abgesagt wird :) Dann haben wir THE BOSS, wie Dosen nicht gerne darauf warten, dass andere auftauchen, und wenn er seinen Patienten verliert, bricht er das Meeting ab.

class MeetingAtendeeTheBoss implements Runnable {

CyclicBarrier myMeetingQuorumBarrier;

public MeetingAtendeeTheBoss(CyclicBarrier myMileStoneBarrier) {
    this.myMeetingQuorumBarrier = myMileStoneBarrier;
}

@Override
public void run() {
    try {
        System.out.println(Thread.currentThread().getName() + "I am THE BOSS - i joined the meeting ...");
        //boss dose not like to wait too much!! he/she waits for 2 seconds and we END the meeting
        myMeetingQuorumBarrier.await(1,TimeUnit.SECONDS);
        System.out.println(Thread.currentThread().getName()+" finally meeting stared ...");
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (BrokenBarrierException e) {
        System.out.println("what WHO canceled The meeting");
    } catch (TimeoutException e) {
        System.out.println("These employees waste my time!!");
    }
 }
}

An einem normalen Tag kommen Mitarbeiter zu Besprechungen und warten darauf, dass andere auftauchen. Wenn einige Teilnehmer nicht kommen, müssen sie auf unbestimmte Zeit warten! In einer besonderen Besprechung kommt der Chef und er wartet nicht gern. (5 Personen müssen mit der Besprechung beginnen, aber nur der Chef und auch ein begeisterter Mitarbeiter.) Er bricht die Besprechung ab (wütend).

CyclicBarrier meetingAtendeeQuorum = new CyclicBarrier(5);
Thread atendeeThread = new Thread(new MeetingAtendee(meetingAtendeeQuorum));
Thread atendeeThreadBoss = new Thread(new MeetingAtendeeTheBoss(meetingAtendeeQuorum));
    atendeeThread.start();
    atendeeThreadBoss.start();

Ausgabe:

//Thread-1I am THE BOSS - i joined the meeting ...
// Thread-0 i joined the meeting ...
// These employees waste my time!!
// Meeting canceled! every body dance <by chic band!>

Es gibt ein anderes Szenario, in dem ein anderer externer Thread (ein Erdbeben) das Meeting abbricht (Call-Reset-Methode). In diesem Fall werden alle wartenden Threads durch eine Ausnahme geweckt.

class NaturalDisasters implements Runnable {

CyclicBarrier someStupidMeetingAtendeeQuorum;

public NaturalDisasters(CyclicBarrier someStupidMeetingAtendeeQuorum) {
    this.someStupidMeetingAtendeeQuorum = someStupidMeetingAtendeeQuorum;
}

void earthQuakeHappening(){
    System.out.println("earth quaking.....");
    someStupidMeetingAtendeeQuorum.reset();
}

@Override
public void run() {
    earthQuakeHappening();
 }
}

Das Ausführen von Code führt zu einer lustigen Ausgabe:

// Thread-1I am THE BOSS - i joined the meeting ...
// Thread-0 i joined the meeting ...
// earth quaking.....
// what WHO canceled The meeting
// Meeting canceled! every body dance <by chic band!>

Sie können dem Besprechungsraum auch eine Sekretärin hinzufügen. Wenn eine Besprechung stattfindet, dokumentiert sie alles, ist jedoch nicht Teil der Besprechung:

class MeetingSecretary implements Runnable {

@Override
public void run() {
        System.out.println("preparing meeting documents");
        System.out.println("taking notes ...");
 }
}

Latches : Wenn der verärgerte Chef eine Ausstellung für Firmenkunden veranstalten möchte, muss alles bereit sein (Ressourcen). Wir stellen eine To-Do-Liste zur Verfügung, in der jeder Arbeiter (Thread) seine Arbeit erledigt, und wir überprüfen die To-Do-Liste (einige Arbeiter malen, andere bereiten das Soundsystem vor ...). Wenn alle Aufgaben in der Aufgabenliste vollständig sind (Ressourcen werden bereitgestellt), können wir den Kunden die Türen öffnen.

public class Visitor implements Runnable{

CountDownLatch exhibitonDoorlatch = null;

public Visitor (CountDownLatch latch) {
    exhibitonDoorlatch  = latch;
}

public void run() {
    try {
        exhibitonDoorlatch .await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    System.out.println("customer visiting exebition");
 }
}

Und die Arbeiter, wie bereiten sie die Ausstellung vor:

class Worker implements Runnable {

CountDownLatch myTodoItem = null;

public Worker(CountDownLatch latch) {
    this.myTodoItem = latch;
}

public void run() {
        System.out.println("doing my part of job ...");
        System.out.println("My work is done! remove it from todo list");
        myTodoItem.countDown();
 }
}

    CountDownLatch preperationTodoList = new CountDownLatch(3);

    // exhibition preparation workers  
    Worker      electricalWorker      = new Worker(preperationTodoList);
    Worker      paintingWorker      = new Worker(preperationTodoList);

    // Exhibition Visitors 
    ExhibitionVisitor exhibitionVisitorA = new ExhibitionVisitor(preperationTodoList);
    ExhibitionVisitor exhibitionVisitorB = new ExhibitionVisitor(preperationTodoList);
    ExhibitionVisitor exhibitionVisitorC = new ExhibitionVisitor(preperationTodoList);

    new Thread(electricalWorker).start();
    new Thread(paintingWorker).start();

    new Thread(exhibitionVisitorA).start();
    new Thread(exhibitionVisitorB).start();
    new Thread(exhibitionVisitorC).start();

Kurz gesagt , um die wichtigsten funktionalen Unterschiede zwischen den beiden zu verstehen :

public class CountDownLatch {
    private Object mutex = new Object();
    private int count;

    public CountDownLatch(int count) {
        this.count = count;
    }

    public void await() throws InterruptedException {
        synchronized (mutex) {
            while (count > 0) {
                mutex.wait();
            }
        }
    }

    public void countDown() {
        synchronized (mutex) {
            if (--count == 0)
                mutex.notifyAll();
        }

    }
}

und

public class CyclicBarrier {
    private Object mutex = new Object();
    private int count;

    public CyclicBarrier(int count) {
        this.count = count;
    }

    public void await() throws InterruptedException {
        synchronized (mutex) {
            count--;
            while(count > 0)
                mutex.wait();
            mutex.notifyAll();
        }
    }
}

mit Ausnahme von Funktionen wie nicht blockierend, zeitgesteuertes Warten, Diagnose und allem, was in den obigen Antworten ausführlich erläutert wurde.

Die oben genannten Klassen sind jedoch voll funktionsfähig und entsprechen innerhalb der bereitgestellten Funktionalität ihren entsprechenden Namensgebern.

In einem anderen Sinne sind die CountDownLatchUnterklassen der inneren Klasse AQSwährend der CyclicBarrierVerwendung ReentrantLock(mein Verdacht ist, dass es anders herum sein könnte oder beide AQS verwenden könnten oder beide Lock verwenden - ohne dass die Leistungseffizienz beeinträchtigt wird).

Ein offensichtlicher Unterschied besteht darin, dass nur N Threads auf einem CyclicBarrier von N darauf warten können, in einem Zyklus freigegeben zu werden. Es kann jedoch eine unbegrenzte Anzahl von Threads auf einen CountDownLatch von N warten. Die Countdown-Dekrementierung kann durch einen Thread N-mal oder N Threads einmal oder durch Kombinationen erfolgen.

Im Fall von CyclicBarrier wird Runnable in der Barriere ausgeführt, sobald ALLE untergeordneten Threads beginnen, barriere.await () aufzurufen. Das Sperren der Barriere in jedem untergeordneten Thread dauert unterschiedlich lange, und alle werden gleichzeitig beendet.

In CountDownLatch warten Hauptthreads darauf , dass andere Threads ihre Ausführung abschließen. In CyclicBarrier warten Arbeitsthreads aufeinander, um ihre Ausführung abzuschließen.

Sie können dieselbe CountDownLatch- Instanz nicht wiederverwenden, wenn der Zählwert Null erreicht und der Latch geöffnet ist. Andererseits kann CyclicBarrier durch Zurücksetzen der Barriere wiederverwendet werden, sobald die Barriere durchbrochen ist.

CountDownLatch ist ein Countdown für alles; CyclicBarrier ist ein Countdown nur für Threads

Angenommen, es gibt 5 Arbeiterfäden und einen Versenderfaden. Wenn Arbeiter 100 Artikel produzieren, werden sie vom Versender versandt.

Bei CountDownLatch kann sich der Zähler auf Arbeitern oder Gegenständen befinden

Für CyclicBarrier kann der Zähler nur auf Arbeiter

Wenn ein Arbeiter mit CountDownLatch für Artikel in den unendlichen Schlaf fällt, kann der Versender versenden. Mit CyclicBarrier kann Shipper jedoch niemals aufgerufen werden

@ Kevin Lee und @ Jon Ich habe CyclicBarrier mit optionalem Runnable ausprobiert. Sieht so aus, als würde es am Anfang und nach dem Kippen des CyclicBarrier ausgeführt. Hier ist der Code und die Ausgabe

statische CyclicBarrier-Barriere;

    public static void main(String[] args) throws InterruptedException {
        barrier = new CyclicBarrier(3, new Runnable() {
            @Override
            public void run() {
                System.out.println("I run in the beginning and after the CyclicBarrier is tipped");
            }
        });

        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);

        System.out.println("Barrier automatically resets.");

        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
        Thread.sleep(1000);
        new Worker().start();
    }

Ausgabe

I run in the beginning and after the CyclicBarrier is tipped
Let's play.
Let's play.
Let's play.
Barrier automatically resets.
I run in the beginning and after the CyclicBarrier is tipped
Let's play.
Let's play.
Let's play.