Einführung in die nebenläufige Programmierung mit Threads

42

Einführung in nebenläufige Programmierung

Traditionelle Programmierung

• Aufruf einer Methode verlagert Kontrollfluss in diese Methode
• Code hinter Methodenaufruf wird erst nach Beendigung der Methode ausgeführt

public class Traditional {
    public static void main(String... args) {
        Traditional x = new Traditional();

        System.out.println("main(): vor run()");
        x.run();
        System.out.println("main(): nach run()");
    }

    public void run() {
        IntStream.range(0, 10).mapToObj(i -> "in run()").forEach(System.out::println);
    }
}

Nebenläufige Programmierung

• Erzeugung eines neuen Kontrollflussfadens (Thread)
  • Läuft (quasi-) parallel zu bisherigem Kontrollfluss
• Threads können unabhängig von einander arbeiten
• Zustandsverwaltung durch Java-VM (oder Unterstützung durch Betriebssystem)
  • Aufruf einer bestimmten Methode erzeugt neuen Kontrollflussfaden
  • Der neue Thread arbeitet "parallel" zum bisherigen Thread
  • Kontrolle kehrt sofort wieder zurück: Code hinter dem Methodenaufruf wird ausgeführt ohne auf die Beendigung der aufgerufenen Methode zu warten
  • Verteilung der Threads auf die vorhandenen Prozessorkerne abhängig von der Java-VM

public class Threaded extends Thread {
    public static void main(String... args) {
        Threaded x = new Threaded();

        System.out.println("main(): vor run()");
        x.start();
        System.out.println("main(): nach run()");
    }

    @Override
    public void run() {
        IntStream.range(0, 10).mapToObj(i -> "in run()").forEach(System.out::println);
    }
}

Erzeugen von Threads

• Ableiten von Thread oder Implementierung von Runnable

  

• Methode run() implementieren, aber nicht aufrufen

• Methode start() aufrufen, aber (i.d.R.) nicht implementieren

Ableiten von Thread

• start() startet den Thread und sorgt für Ausführung von run()
• start() nur einmal aufrufen

Implementierung von Runnable

• Ebenfalls run() implementieren
• Neues Thread-Objekt erzeugen, Konstruktor das eigene Runnable übergeben
• Für Thread-Objekt die Methode start() aufrufen
  • Startet den Thread (das Runnable) und sorgt für Ausführung von run()

Vorteil von Runnable: Ist ein Interface, d.h. man kann noch von einer anderen Klasse erben

Zustandsmodell von Threads (vereinfacht)

Threads haben einen Lebenszyklus: Nach dem Erzeugen der Objekte mit new wird der Thread noch nicht ausgeführt. Er ist sozusagen in einem Zustand "erzeugt". Man kann bereits mit dem Objekt interagieren, also auf Attribute zugreifen und Methoden aufrufen.

Durch den Aufruf der Methode start() gelangt der Thread in einen Zustand "ausführungsbereit", er läuft also aus Nutzersicht. Allerdings hat er noch keine Ressourcen zugeteilt (CPU, ...), so dass er tatsächlich noch nicht rechnet. Sobald er vom Scheduler eine Zeitscheibe zugeteilt bekommt, wechselt er in den Zustand "rechnend" und führt den Inhalt der run()-Methode aus. Von hier kann er nach Ablauf der Zeitscheibe durch den Scheduler wieder nach "ausführungsbereit" zurück überführt werden. Dieses Wechselspiel passiert automatisch und i.d.R. schnell, so dass selbst auf Maschinen mit nur einem Prozessor/Kern der Eindruck einer parallelen Verarbeitung entsteht.

Nach der Abarbeitung der run()-Methode oder bei einer nicht gefangenen Exception wird der Thread beendet und kann nicht wieder neu gestartet werden. Auch wenn der Thread abgelaufen ist, kann man mit dem Objekt wie üblich interagieren (nur eben nicht mehr parallel).

Bei Zugriff auf gesperrte Ressourcen oder durch Aufrufe von Methoden wie sleep() oder join() kann ein Thread blockiert werden. Hier führt der Thread nichts aus, bekommt durch den Scheduler aber auch keine neue Zeitscheibe zugewiesen. Aus diesem Zustand gelangt der Thread wieder heraus, etwa durch Interrupts (Aufruf der Methode interrupt() auf dem Thread-Objekt) oder nach Ablauf der Schlafzeit (in sleep()) oder durch ein notify, und wird wieder zurück nach "ausführungsbereit" versetzt und wartet auf die Zuteilung einer Zeitscheibe durch den Scheduler.

Sie finden in [Boles2008, Kapitel 5.2 "Thread-Zustände"] eine schöne ausführliche Darstellung.

Threads können wie normale Objekte kommunizieren

• Zugriff auf (public) Attribute (oder eben über Methoden)
• Aufruf von Methoden

Threads können noch mehr

• Eine Zeitlang schlafen: Thread.sleep(<duration_ms>)

  • Statische Methode der Klasse Thread (Klassenmethode)
  • Aufrufender Thread wird bis zum Ablauf der Zeit oder bis zum Aufruf der interrupt()-Methode des Threads blockiert
  • "Moderne" Alternative: TimeUnit, beispielsweise TimeUnit.SECONDS.sleep( 2 );

• Prozessor abgeben und hinten in Warteschlange einreihen: yield()

• Andere Threads stören: otherThreadObj.interrupt()

  • Die Methoden sleep(), wait() und join() im empfangenden Thread otherThreadObj lösen eine InterruptedException aus, wenn sie durch die Methode interrupt() unterbrochen werden. Das heißt, interrupt() beendet diese Methoden mit der Ausnahme.
  • Empfangender Thread verlässt ggf. den Zustand "blockiert" und wechselt in den Zustand "ausführungsbereit"

• Warten auf das Ende anderer Threads: otherThreadObj.join()

  • Ausführender Thread wird blockiert (also nicht otherThreadObj!)
  • Blockade des Aufrufers wird beendet, wenn der andere Thread (otherThreadObj) beendet wird.

Hinweis: Ein Thread wird beendet, wenn

• die run()-Methode normal endet, oder
• die run()-Methode durch eine nicht gefangene Exception beendet wird, oder
• von außen die Methode stop() aufgerufen wird (Achtung: Deprecated! Einen richtigen Ersatz gibt es aber auch nicht.).

Hinweis: Die Methoden wait(), notify()/notifyAll() und die "synchronized-Sperre" werden in der Sitzung ["Threads: Synchronisation"](threads-intro. besprochen.

Wrap-Up

Threads sind weitere Kontrollflussfäden, von Java-VM (oder (selten) von OS) verwaltet

• Ableiten von Thread oder implementieren von Runnable
• Methode run enthält den auszuführenden Code
• Starten des Threads mit start (nie mit run!)

Synchronisation: Verteilter Zugriff auf gemeinsame Ressourcen

Motivation: Verteilter Zugriff auf gemeinsame Ressourcen

public class Teaser implements Runnable {
    private int val = 0;

    public static void main(String... args) {
        Teaser x = new Teaser();
        new Thread(x).start();
        new Thread(x).start();
    }

    private void incrVal() {
        ++val;
        System.out.println(Thread.currentThread().getId() + ": " + val);
    }

    public void run() {
        IntStream.range(0, 5).forEach(i -> incrVal());
    }
}

Zugriff auf gemeinsame Ressourcen: Mehrseitige Synchronisierung

synchronized (<Object reference>) {
    <statements (synchronized)>
}

=> "Mehrseitige Synchronisierung"

Fallunterscheidung: Thread T1 führt synchronized-Anweisung aus:

• Sperre im Sperr-Objekt nicht gesetzt:
  1. T1 setzt Sperre beim Eintritt,
  2. führt den Block aus, und
  3. löst Sperre beim Verlassen
• Sperre durch T1 gesetzt:
  1. T1 führt den Block aus, und
  2. löst Sperre beim Verlassen nicht
• Sperre durch T2 gesetzt: => T1 wird blockiert, bis T2 die Sperre löst

Anmerkung: Das für die Synchronisierung genutzte Objekt nennt man "Wächter-Objekt" oder auch "Sperr-Objekt" oder auch "Synchronisations-Objekt".

Damit könnte man den relevanten Teil der Methode incrVal() beispielsweise in einen geschützten Bereich einschließen und als Sperr-Objekt das eigene Objekt (this) einsetzen:

    private void incrVal() {
        synchronized (this) { ++val; }
    }

Synchronisierte Methoden

void f() {
    synchronized (this) {
        ...
    }
}

synchronized void f() {
    ...
}

Kurzschreibweise: Man spart das separate Wächter-Objekt und synchronisiert auf sich selbst ...

Die Methode incrVal() könnte entsprechend so umgeschrieben werden:

    private synchronized void incrVal() {
        ++val;
    }

Probleme bei der (mehrseitigen) Synchronisierung: Deadlocks

public class Deadlock {
    private final String name;

    public synchronized String getName() { return name; }
    public synchronized void foo(Deadlock other) {
        System.out.format("%s: %s.foo() \n", Thread.currentThread().getName(), name);
        System.out.format("%s: %s.name()\n", Thread.currentThread().getName(), other.getName());
    }

    public static void main(String... args) {
        final Deadlock a = new Deadlock("a");
        final Deadlock b = new Deadlock("b");

        new Thread(() -> a.foo(b)).start();
        new Thread(() -> b.foo(a)).start();
    }
}

Viel hilft hier nicht viel! Durch zu großzügige mehrseitige Synchronisierung kann es passieren, dass Threads gegenseitig aufeinander warten: Thread A belegt eine Ressource, die ein anderer Thread B haben möchte und Thread B belegt eine Ressource, die A gerne bekommen würde. Da es dann nicht weitergeht, nennt man diese Situation auch "Deadlock" ("Verklemmung").

Im Beispiel ruft der erste Thread für das Objekt a die foo()-Methode auf und holt sich damit den Lock auf a. Um die Methode beenden zu können, muss noch die getName()-Methode vom Objekt b durch diesen ersten Thread aufgerufen werden. Dafür muss der erste Thread den Lock auf b bekommen.

Dummerweise hat parallel der zweite Thread auf dem Objekt b die foo()-Methode aufgerufen und sich damit den Lock auf b geholt. Damit muss der erste Thread so lange warten, bis der zweite Thread den Lock auf b freigibt.

Das wird allerdings nicht passieren, da der zweite Thread zur Beendigung der foo()-Methode noch getName() auf a ausführen muss und dazu den Lock auf b holen, den aber aktuell der erste Thread hält.

Und schon geht's nicht mehr weiter :-)

Warten auf andere Threads: Einseitige Synchronisierung

Problem

• Thread T1 wartet auf Arbeitsergebnis von T2
• T2 ist noch nicht fertig

Mögliche Lösungen

1. Aktives Warten (Polling): Permanente Abfrage
   • Kostet unnötig Rechenzeit
2. Schlafen mit Thread.sleep()
   • Etwas besser; aber wie lange soll man idealerweise schlafen?
3. Warten mit T2.join()
   • Macht nur Sinn, wenn T1 auf das Ende von T2 wartet
4. Einseitige Synchronisierung mit wait() und notify()
   • Das ist DIE Lösung für das Problem :)

Einseitige Synchronisierung mit wait und notify

• wait: Warten auf Erfüllung einer Bedingung (Thread blockiert):

  synchronized (obj) {    // Geschützten Bereich betreten
      while (!condition) {
          try {
              obj.wait(); // Thread wird blockiert
          } catch (InterruptedException e) {}
      }
      ...     // Condition erfüllt: Tue Deine Arbeit
  }

  => Bedingung nach Rückkehr von wait erneut prüfen!

Eigenschaften von wait

• Thread ruft auf Synchronisations-Objekt die Methode wait auf
• Prozessor wird entzogen, Thread blockiert
• Thread wird in interne Warteschlange des Synchronisations-Objekts eingetragen
• Sperre auf Synchronisations-Objekt wird freigegeben

=> Geht nur innerhalb der synchronized-Anweisung für das Synchronisations-Objekt!

Einseitige Synchronisierung mit wait und notify (cnt.)

• notify: Aufwecken von wartenden (blockierten) Threads:

  synchronized (obj) {
      obj.notify();       // einen Thread "in" obj aufwecken
      obj.notifyAll();    // alle Threads "in" obj wecken
  }

Eigenschaften von notify bzw. notifyAll

• Thread ruft auf einem Synchronisations-Objekt die Methode notify oder notifyAll auf
• Falls Thread(s) in Warteschlange des Objekts vorhanden, dann
  • notify: Ein zufälliger Thread wird aus Warteschlange entfernt und in den Zustand "ausführungsbereit" versetzt
  • notifyAll: Alle Threads werden aus Warteschlange entfernt und in den Zustand "ausführungsbereit" versetzt

=> Geht nur innerhalb der synchronized-Anweisung für das Synchronisations-Objekt!

Wrap-Up

Synchronisierungsbedarf bei verteiltem Zugriff auf gemeinsame Ressourcen:

• Vorsicht mit konkurrierendem Ressourcenzugriff: Synchronisieren mit synchronized => Mehrseitige Synchronisierung

• Warten auf Ereignisse mit wait und notify/notifyAll => Einseitige Synchronisierung

High-Level Concurrency

Explizite Lock-Objekte

Sie kennen bereits die Synchronisierung mit dem Schlüsselwort synchronized.

// Synchronisierung der gesamten Methode
public synchronized int incrVal() {
    ...
}

// Synchronisierung eines Blocks (eines Teils einer Methode)
public int incrVal() {
    ...
    synchronized (someObj) {
        ...
    }
    ...
}

Dabei wird implizit ein Lock über ein Objekt (das eigene Objekt im ersten Fall, das Sperrobjekt im zweiten Fall) benutzt.

Seit Java5 kann man alternativ auch explizite Lock-Objekte nutzen:

// Synchronisierung eines Teils einer Methode über ein
// Lock-Objekt (seit Java 5)
// Package `java.util.concurrent.locks`
public int incrVal() {
    Lock waechter = new ReentrantLock();
    ...
    waechter.lock();
    ... // Geschützter Bereich
    waechter.unlock();
    ...
}

Locks aus dem Paket java.util.concurrent.locks arbeiten analog zum impliziten Locken über synchronized. Sie haben darüber hinaus aber einige Vorteile:

• Methoden zum Abfragen, ob ein Lock möglich ist: Lock#tryLock
• Methoden zum Abfragen der aktuellen Warteschlangengröße: Lock#getQueueLength
• Verfeinerung ReentrantReadWriteLock mit Methoden readLock und writeLock
  • Locks nur zum Lesen bzw. nur zum Schreiben
• Lock#newCondition liefert ein Condition-Objekt zur Benachrichtigung ala wait/notify: await/signal => zusätzliches Timeout beim Warten möglich

Nachteile:

• Bei Exceptions werden implizite Locks durch synchronized automatisch durch das Verlassen der Methode freigegeben. Explizite Locks müssen durch den Programmierer freigegeben werden! => Nutzung des finally-Block!

Thread-Management: Executor-Interface und Thread-Pools

Wiederverwendung von Threads

• Normale Threads sind immer Einmal-Threads: Man kann sie nur einmal in ihrem Leben starten (auch wenn das Objekt anschließend noch auf Nachrichten bzw. Methodenaufrufe reagiert)

• Zusätzliches Problem: Threads sind Objekte:

  • Threads brauchen relativ viel Arbeitsspeicher
  • Erzeugen und Entsorgen von Threads kostet Ressourcen
  • Zu viele Threads: Gesamte Anwendung hält an

• Idee: Threads wiederverwenden und Thread-Management auslagern => Executor-Interface und Thread-Pool

Executor-Interface

public interface Executor {
    void execute(Runnable command);
}

• Neue Aufgaben als Runnable an einen Executor via execute übergeben
• Executor könnte damit sofort neuen Thread starten (oder alten wiederverwenden): e.execute(r); => entspricht in der Wirkung (new Thread(r)).start();

Thread-Pool hält Menge von "Worker-Threads"

• Statische Methoden von java.util.concurrent.Executors erzeugen Thread-Pools mit verschiedenen Eigenschaften:

  • Executors#newFixedThreadPool erzeugt ExecutorService mit spezifizierter Anzahl von Worker-Threads
  • Executors#newCachedThreadPool erzeugt Pool mit Threads, die nach 60 Sekunden Idle wieder entsorgt werden

• Rückgabe: ExecutorService (Thread-Pool)

  public interface ExecutorService extends Executor { ... }

• Executor#execute übergibt Runnable dem nächsten freien Worker-Thread (oder erzeugt ggf. neuen Worker-Thread bzw. hängt Runnable in Warteschlange, je nach erzeugtem Pool)

• Methoden zum Beenden eines Thread-Pools (Freigabe): shutdown(), isShutdown(), ...

MyThread x = new MyThread();    // Runnable oder Thread

ExecutorService pool = Executors.newCachedThreadPool();

pool.execute(x);    // x.start()
pool.execute(x);    // x.start()
pool.execute(x);    // x.start()

pool.shutdown();    // Feierabend :)

Hintergrund (vereinfacht)

Der Thread-Pool reserviert sich "nackten" Speicher, der der Größe von $n$ Threads entspricht, und "prägt" die Objektstruktur durch einen Cast direkt auf (ohne wirkliche neue Objekte zu erzeugen). Dieses Vorgehen ist in der C-Welt wohlbekannt und schnell (vgl. Thema Speicherverwaltung in der LV "Systemprogrammierung"). In Java wird dies durch eine wohldefinierte Schnittstelle vor dem Nutzer verborgen.

Ausblick

Hier haben wir nur die absoluten Grundlagen angerissen. Wir können auch Callables anstatt von Runnables übergeben, auf Ergebnisse aus der Zukunft warten (Futures), Dinge zeitgesteuert (immer wieder) starten, ...

Schauen Sie sich bei Interesse die weiterführende Literatur an, beispielsweise die Oracle-Dokumentation oder auch [Ullenboom2021] (insbesondere den Abschnitt 16.4 "Der Ausführer (Executor) kommt").

Fork/Join-Framework: Teile und Herrsche

Spezieller Thread-Pool zur rekursiven Bearbeitung parallelisierbarer Tasks

• java.util.concurrent.ForkJoinPool#invoke startet Task

• Task muss von RecursiveTask<V> erben:

  public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
      protected abstract V compute();
  }

Prinzipieller Ablauf:

public class RecursiveTask extends ForkJoinTask<V> {
    protected V compute() {
        if (task klein genug) {
            berechne task sequentiell
        } else {
            teile task in zwei subtasks:
                left, right = new RecursiveTask(task)
            rufe compute() auf beiden subtasks auf:
                left.fork();          // starte neuen Thread
                r = right.compute();  // nutze aktuellen Thread
            warte auf ende der beiden subtasks: l = left.join()
            kombiniere die ergebnisse der beiden subtasks: l+r
        }
    }
}

Swing und Threads

Lange Berechnungen in Listenern blockieren Swing-GUI

• Problem: Events werden durch einen Event Dispatch Thread (EDT) sequentiell bearbeitet
• Lösung: Berechnungen in neuen Thread auslagern
• Achtung: Swing ist nicht Thread-safe! Komponenten nicht durch verschiedene Threads manipulieren!

Lösung

=> javax.swing.SwingWorker ist eine spezielle Thread-Klasse, eng mit Swing/Event-Modell verzahnt.

• Implementieren:

  • SwingWorker#doInBackground: Für die langwierige Berechnung (muss man selbst implementieren)
  • SwingWorker#done: Wird vom EDT aufgerufen, wenn doInBackground fertig ist

• Aufrufen:

  • SwingWorker#execute: Started neuen Thread nach Anlegen einer Instanz und führt dann automatisch doInBackground aus
  • SwingWorker#get: Return-Wert von doInBackground abfragen

Anmerkungen

• SwingWorker#done ist optional: kann überschrieben werden
  • Beispielweise, wenn nach Beendigung der langwierigen Berechnung GUI-Bestandteile mit dem Ergebnis aktualisiert werden sollen
• SwingWorker<T, V> ist eine generische Klasse:
  • T Typ für das Ergebnis der Berechnung, d.h. Rückgabetyp für doInBackground und get
  • V Typ für Zwischenergebnisse

Letzte Worte :-)

• Viele weitere Konzepte

  • Semaphoren, Monitore, ...
  • Leser-Schreiber-Probleme, Verklemmungen, ...

  => Verweis auf LV "Betriebssysteme" und "Verteilte Systeme"

• Achtung: Viele Klassen sind nicht Thread-safe!

  Es gibt aber meist ein "Gegenstück", welches Thread-safe ist.

  Beispiel Listen:

  • java.util.ArrayList ist nicht Thread-safe
  • java.util.Vector ist Thread-sicher

  => Siehe Javadoc in den JDK-Klassen!

• Thread-safe bedeutet Overhead (Synchronisierung)!

Wrap-Up

Multi-Threading auf höherem Level: Thread-Pools und Fork/Join-Framework

• Feingranulareres und flexibleres Locking mit Lock-Objekten und Conditions
• Wiederverwendung von Threads: Thread-Management mit Executor-Interface und Thread-Pools
• Fork/Join-Framework zum rekursiven Zerteilen von Aufgaben und zur parallelen Bearbeitung der Teilaufgaben
• SwingWorker für die parallele Bearbeitung von Aufgaben in Swing