SemaTrain Logo Ein Fachportal von SemaTrain

Multithreading & parallele Programmierung

Sie lernen, konkurrenten Code sicher zu schreiben: Executors statt „rohe Threads“, Synchronisation ohne Chaos, und Sie erkennen typische Fehler wie Race Conditions, Deadlocks und Visibility-Probleme – im SemaTrain-Trainingskontext.

Hinweis: Parallele Programmierung ist weniger „mehr Threads = schneller“, sondern vor allem korrekte Zustände + kontrollierte Ressourcen.

Java Fortgeschrittenen Schulung – Kursbezug

Dieses Kapitel ist Teil des Lernpfads zur Java Fortgeschrittenen Schulung. Fokus: Parallelität sicher nutzen – z.B. wenn SemaTrain Kursdaten/Reports parallel aus mehreren Quellen zusammenstellt.

Ziel dieses Kapitels: Sie führen Aufgaben parallel aus (ExecutorService), koordinieren Ergebnisse (Future/CompletableFuture) und schützen kritische Bereiche korrekt.

Worum geht’s?

Lehr-/Lernziele

Nach diesem Kapitel können Sie …

ExecutorService: kontrollierte Parallelität

SemaTrain-Szenario: Ein Kursreport sammelt parallel Teilinfos (z.B. Trainerprofil, Buchungen, Auslastung).

ExecutorService: Kursreport-Teile parallel laden + sauber beenden (Java)
import java.util.concurrent.*;

public class KursReportPoolDemo {

  public static void main(String[] args) throws Exception {

    ExecutorService pool = Executors.newFixedThreadPool(4);

    Future<String> trainerInfo = pool.submit(() -> {
      Thread.sleep(120);
      return "Trainer: verfügbar";
    });

    Future<String> auslastung = pool.submit(() -> {
      Thread.sleep(180);
      return "Auslastung: 80%"; // Demo
    });

    System.out.println("Report-Teil 1: " + trainerInfo.get());
    System.out.println("Report-Teil 2: " + auslastung.get());

    pool.shutdown();
    if(!pool.awaitTermination(2, TimeUnit.SECONDS)) {
      pool.shutdownNow();
    }
  }
}

Race Condition: gemeinsamer Zähler

Szenario: Mehrere Checks laufen parallel und erhöhen einen „fertig“-Zähler. Ohne Schutz gehen Updates verloren.

Race: fertig++ ist nicht atomar (Java)
import java.util.concurrent.*;

public class KursCheckRace {

  static int fertig = 0;

  public static void main(String[] args) throws Exception {
    ExecutorService pool = Executors.newFixedThreadPool(8);

    for(int i=0;i<1000;i++){
      pool.submit(() -> fertig++); // NICHT thread-safe
    }

    pool.shutdown();
    pool.awaitTermination(2, TimeUnit.SECONDS);

    System.out.println("fertig=" + fertig); // oft < 1000
  }
}
Fix: AtomicInteger (atomare Operation) (Java)
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class KursCheckSafe {

  static AtomicInteger fertig = new AtomicInteger(0);

  public static void main(String[] args) throws Exception {
    ExecutorService pool = Executors.newFixedThreadPool(8);

    for(int i=0;i<1000;i++){
      pool.submit(() -> fertig.incrementAndGet());
    }

    pool.shutdown();
    pool.awaitTermination(2, TimeUnit.SECONDS);

    System.out.println("fertig=" + fertig.get()); // 1000
  }
}

Kritischer Bereich: Lock / synchronized

Szenario: Report zählt konsistent OK/Fehler und hält die beiden Zähler zusammen.

Lock: mehrere Zähler konsistent schützen (Java)
import java.util.concurrent.locks.ReentrantLock;

public class KursReportStatus {
  private int ok = 0;
  private int fehler = 0;
  private final ReentrantLock lock = new ReentrantLock();

  public void markOk() {
    lock.lock();
    try { ok++; }
    finally { lock.unlock(); }
  }

  public void markFehler() {
    lock.lock();
    try { fehler++; }
    finally { lock.unlock(); }
  }

  public String summary() {
    return "OK=" + ok + " | Fehler=" + fehler;
  }
}

CompletableFuture: parallel laden, dann aggregieren

Szenario: SemaTrain lädt parallel Kennzahlen (Buchungen, Umsatz, Auslastung) und baut daraus einen Report mit Timeout & Fallback.

CompletableFuture.allOf: parallel + aggregieren + Timeout/Fallback (Java)
import java.util.concurrent.*;

public class DashboardReportDemo {

  public static void main(String[] args) {

    CompletableFuture<Integer> buchungen =
      CompletableFuture.supplyAsync(() -> slow(200, 12));

    CompletableFuture<Double> umsatz =
      CompletableFuture.supplyAsync(() -> slow(350, 2499.0));

    CompletableFuture<String> auslastung =
      CompletableFuture.supplyAsync(() -> slow(250, "80%")); // demo

    CompletableFuture<String> report =
      CompletableFuture.allOf(buchungen, umsatz, auslastung)
        .orTimeout(1, TimeUnit.SECONDS)
        .thenApply(v ->
          "Kursreport: Buchungen=" + buchungen.join()
          + " | Umsatz=" + umsatz.join()
          + " | Auslastung=" + auslastung.join()
        )
        .exceptionally(ex ->
          "Kursreport nicht verfügbar: " + ex.getClass().getSimpleName()
        );

    System.out.println(report.join());
  }

  static <T> T slow(int ms, T value) {
    try { Thread.sleep(ms); } catch (InterruptedException ignored) {}
    return value;
  }
}

Deadlocks vermeiden (Merksatz)

Praxisaufgabe (Mini)

Beitrag zu den Lehr-/Lernzielen: LZ1 (Pool) · LZ2 (Thread-Safety) · LZ3 (Koordination)

  1. Erstellen Sie einen ExecutorService mit 4 Threads.
  2. Starten Sie 20 Tasks: jeder Task simuliert einen „Kurs-Datencheck“ (OK/Fehler + Meldung).
  3. Sammeln Sie Ergebnisse ein (Timeout pro Task).
  4. Zählen Sie OK/Fehler und geben Sie Fehlerdetails aus.
  5. Beenden Sie den Pool sauber (shutdown/awaitTermination).
Lösung anzeigen
Lösung: SemaTrain Kurs-Batch-Check (Pool + Future + Timeout + Summary) (Java)
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class KursBatchCheck {

  // Ergebnis-Typ (Java 16+ Record; sonst: normale Klasse)
  public record CheckErgebnis(int kursNr, boolean ok, String meldung) {}

  // Simulierter Check (z.B. Datenvollständigkeit/Plätze/Trainer)
  static CheckErgebnis pruefe(int kursNr) {
    try { Thread.sleep(50); } catch (InterruptedException ignored) {}

    // Demo-Regel: jeder 5. Kurs ist "auffällig"
    boolean ok = (kursNr % 5 != 0);
    String msg = ok
      ? "OK: Kursdaten plausibel"
      : "FEHLER: Trainer fehlt oder Plätze unplausibel";

    return new CheckErgebnis(kursNr, ok, msg);
  }

  public static void main(String[] args) throws Exception {

    ExecutorService pool = Executors.newFixedThreadPool(4);
    List<Future<CheckErgebnis>> futures = new ArrayList<>();
    AtomicInteger fertig = new AtomicInteger(0);

    for (int i = 1; i <= 20; i++) {
      final int nr = i;
      futures.add(pool.submit(() -> {
        try {
          return pruefe(nr);
        } finally {
          fertig.incrementAndGet();
        }
      }));
    }

    int okCount = 0;
    int fehlerCount = 0;
    List<String> fehlerDetails = new ArrayList<>();

    for (Future<CheckErgebnis> f : futures) {
      try {
        CheckErgebnis erg = f.get(300, TimeUnit.MILLISECONDS);
        if (erg.ok()) okCount++;
        else {
          fehlerCount++;
          fehlerDetails.add("Kurs " + erg.kursNr() + ": " + erg.meldung());
        }
      } catch (TimeoutException ex) {
        fehlerCount++;
        fehlerDetails.add("Timeout: Ein Check dauerte zu lange");
      }
    }

    pool.shutdown();
    if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
      pool.shutdownNow();
    }

    System.out.println("Fertig: " + fertig.get() + "/20");
    System.out.println("OK: " + okCount + " | Fehler: " + fehlerCount);

    if (!fehlerDetails.isEmpty()) {
      System.out.println("\nFehlerdetails:");
      for (String m : fehlerDetails) System.out.println("- " + m);
    }
  }
}
Optional: typische Stolperfallen
  • Pool nie beendet: shutdown() + awaitTermination() nutzen.
  • Shared State ohne Schutz: counter++ ist nicht atomar → Atomic/Lock/synchronized.
  • Blockieren im Pool: lange blockierende Calls „verstopfen“ Threads → Timeouts/Design prüfen.
  • Zu viele Threads: mehr Threads ≠ schneller (Overhead, Context Switch).

Kurz-Takeaways

Quiz: Multithreading (Lehr-/Lernziele Check)

1. (LZ1) Warum ist ExecutorService oft besser als Threads direkt zu starten?

2. (LZ2) Warum ist counter++ in mehreren Threads problematisch?

3. (LZ3) Was ist ein typischer Vorteil von CompletableFuture?

4. (LZ2) Welche Maßnahme hilft gegen Race Conditions bei einem Zähler?

Praxisaufgabe

Mini-Projekt: „Batch-Check“ (Parallelität sicher nutzen im SemaTrain-Kurskontext)

Beitrag zu den Lehr-/Lernzielen: LZ1 (ExecutorService + Shutdown) · LZ2 (Thread-Safety) · LZ3 (Koordination/Aggregation + Timeouts)

Szenario: SemaTrain führt einen schnellen „Qualitäts-/Status-Check“ über viele Kurse aus (z.B. Daten vollständig? Plätze plausibel? Trainer zugeordnet?) und erzeugt am Ende eine kurze Zusammenfassung. Wichtig: kontrollierte Parallelität (Pool), thread-sichere Zählung und sauberes Beenden.

Lösung anzeigen
Teil 1: Ergebnis-Typ (verständlich + erweiterbar) (Java)
public record CheckErgebnis(
  int kursNummer,
  boolean ok,
  String meldung
) {}
Teil 2: Check-Funktion (simuliert, aber praxisnah formulierbar) (Java)
public class KursCheck {

  // Simuliert einen Check (z.B. Datenvollständigkeit)
  public static CheckErgebnis pruefe(int kursNummer) {
    try { Thread.sleep(50); } catch (InterruptedException ignored) {}

    // Demo-Regel: jeder 5. Kurs ist "auffällig"
    boolean ok = (kursNummer % 5 != 0);
    String meldung = ok
      ? "OK: Kursdaten sind plausibel"
      : "FEHLER: Kursdaten unvollständig oder Plätze unplausibel";

    return new CheckErgebnis(kursNummer, ok, meldung);
  }
}
Teil 3: Batch-Check (ExecutorService + Future + AtomicInteger + Timeouts + Shutdown) (Java)
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class BatchCheck {

  public static void main(String[] args) throws Exception {

    ExecutorService pool = Executors.newFixedThreadPool(4);

    List<Future<CheckErgebnis>> futures = new ArrayList<>();
    AtomicInteger fertig = new AtomicInteger(0);

    // 20 Checks starten
    for (int i = 1; i <= 20; i++) {
      final int kursNummer = i;
      futures.add(pool.submit(() -> {
        try {
          CheckErgebnis erg = KursCheck.pruefe(kursNummer);
          return erg;
        } finally {
          fertig.incrementAndGet();
        }
      }));
    }

    int anzahlOk = 0;
    int anzahlFehler = 0;
    List<String> fehlerMeldungen = new ArrayList<>();

    // Ergebnisse einsammeln (mit Timeout je Task)
    for (Future<CheckErgebnis> f : futures) {
      try {
        CheckErgebnis erg = f.get(300, TimeUnit.MILLISECONDS);
        if (erg.ok()) anzahlOk++;
        else {
          anzahlFehler++;
          fehlerMeldungen.add("Kurs " + erg.kursNummer() + ": " + erg.meldung());
        }
      } catch (TimeoutException ex) {
        anzahlFehler++;
        fehlerMeldungen.add("Zeitüberschreitung: Ein Check hat zu lange gedauert");
      }
    }

    // Pool sauber beenden
    pool.shutdown();
    if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
      pool.shutdownNow();
    }

    System.out.println("Fertig: " + fertig.get() + "/20");
    System.out.println("OK: " + anzahlOk + " | Fehler: " + anzahlFehler);

    if (!fehlerMeldungen.isEmpty()) {
      System.out.println("\nDetails (Fehler):");
      for (String m : fehlerMeldungen) System.out.println("- " + m);
    }
  }
}
Teil 4 (optional): CompletableFuture.allOf + Timeout + Fallback (Java)
import java.util.*;
import java.util.concurrent.*;

public class BatchCheckMitCompletableFuture {

  public static void main(String[] args) {

    List<CompletableFuture<CheckErgebnis>> tasks = new ArrayList<>();

    for (int i = 1; i <= 20; i++) {
      final int kursNummer = i;
      tasks.add(CompletableFuture
        .supplyAsync(() -> KursCheck.pruefe(kursNummer))
        .orTimeout(300, TimeUnit.MILLISECONDS)
        .exceptionally(ex -> new CheckErgebnis(
          kursNummer,
          false,
          "Fehler im Check: " + ex.getClass().getSimpleName()
        ))
      );
    }

    CompletableFuture<Void> alle = CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0]));

    List<CheckErgebnis> ergebnisse = alle.thenApply(v ->
      tasks.stream().map(CompletableFuture::join).toList()
    ).join();

    long ok = ergebnisse.stream().filter(CheckErgebnis::ok).count();
    long fehler = ergebnisse.size() - ok;

    System.out.println("OK: " + ok + " | Fehler: " + fehler);
  }
}
Teil 5: Hinweise (Best Practices) (Text)
// Typische Stolperfallen (kurz):
// - Pool nicht beenden: shutdown/awaitTermination vergessen
// - Shared State ohne Schutz: counter++ ist nicht atomar (Atomic/Lock/synchronized)
// - Blockierende Calls im Pool: zu kleine Pools „verstopfen“ schnell
// - Deadlocks: Locks immer in gleicher Reihenfolge oder shared state reduzieren
// - Ergebnisse lieber sammeln (immutable) statt globale, veränderliche Strukturen teilen