9.12 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.

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. Während des Wartens wird der Monitor des Objekts freigegeben, damit benachrichtigende Threads auf dieses Objekt synchronisieren können.

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.12.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 einer 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.

9.12.2 Mehrere Wartende und notifyAll() 

Wir haben gerade gesehen, dass es vorkommen kann, 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.

9.12.3 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 trotzdem weitermachen wollen. In Java kann dazu dem wait() eine Zeit in Millisekunden als Parameter übergeben werden.

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

Die Synchronisationsblöcke müssen bei einem lauffähigen Beispiel noch hinzugefügt werden.

9.12.4 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), schreibt diese in einen Vektor, und der andere Thread nimmt Daten aus dem Vektor heraus und schreibt diese auf den Bildschirm.

void notify() Weckt einen beliebigen Thread auf, der an diesem Objekt wartet. Mit Threads, die auf den Monitor des Objekts warten (wegen synchronized), hat notify() nichts zu tun.

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

Hier kommt nun das Erzeuger-Verbraucher-Programm:

Listing 9.15   ErzeugerVerbraucherDemo.java

import java.util.Vector;

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

  private Vector   nachrichten = new Vector();

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

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

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

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

    nachrichten.removeElement( info );

    return info;
  }
}

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

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

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


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

Die gesamte Klasse Erzeuger erweitert Thread. Als Objektvariable wird eine Warteschlange als Objekt der Klasse Vector definiert, das die Daten aufnimmt, auf die die Threads dann zurückgreifen. Die erste definierte Funktion ist sendeNachricht(). 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 nutzten die Funktion getMessage(). Sind in der Warteschlange keine Daten vorhanden, so wartet der Thread durch wait(). Dieses wird nur dann unterbrochen, wenn ein notifyAll() vom Erzeuger kommt. Die Klasse Verbraucher implementiert run(), das eine Nachricht holt und eine Sekunde wartet.

Wichtig ist der Programmblock in der Schleife:

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

Das 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.

Das Hauptprogramm in ErzeugerVerbraucherDemo konstruiert einen Erzeuger und drei Verbraucher. Wir übergeben im Konstruktor den Erzeuger und dann holen sich die Verbraucher mittels getMessage() selbstständig die Daten ab. Es ergibt sich eine Ausgabe ähnlich dieser:

Eins holt Nachricht: Thu Sep 20 13:45:56 CEST 2001
Zwei holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
Drei holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
Drei holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
Drei holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
Eins holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
Zwei holt Nachricht: Thu Sep 20 13:45:57 CEST 2001
... bis in die Unendlichkeit

9.12.5 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 versehen. Denn eine große Schwierigkeit ist es, synchronisierende Programme zu entwickeln und zu warten. Die Synchronisationsvariablen verstreuen sich mitunter über große Programmteile und sind leider schwer verständlich. Teile, die bewusst atomar ausgeführt werden müssen, benötigen zwingend einen Programmblock und eine Variable. Und das heißt für uns Entwickler, dass wir einen vorher einfachen Block jetzt 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, sodass 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 auch 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. Da für uns P() und V() nicht so intuitiv ist und wir keine Eisenbahner sind, verwenden wir die Methode enter() für das Eintreten und leave() für das Austreten. 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.16   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();
  }
}

Die hier gewählte Implementierung lässt zwei Semaphoren zu. So genannte binäre Semaphoren, die unserem klassischem wait() und notify() entsprechen, sowie allgemeine Semaphoren, die eine bestimmte begrenzte Menge an Threads in einen kritischen Abschnitt lassen. Letzteres vereinfacht 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. Schön ist, dass die allgemeine Semaphore mit mehreren Wartenden letztendlich in den kritischen Abschnitt auf eine binäre Semaphore abgebildet wird, die ja nichts anderes als ein einzelnes wait() und notify() ist.

Es könnte gut sein, dass in der nächsten Java-Version eine Semaphoren-Klasse eingeführt wird. Im Java Community Process beschreibt die JSR 166, Concurrency Utilities (http://www.jcp.org/jsr/detail/166.jsp), eine Erweiterung der Thread-Mechanismen.

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

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

³   Bedauerlicherweise erleben wir zu oft das Gegenteil.