9.9 Synchronisation über Warten und Benachrichtigen 

Die Synchronisation von Methoden oder Blöcken ist eine einfache Möglichkeit, konkurrierende Zugriffe von der virtuellen Maschine auflösen zu lassen. Obwohl die Umsetzung mit den Locks die Programmierung einfach macht, reicht dies für viele Aufgabenstellungen nicht aus. Wir können zwar Daten austauschen, doch gewünscht ist dieser Austausch in einer synchronisierten Abfolge. Gerne möchte ein Thread das Ankommen von Informationen signalisieren. Andere Threads wollen informiert werden, wenn Daten ankommen.

In Java wird dies durch die speziellen Objektmethoden wait() und notify() realisiert. Sie sind in der Klasse Object definiert (also hat sie jedes Objekt) und werden vom Thread aufgerufen, der gerade den Lock besitzt. Den kann er nur dann besitzen, wenn er sich in einem synchronisierten Block aufhält.

Hier klicken, um das Bild zu Vergrößern

Wenn wir den Lock am Objekt o festmachen und warten, dann schreiben wir:

  synchronized( o )  
{
  try {
    o.wait();
  } catch ( InterruptedException e ) {}
}

Ein wait() kann mit einer InterruptedException vorzeitig abbrechen, wenn der wartende Thread per interrupt()-Methode unterbrochen wird.

Wenn der aktuelle Thread das Lock eines Objekts besitzt, kann er Threads aufwecken, die auf dieses Objekt warten. Er bekommt den Lock durch das Synchronisieren der Methode, was ja bei Objektmethoden synchronized(this) entspricht.

synchronized void benachrichtige()
{
   notifyAll();
}

Um notify() muss es keinen eine Exception auffangenden Block geben.

9.9.1 Warten mit wait() und Aufwecken mit notify() 

Wartet ein Thread auf eine Nachricht, dann versetzt er sich mit wait() in eine Art Trance. Er geht in den Zustand nicht ausführend über und gibt für diesen Zeitraum seinen Monitor frei. Würde der wait() aufrufende Thread den Monitor nicht freigeben, könnte er auch nicht von anderen Threads reserviert werden. Somit könnte kein anderer Thread synchronisierten Programmcode ausführen, da ja der Monitor belegt ist und normalerweise sichergestellt sein muss, dass sich nur ein Thread im kritischen Abschnitt befindet. Doch in dem speziellen Fall kann nach dem wait() ein anderer Thread in eine synchronisierte Methode oder einen synchronisierten Block eintreten, die (beziehungsweise der) über den Monitor des Objekts synchronisiert. Dort kann der wartende Thread (es können auch mehrere sein) aufgeweckt werden. Dazu ruft der weckende notify() auf. Das heißt, die Methoden kommen immer paarweise vor: wo es ein wait() gibt, da darf auch ein notify() nicht fehlen. Wenn wir ein Programm mit nur einem Thread haben, dann macht natürlich so ein Pärchen keinen Sinn.

Nach einem notify() sollten wir den synchronisierten Abschnitt beenden und den Lock wieder freigegeben. Der wartende Thread wird nun wieder aufgeweckt, bekommt seinen Monitor zurück und kann weiterarbeiten. Wenn mehrere Threads gewartet haben, dann wählt ein Algorithmus zufällig einen Thread aus und weckt ihn auf.

Daten liefern und auf Daten warten

Szenarien mit wait() und notify() sind oft Produzenten-Konsumenten-Beispiele. Ein Thread liefert Daten, die ein anderer Thread verwenden möchte. Da er nicht in einer kostspieligen Schleife auf die Information warten soll, synchronisieren sich die Partner über ein beiden bekanntes Objekt. Erst dann, wenn der Produzent sein OK gegeben hat, macht es für den Datennutzer Sinn weiterzuarbeiten; jetzt hat er seine benötigten Daten. So wird keine unnötige Zeit in Warteschleifen vergeudet, und der Prozessor kann die übrige Zeit anderen Threads zuteilen.

synchronized( o )
{
    o.wait();  
  // Habe gewartet, kann jetzt loslegen.
}

synchronized( o )
{
  // Habe etwas gemacht und informiere jetzt meinen Wartenden.
    o.notify();  
}

Wir werden gleich noch ein umfassenderes Beispiel für das Konsumenten-Produzenten-Paar anführen.

void notify() Weckt einen beliebigen Thread auf, der an diesem Objekt wartet.

void wait() throws InterruptedException Der aktuelle Thread wartet an dem aufrufenden Objekt darauf, dass er nach einem notify()/notifyAll() weiterarbeiten kann. Der aktive Thread muss natürlich den Monitor des Objekts belegt haben. Andernfalls kommt es zu einer IllegalMonitorStateException.

9.9.2 Falls der Lock fehlt: IllegalMonitorStateException 

Wenn wait() oder notify() aufgerufen werden, uns aber der entsprechende Lock für das Objekt fehlt, kommt es zu einem Laufzeitfehler. Das Problem ist mit einem synchronized-Block zu lösen. Wir wollen uns ein paar Situationen anschauen und eine Korrektur versuchen. Was wird bei folgendem Programm passieren?

Listing 9.13   NotOwner.java

class NotOwner
{
  public static void main( String args[] ) throws InterruptedException
  {
    new NotOwner().wait();
  }
}

Der Compiler kann das Programm übersetzen, doch zur Laufzeit wird es zu einem Fehler kommen:

java.lang.IllegalMonitorStateException: current thread not owner
 at java.lang.Object.wait(Native Method)
 at java.lang.Object.wait(Object.java:426)
 at NotOwner.main(NotOwner.java:5)
Exception in thread "main"

Der Fehler zeigt an, dass der aktuelle ausführende Thread (current thread) nicht den nötigen Lock besitzt, um wait() auszuführen. Den bekommen wir aber über ein synchronized. Um den Fehler zu beheben, setzen wir:

NotOwner o = new NotOwner();

synchronized( o )
{
  o.wait();
}

Das zeigt, dass das Objekt o, welches den Lock besitzt, für ein wait() »bereit« sein muss. In die richtige Stimmung wird es nur mit synchronized gebracht.

synchronized( NotOwner.class )
{
  new NotOwner().wait();
}

Doch natürlich könnten wir auch am Klassenobjekt synchronisieren:

synchronized( NotOwner.class )
{
  NotOwner.class.wait();
}

9.9.3 Mehrere Wartende und notifyAll() 

Es kann durchaus vorkommen, dass mehrere Threads in einer Warteposition an demselben Objekt sind und aufgeweckt werden wollen. Mit notify() lässt sich ein willkürlich ausgewählter Wartender aufwecken. Den anderen Threads ist damit allerdings noch nicht geholfen. Damit sich jetzt nicht jeder Wartende in eine Liste eintragen muss – und wir jeden einzelnen mit notify() aufwecken müssen – bietet jedes Objekt auch die Methode notifyAll() an. Damit werden alle an diesem Objekt wartenden Threads informiert und können weitermachen.

void notifyAll() Benachrichtigt alle Threads, die auf diesem Objekt warten.

9.9.4 wait() mit einer Zeitspanne 

Ein wait() wartet im schlechtesten Fall bis zum Nimmerleinstag, wenn es kein notify() gibt. Es gibt jedoch Situationen, in denen wir eine gewisse Zeit warten, aber bei Fehlen der Benachrichtigung weitermachen wollen. Dazu kann dem wait() eine Zeit mitgegeben werden.

void wait( long timeout ) throws InterruptedException Wartet auf ein notify()/notifyAll() eine gegebene Anzahl von Millisekunden. Nach der abgelaufenen Zeit geht es ohne Fehler weiter.

void wait( long timeout, int nanos ) throws InterruptedException Wartet auf ein notify()/notifyAll() angenähert 1.000.000 * timeout + nanos Nano-Sekunden.

o1.wait( 2000 );
o2.wait();

9.9.5 Beispiel Erzeuger-Verbraucher-Programm 

Ein kleines Erzeuger-Verbraucher-Programm soll die Anwendung von Threads kurz demonstrieren. Zwei Threads greifen auf eine gemeinsame Datenbasis zurück. Ein Thread produziert unentwegt Daten (in dem Beispiel ein Zeit-Datum) und schreibt diese in einen Vektor. Der andere Thread nimmt Daten aus dem Vektor heraus und schreibt diese auf den Bildschirm.

Hier klicken, um das Bild zu Vergrößern

Listing 9.14   ErzeugerVerbraucherDemo.java, Teil 1

import java.util.LinkedList;
import java.util.Queue;

class Erzeuger extends Thread
{
  private final static int MAXQUEUE = 13;

  private Queue<String> nachrichten = new LinkedList<String>();

  public void run()
  {
    try
    {
      while ( true )
      {
        erzeuge();
        sleep( (int)(Math.random()*1000) );
      }
    } catch ( InterruptedException e ) { }
  }

  public synchronized void erzeuge()
    throws InterruptedException
  {
    while ( nachrichten.size() == MAXQUEUE )
      wait();

    nachrichten.add( new java.util.Date().toString() );
    notifyAll();    // oder notify();
  }

  // vom Verbraucher aufgerufen
  public synchronized String verbrauche()
    throws InterruptedException
  {
    while ( nachrichten.size() == 0 )
      wait();

    notifyAll();

    return nachrichten.poll();
  }
}

Die gesamte Klasse Erzeuger erweitert Thread. Als Objektvariable wird eine Warteschlange als Objekt vom Typ Queue definiert, das die Daten aufnimmt, auf die die Threads dann zurückgreifen. Die erste definierte Funktion ist erzeuge(). Wenn noch Platz in der Warteschlange ist, dann hängt die Funktion das Erstellungsdatum an. Anschließend informiert der Erzeuger über notifyAll() alle eventuell wartenden Threads.

Die Verbraucher nutzen die Funktion verbrauche(). Sind in der Warteschlange keine Daten vorhanden, so wartet der Thread durch wait(). Wichtig ist der Programmblock in der Schleife zum Holen der Nachricht:

while ( nachrichten.size() == MAXQUEUE )
  wait();

Es ist typisch für Wartesituationen, dass wait() in einem Schleifenrumpf aufgerufen wird. Denn falls ein notifyAll() aus dem wait() erlöst, kann gleichzeitig auch ein anderer Thread fertig werden und aus der Schleife herauskommen. Das ist der Fall, wenn gleichzeitig Threads auf das Ankommen neuer Güter warten. Ein einfaches if würde dazu führen, dass einer der beiden ein erzeugtes Gut entnimmt und der andere Verbraucher dann keins mehr bekommen kann. Die Schleifenbedingung ist das Gegenteil der Bedingung, auf die gewartet werden soll.

Der Quellcode der Klasse Verbraucher ist kleiner.

class Verbraucher extends Thread
{
  Erzeuger erzeuger;
  String   name;

  Verbraucher( String name, Erzeuger erzeuger )
  {
    this.erzeuger = erzeuger;
    this.name = name;
  }

  public void run()
  {
    try
    {
      while ( true )
      {
        String info = erzeuger.verbrauche();

        System.out.println( name +" holt Nachricht: "+ info );

        Thread.sleep( (int)(Math.random()*1000) );
      }
    } catch ( InterruptedException e ) { }
  }
}

Das Hauptprogramm in ErzeugerVerbraucherDemo konstruiert einen Erzeuger und drei Verbraucher. Wir übergeben im Konstruktor den Erzeuger, und dann holen sich die Verbraucher mittels verbrauche() selbstständig die Daten ab.

public class ErzeugerVerbraucherDemo
{
  public static void main( String args[] )
  {
    Erzeuger erzeuger = new Erzeuger();
    erzeuger.start();

    new Verbraucher( "Eins", erzeuger ).start();
    new Verbraucher( "Zwei", erzeuger ).start();
    new Verbraucher( "Drei", erzeuger ).start();
  }
}

Es ergibt sich eine Ausgabe ähnlich dieser:

Eins holt Nachricht: Tue May 25 14:57:19 CEST 2004
Zwei holt Nachricht: Tue May 25 14:57:20 CEST 2004
Drei holt Nachricht: Tue May 25 14:57:20 CEST 2004
Eins holt Nachricht: Tue May 25 14:57:21 CEST 2004
Zwei holt Nachricht: Tue May 25 14:57:22 CEST 2004
... bis in die Unendlichkeit

9.9.6 Semaphoren 

Wir haben gesehen, dass sich Synchronisationsprobleme durch kritische Abschnitte und Wartesituationen mit wait() und notify() lösen lassen. Dennoch ist der eingebaute Mechanismus auch mit Nachteilen verbunden. Denn eine große Schwierigkeit ist es, synchronisierende Programme zu entwickeln und zu warten. Die Synchronisationsvariablen verstreuen sich mitunter über große Programmteile und machen die Wartung schwierig. Teile, die bewusst atomar ausgeführt werden müssen, benötigen zwingend einen Programmblock und eine Synchronisationsvariable. Und das heißt für uns Entwickler, dass wir einen vorher einfachen Block durch wait() und notify() ersetzen müssen, der synchronisiert ist. Und wir müssen uns um eine Variable kümmern. Das ist unangenehm, und wir wünschen uns ein einfacheres Konzept, so dass eine Umstellung leicht ist. Hier bieten sich Funktionsaufrufe an. Es ist schön, die Wartesituation hinter einem Paar von Funktionen wie enter() und leave() zu verstecken.

Die Idee für diese Realisierung kommt von dem niederländischen¹  Informatiker Edsger Wybe Dijkstra²  . Neben vielen anderen Problemen aus der Informatik beschäftigte er sich mit der Wahl der kürzesten Wege und mit der Synchronisation von Prozessen. Zur damaligen Zeit wurde Parallelität noch durch Variablen und Warteschleifen realisiert, Programmiersprachen mit höheren Konzepten, wie sie Java bietet, waren nicht kommerziell verbreitet. Dijkstra schlug einen Satz von Funktionen P() und V() vor, die das Eintreten und Verlassen in und aus einem atomaren Block umsetzen. Dijkstra assoziierte mit den Funktionsnamen die Wörter pass und vrij, was auf Niederländisch frei heißt. Er nahm zur Verdeutlichung ein Beispiel aus dem Eisenbahnverkehr. Dort darf sich nur ein Zug auf einem Streckenabschnitt befinden, und wenn ein weiterer Zug einfahren will, so muss er warten. Er kann dann weiterfahren, wenn der erste Zug die Strecke verlassen hat. Wir erkennen hier sofort einen kritischen Abschnitt wieder, den wir in Java mit synchronized schützen würden. Da für uns P() und V() nicht so intuitiv ist und wir keine Eisenbahner sind, verwenden wir für das Eintreten die Methode enter() und für das Austreten leave(). Verglichen mit einer Sperre, hätten wir die Funktionen auch lock() und unlock() nennen können.

Der Datentyp, der diese beiden Funktionen jetzt implementiert, nennt sich Semaphore. In ihm ist intern auch noch die Synchronisationsvariable versteckt, doch dies bleibt für uns als Nutzer natürlich unsichtbar. Die Klasse lässt sich in Java leicht realisieren.

Listing 9.15   Semaphore.java

public class Semaphore
{
  private int cnt = 1;

  public Semaphore( int cnt )
  {
    if ( cnt > 1 )  // nur Zähler größer 1 akzeptieren
      this.cnt = cnt;
  }

  public synchronized void   enter()    // P
  {
    while ( cnt <= 0 )
      try {
        wait();
      } catch( InterruptedException e ) {}

    cnt--;
  }

  public synchronized void   leave()    // V
  {
    cnt++;
//    if ( cnt > 0 )
    notify();
  }
}

Um zu kontrollieren, wie viele Threads auf ein Programmstück zugreifen, wird die Semaphore verwendet. Unterschieden werden zwei Arten:

Binäre Semaphoren lassen höchstens einen Thread auf ein Programmstück zu.

Allgemeine Semaphoren lassen eine bestimmte begrenzte Menge an Threads in einen kritischen Abschnitt.

Eine binäre Semaphore wird mit dem klassischem wait() und notify() realisiert. Allgemeine Semaphoren vereinfachen das Konsumenten-Produzenten-Problem. Die verbleibende Größe des Puffers ist damit auch automatisch die maximale Anzahl von Produzenten, die sich parallel im Einfügeblock befinden können.

java.util.concurrent.Semaphore

Die Klasse java.util.concurrent.Semaphore seit Version 5 implementiert das Konzept einer Semaphore. Im Konstruktor bestimmt eine Ganzzahl die Anzahl Threads, die einen Programmteil ausführen können. Der zweite Parameter im Konstruktor gibt an, ob der am längsten Wartende der nächste ist, oder nicht. Das wird Fairness genannt.

¹   Holland ist im Übrigen nur eine Provinz der Niederlande.

²   Einige Infos über ihn unter http://henson.cc.kzoo.edu/~k98mn01/dijkstra.html.