Im Rahmen unserer Artikelreihe über die Neuerungen in Java 7 wollen wir dieses Mal den Phaser ansehen.  Dabei geht es nicht um jene galaktische Strahlenkanone, mit der sich die Star-Trek-Helden verteidigt haben, sondern es gibt einen Synchronizer im JDK-Package java.util.concurrent , der so heißt, weil er in Phasen abläuft. Er hat ähnliche Eigenschaften wie die bereits seit Längerem existierenden Synchronisationsmittel CountDownLatch und CyclicBarrier , ist aber deutlich flexibler.

Der Phaser ist ebenso wie das CountDownLatch und die CyclicBarrier eine der Abstraktionen aus dem JDK, die die Synchronisation von Threads unterstützen.  Allen drei Abstraktionen ist gemeinsam, dass sie ein sogenanntes Thread-Rendezvous implementieren.  Das Thread-Rendezvous-Pattern beschreibt das Treffen von zwei oder mehr Threads, die zusammen kommen, um sich zu synchronisieren, Daten auszutauschen oder ihre Aktionen miteinander zu koordinieren.  Dabei unterscheidet man zwischen waiting , timed , und balking Rendezvous, je nachdem ob die am Treffpunkt ankommenden Threads beliebig lange aufeinander warten (waiting) oder nach einer gewissen Wartezeit den Treffpunkt verlassen (timed) oder sofort weggehen, wenn der oder die anderen Threads nicht schon da sind (balking).  Ein Beispiel für ein wartendes Rendezvous wäre das Treffen mehrerer Worker-Threads, die darauf warten, dass ein anderer Thread alle erforderlichen Vorbereitungen erledigt und dann zum Treffpunkt kommt, ehe die Worker-Threads mit ihrer Arbeit beginnen; die Ankunft des vorbereitenden Threads ist dann so eine Art Start-Signal.

Hinter allen Rendezvous-Implementierungen steckt in der Regel ein Zähler (z.B. für die schon angekommenen oder die noch am Treffpunkt erwarteten Threads), der rauf oder runter gezählt wird, bis er einen Schwellenwert erreicht.  Das Erreichen des Schwellenwerts löst ein Ereignis aus, auf das man warten oder abfragen kann.
 

Die tradionellen Synchronizer CyclicBarrier und CountDownLatch

Die CyclicBarrier ist ähnlich.  Sie ist für das Rendezvous mehrerer Threads konzipiert, die sich treffen und aufeinander warten wollen, bis alle Parteien am Treffpunkt angekommen sind.  Eine CyclicBarrier wird mit der Anzahl der beteiligten Parteien konstruiert.  Auch hier wird ein Zähler runter gezählt:  es wird gezählt, wie viele Parteien noch ankommen müssen, ehe alle da sind.  Wenn alle am Treffpunkt angekommen sind, d.h. wenn der Zähler der noch fehlenden Parteien den Wert 0 erreicht, dann wachen alle Threads auf, die am Treffpunkt gewartet haben.  Beides, das Ankommen und das Warten, ist in der Methode await() kombiniert.  Auch mit einer CyclicBarrier kann man Start- und Ende-Signale geben.  Für ein Start-Signal würde man zum Beispiel eine CyclicBarrier(n+1) verwenden, wobei n die Zahl der Worker-Threads ist.  Die n Worker-Threads würden wieder am Anfang einen await() an der Barriere machen.  Dann sind bereits n der n+1 Parteien angekommen und warten darauf, dass der fehlende Thread ankommt. Der „fehlende“ Thread ist der vorbereitende Thread, der erst an der Barriere ankommt, wenn er seine Vorbereitungen erledigt hat.  Sein Aufruf von await() ist dann das Start-Signal für die wartenden Worker-Threads.

Der wesentliche Unterschied zwischen der CyclicBarrier und dem CountDownLatch ist die Wiederverwendbarkeit.  Bei der CyclicBarrier kann man das Treffen beliebig oft „cyclic“ (= zyklisch) wiederholen.  Ein CountDownLatch hingegen kann man nur einmal verwenden, wie der Namensbestandteil „latch“ (= Riegel, Schnappverschluss) andeutet: es erreicht nur einmal seinen Endzustand und rastet dann ein.  Um im obigen Beispiel ein Ende-Signal von den Worker-Threads an einen nachbereitenden Thread zu schicken, könnte man dieselbe CyclicBarrier verwenden, die auch bereits für das Start-Signal benutzt wurde.  In der Lösung mit dem CountDownLatch benötigt man hingegen wegen der fehlenden Wiederverwendbarkeit zwei verschiedene Latches.

CyclicBarrier und CountDownLatch haben auch Gemeinsamkeiten, insbesondere im Unterschied zum Phaser . Gemeinsam ist ihr Mangel an Flexibilität, denn in beiden Fällen muss die Zahl der beteiligten Parteien bzw. der Anfangswert des Zählers bei der Konstruktion schon bekannt sein und kann auch später nicht mehr dynamisch geändert werden.  Probleme, bei denen sich die Zahl der beteiligten Parteien dynamisch verändert, lassen sich weder mit der CyclicBarrier noch mit dem CountDownLatch lösen.  Eine solche Problemstellung wäre zum Beispiel das Durchsuchen von Baumstrukturen.  Wenn man für jeden Zweig im Baum einen neuen Worker-Thread losschicken will, dann ergibt sich die Zahl der benötigten Worker-Threads erst im Laufe der Navigation.  Zum Konstruktionszeitpunkt von CyclicBarrier oder CountDownLatch weiß man also die Zahl der beteiligten Parteien noch nicht.  Der Phaser hat dieses Defizit nicht: beim Phaser können jederzeit weitere Threads registriert werden, so dass sich das „Baum“-Problem elegant lösen lässt. Ansonsten ähnelt der Phaser stärker der CyclicBarrier als dem CountDownLatch , weil er ebenfalls mehrfach verwendet werden kann.  Was man bei der CyclicBarrier als Zyklus bezeichnet, entspricht der Phase beim Phaser .
 

Der neue Phaser-Synchronizer

• der Zahl der registrierten Parteien,
• der Zahl der bereits am Treffpunkt eingetroffenen Parteien,
• der Zahl der noch am Treffpunkt erwarteten (d.h. noch nicht eingetroffenen) Parteien,
• der Nummer der Phase, und
• dem Terminierungszustand.

Die beteiligten Parteien können bereits bei der Konstruktion registriert werden.  Es können aber auch nach der Konstruktion noch Parteien über die Methoden register() und bulkRegister(int parties) hinzugefügt werden.  Das ist anders als bei der CyclicBarrier , bei der die Zahl der beteiligten Parteien bei der Konstruktion festgelegt wird und nicht mehr geändert werden kann.

Die Ankunft einer Partei am Treffpunkt wird durch die Methode arrive() ausgelöst. Die arrive() -Methode kehrt sofort zurück, d.h. die eingetroffene Partei sagt, dass sie am Treffpunkt war, und macht sofort weiter; sie wartet nicht auf das Eintreffen der anderen Parteien.  Intern wird beim arrive() lediglich die Zahl der eingetroffenen Parteien inkrementiert bzw. die der noch erwarteten Parteien dekrementiert.  Wenn man warten will, bis alle da sind, dann kann man die Methode arriveAndAwaitAdvance() verwenden.  Sie entspricht der await() -Methode der CyclicBarrier .

Wenn alle erwarteten Parteien am Treffpunkt angekommen sind, d.h. wenn die Zahl der eingetroffenen Parteien gleich der Zahl der registrierten Parteien ist (und damit die Zahl der noch erwarteten Parteien gleich Null), dann passiert der sogenannte Advance . Dabei rückt der Phaser in die nächste Phase vor: die Phasen-Nummer wird inkrementiert und die Zahl der eingetroffenen und noch nicht eingetroffenen Parteien werden zurückgesetzt.  Das ist ähnlich wie bei der CyclicBarrier , mit dem Unterschied, dass die CyclicBarrier ihre Zyklen nicht numeriert.

Die Phasennummer kann in der Methode awaitAdvance(int phase) verwendet werden, wenn auf das Ende einer bestimmten Phase gewartet werden soll.

Registrierte Parteien können sich vom Phaser verabschieden, indem sie über die Methode arriveAndDeregister() mitteilen, dass sie in dieser Phase angekommen sind, aber an den nachfolgenden Phasen nicht mehr teilnehmen wollen. Intern wird die Zahl der registrierten Parteien dekrementiert.  Die Zahl der noch erwarteten Parteien wird ebenfalls dekrementiert, weil die Partei angekommen ist, also nicht mehr erwartet wird.  Die Zahl der angekommenen Parteien bleibt wegen der gleichzeitigen Deregistrierung unverändert, denn die Summe der eingetroffenen und noch erwarteten Parteien muss immer gleich der Zahl der registrierten Parteien sein.

Wenn sich alle registrierten Parteien verabschiedet haben, d.h. die Zahl der registrierten Parteien auf Null sinkt, dann geht der Phaser in seinen Terminierungszustand.  In diesem Zustand ist die Phasennummer kleiner Null, also ungültig. Alle wartenden Methoden kehren bei Erreichen des Terminierungszustands zurück, weil die letzte Phase vorbei ist.
 

Eine einfache Benutzung des Phasers

Unter Verwendung einer CyclicBarrier sieht es so aus:

    public void split(List<Runnable> tasks, Runnable setup) {

        final CyclicBarrier barrier = new CyclicBarrier( tasks.size()+1 );

        // create and start all worker threads

        for (final Runnable task : tasks) {
            new Thread("Worker-Thread-") {
                public void run() {
                   barrier.await() ;   // wait for start signal
                    task.run();
                }
              }.start();
        }

        // perform setup and send start signal

        setup.run();
        barrier.await() ;   // send start signal
    }

Unter Verwendung eines Phaser s sieht es ganz ähnlich aus:

    public void split(List<Runnable> tasks, Runnable setup) {
        final Phaser phaser = new Phaser( tasks.size()+1 );

        // create and start all worker threads

        for (final Runnable task : tasks) {
            new Thread("Worker-Thread-") {
                public void run() {
                   phaser.arriveAndAwaitAdvance() ;  // wait for start signal
                    task.run();
                }
            }.start();
        }

        // perform setup and send start signal

        setup.run();
        phaser.arriveAndAwaitAdvance() ;   // send start signal
    }

Man kann es mit dem Phaser s auch ein wenig anders machen, was in diesem simplen Beispiel mit nur einer einzigen Phase und einer fixen Zahl von Beteiligten nicht nötig wäre.  Also, nur zur Demonstration der Phaser -spezifischen Methoden:

    public void split(List<Runnable> tasks, Runnable setup) {
        final Phaser phaser = new Phaser();

        phaser.register() ; // register preparing thread

        // create and start all worker threads

        for (final Runnable task : tasks) {
           phaser.register() ; // register worker thread
            new Thread("Worker-Thread-") {
                public void run() {
                    phaser.arriveAndAwaitAdvance();
                    task.run();
                }
            }.start();
        }

        // perform setup and send start signal

        setup.run();
        phaser.arrive() ;
    }

Wir haben dieses Mal das Start-Signal nicht mit arriveAndAwaitAdvance() gegeben, sondern mit arrive() .  Das bedeutet, dass der vorbereitende Thread nicht auf das Ende der Phase wartet.  Das ist auch gar nicht nötig, weil der vorbereitende Thread ohnehin der letzte ankommende Thread ist und damit derjenige, der die Phase beendet.

Außerdem haben wir die Zahl der Parteien nicht vorher berechnet und bei der Konstruktion des Phaser s angegeben, sondern wir haben erst einen leeren Phaser konstruiert und die beteiligten Parteien später nach und nach registriert.  Der Effekt ist der derselbe, aber man kann gut sehen, wie man beim Phaser eine vorher nicht bekannte Anzahl von Parteien registrieren würde.  Außerdem ist diese Form der Registrierung vom Programmierstil her besser, weil sie das Prinzip der Ownership respektiert.
 

Eine flexiblere Benutzung des Phasers

public class DirectoryTraverser {
 private static final ExecutorService pool = Executors.newFixedThreadPool(4);
 private static final AtomicInteger fileCount = new AtomicInteger();
 private static final Phaser phaser = new Phaser() ;

 private static class CounterTask implements Runnable {
  private final File theFile;

  public CounterTask(File file) {
   phaser.register() ;                // register task
   theFile = file;
  }
  public void run() {
   File[] files = theFile.listFiles(new FileFilter() {
    public boolean accept(File pathname) {
     if (pathname.isDirectory()) {
      pool.execute(new CounterTask(pathname));
      return true;
     }
     if (pathname.isFile()) {
      fileCount.getAndIncrement();
      return false;
     }
     return false;
    }
   });
   phaser.arriveAndDeregister() ;    // deregister task
  }
 }

 public static void main(String[] dirs) throws IOException {
  Path directory = Paths.get(dirs[0]);
  pool.execute(new CounterTask(directory.toFile()));
  phaser.awaitAdvance(0) ;
  pool.shutdown();
  System.out.println("contains: "+fileCount+" files");
 }
}

Die Terminierung des Phasers und die onAdvance() -Methode

Diese onAdvance() -Methode ähnelt der Barrier Action bei der CyclicBarrier : sie wird aufgerufen, wenn die Phase endet, und zwar von dem Thread, der als letzter am Treffpunkt ankommt und das Phasen-Ende auslöst.  Man kann die onAdvance() -Methode verwenden, um damit Nachverarbeitungen zu erledigen.  Ihr Returnwert hat darüber hinaus eine besondere Bedeutung: er steuert den Übergang in den Terminierungszustand.  Solange die onAdvance() -Methode false zurückliefert, gibt es weitere Phasen.  Wenn die onAdvance() -Methode true liefert, dann bedeutet es: „Dies war die letzte Phase.“ Dann geht der Phaser in den Terminierungszustand.

Die Default-Implementierung der onAdvance() -Methode sieht so aus:

     protected boolean onAdvance(int phase, int registeredParties) {
       return registeredParties == 0;
     }

Man kann die Terminierungsbedingung also nach Belieben anpassen.  Zum Beispiel könnte man dieses Mittel verwenden, um ein Reihe von Tasks mehrfach zu wiederholen.  Das könnte so aussehen:

void repeatTasks(List<Runnable> tasks, final int repetitions, boolean await_termination) {

        final Phaser phaser = new Phaser() {
            protected boolean onAdvance (int phase, int registeredParties) {
              return phase+1 >= repetitions || registeredParties == 0;
            }
        };

        // create and start worker threads
        for (final Runnable task : tasks) {
            new Thread() {
              { phaser. register (); }
              public void run() {
                while (!phaser. isTerminated ()) {
                  task.run();
                  phaser. arriveAndAwaitAdvance ();
                };
              }
            }.start();
        }
 
        if (await_termination) {
              phaseNumber = phaser. register ();
              while (!phaser. isTerminated ())
                  phaser. arriveAndAwaitAdvance ();
        }
}

Zusammenfassung

• Die Zahl der teilnehmenden Parteien kann sich beim Phaser dynamisch ändern; bei der CyclicBarrier ist sie fix.
• Das Warten auf das Eintreffen der registrierten Parteien ist flexibler.  Wer nicht am Phaser warten will, kann sein Eintreffen melden und sofort weitermachen. Bei der CyclicBarrier sind Ankunft und Warten untrennbar aneinander gekoppelt.
• Der Phaser terminiert und der Trigger für die Terminierung kann flexibel angepasst werden.  Die CyclicBarrier hat keinen Terminierungszustand.

Die gesamte Serie über Java 7:

If you are interested to hear more about this and related topics you might want to check out the following seminar:

Seminar   Effective Java - Advanced Java Programming Idioms  4 day seminar ( open enrollment and on-site) Concurrent Java - Java Multithread Programming 4 day seminar ( open enrollment and on-site)