Java Executor Framework: ExecutorService & Executors

Das vorherige Kapitel hat beschrieben, was ein Thread-Pool ist. Dieses Kapitel behandelt die Typhierarchie, mit der man mit einem Thread-Pool spricht — die Interfaces Executor, ExecutorService und ScheduledExecutorService. Zusammen nennt man sie das Executor-Framework, das in Java 5 eingeführt wurde, um „die Arbeit" von „den Threads, die sie ausführen" zu entkoppeln. Man schreibt Callable<Result> und Runnable; man übergibt sie; das Framework kümmert sich um Thread-Zuweisung, Warteschlange und Ergebnisübergabe.

Die dreistufige Hierarchie

Executor          // execute(Runnable)
   |
ExecutorService   // + submit/invokeAll/invokeAny/shutdown/awaitTermination
   |
ScheduledExecutorService  // + schedule/scheduleAtFixedRate/scheduleWithFixedDelay

Man programmiert gegen das allgemeinste Interface, das die benötigte Funktionalität bietet:

Executor — die einmethodige Basis. Verwende dieses Interface, wenn man nur „fire and forget" benötigt. Ein Methodenparameter vom Typ Executor ist der allgemeinste Vertrag „gib mir irgendetwas, das ein Runnable ausführen kann".
ExecutorService — das Arbeitspferd. Fast der gesamte Produktionscode verwendet diesen Typ. Fügt submit (mit einem Future-Ergebnis), Massenoperationen und Lebenszyklus-Management hinzu.
ScheduledExecutorService — wenn man verzögerte oder wiederholende Ausführung benötigt.

`Executor.execute` — fire and forget

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

Das ist das gesamte Interface. execute nimmt ein Runnable, führt es irgendwann in der Zukunft aus und gibt nichts zurück. Wenn die Arbeit eine Ausnahme wirft, erfährt man das nicht — die Ausnahme geht an den Uncaught-Exception-Handler des Worker-Threads.

execute ist der richtige Aufruf, wenn:

Die Arbeit keinen Rückgabewert hat.
Man nicht auf sie warten oder ihr Ergebnis abfragen muss.
Man sie nicht abbrechen muss.

Für alles Aufwändigere verwendet man submit.

`ExecutorService.submit`, die umfangreichere Variante

public interface ExecutorService extends Executor {
  <T> Future<T> submit(Callable<T> task);
  Future<?> submit(Runnable task);
  <T> Future<T> submit(Runnable task, T result);
  // ... lifecycle, bulk ops
}

submit gibt ein Future zurück, das Folgendes ermöglicht:

Auf den Abschluss warten (get() blockiert).
Das Ergebnis lesen (get() gibt den Wert des Callable zurück).
Die Aufgabe abbrechen (cancel(boolean mayInterrupt)).
Die Ausnahme der Aufgabe abfangen (get() wirft sie erneut).

Future und Callable werden im nächsten Kapitel ausführlich behandelt; für jetzt ist der Kontrast zu execute das Wesentliche. execute ist einwegig; submit öffnet einen Rückkanal.

ExecutorService pool = Executors.newFixedThreadPool(4);
Future<Integer> result = pool.submit(() -> {
  // Callable<Integer>; can throw, returns a value
  return expensive();
});

Integer value = result.get();                       // waits, throws ExecutionException if task failed

Massenoperationen: `invokeAll` und `invokeAny`

Wenn man eine Sammlung von Aufgaben hat:

List<Callable<Integer>> tasks = makeTasks();

List<Future<Integer>> futures = pool.invokeAll(tasks);          // run all, wait for all
Integer first = pool.invokeAny(tasks);                          // run all, return first success, cancel the rest

invokeAll(tasks, timeout, unit) führt die Aufgaben aus, bricht jedoch nach einer Frist ab; Aufgaben, die nicht fertig wurden, werden als Futures zurückgegeben, deren isDone() true ist, die aber abgebrochen wurden.

invokeAny ist das richtige Werkzeug für redundante Abfragen — drei DNS-Server anfragen, den ersten Antwortenden nehmen, die anderen abbrechen.

`ScheduledExecutorService` — Verzögerungen und Wiederholungen

Wenn man eine Verzögerung oder einen periodischen Zeitplan benötigt:

ScheduledExecutorService sched = Executors.newScheduledThreadPool(2);

sched.schedule(() -> log("once, after 5 seconds"), 5, TimeUnit.SECONDS);

sched.scheduleAtFixedRate(this::flush, 0, 1, TimeUnit.SECONDS);
// runs at t=0, t=1, t=2, ... — even if a run takes longer, the next one queues

sched.scheduleWithFixedDelay(this::poll, 0, 1, TimeUnit.SECONDS);
// runs at t=0, then 1 second AFTER the previous finished — back-to-back delay is what's fixed

Der Unterschied zwischen atFixedRate und withFixedDelay liegt darin, ob die Periode zwischen Starts oder zwischen Ende und nächstem Start gemessen wird. Für „Ich möchte jede Sekunde nach der Uhr flushen" verwendet man atFixedRate; für „Ich möchte einen 1-Sekunden-Abstand zwischen Läufen, egal wie lange sie dauern" verwendet man withFixedDelay.

Wenn eine geplante Aufgabe eine Ausnahme wirft, werden die zukünftigen Ausführungen stillschweigend abgebrochen. Der Scheduler protokolliert nichts. Geplante Aufgaben immer in einem übergeordneten try/catch einwickeln, damit sie weiter laufen:

sched.scheduleAtFixedRate(() -> {
  try { flush(); }
  catch (Throwable t) { log.error("flush failed", t); }
}, 0, 1, TimeUnit.SECONDS);

Das Vergessen dieses Musters ist der häufigste Scheduler-Bug in Java-Produktionsumgebungen.

Pool-Größe bestimmen

Die richtige Pool-Größe hängt davon ab, was die Aufgaben tun.

Für CPU-gebundene Arbeit lautet die Faustregel N + 1 Threads auf einer N-Kern-Maschine. Jeder Thread hält einen Kern beschäftigt; das +1 überbrückt den seltenen Moment, in dem ein Thread auf einen Speicher-Stall trifft.

Für I/O-gebundene Arbeit ist die richtige Zahl deutlich größer. Die grobe Formel:

threads = cores * (1 + (wait_time / compute_time))

Wenn die Aufgaben zu 90 % auf die Datenbank warten, beträgt der Multiplikator 10x — 80 Threads auf 8 Kernen. Die genaue Zahl hängt vom spezifischen I/O-Muster ab; profilieren und anpassen.

In der Praxis betreibt man zwei Pools: einen kleinen für CPU-Arbeit und einen großen für I/O. Beide nicht mischen — ein langsamer Datenbankaufruf innerhalb eines CPU-Pool-Threads blockiert einen Kern, der eigentlich rechnen sollte.

Java 21 Virtual Threads ändern diese Rechnung grundlegend: Das Blockieren bei I/O verschwendet keinen Plattform-Thread mehr, sodass man einen Virtual-Thread-per-Task-Executor verwenden und das Sizing vollständig vermeiden kann. Das wird am Ende des Teils behandelt.

`Executors`-Factories — Kurzreferenz

Die Factory-Methoden geben alle ExecutorService (oder ein Sub-Interface) zurück. Jede ist ein ThreadPoolExecutor mit bestimmten Konfigurationswerten:

Factory	Zugrunde liegende Konfiguration	Wann zu verwenden
`newFixedThreadPool(n)`	`core=max=n`, unbegrenzte `LinkedBlockingQueue`	Vorhersehbare Parallelität; unbegrenzte Warteschlange ist die Falle
`newCachedThreadPool`	`core=0`, `max=MAX_VALUE`, `SynchronousQueue`, 60s Keep-Alive	Kurze Burst-Aufgaben; unbegrenzte Thread-Anzahl ist die Falle
`newSingleThreadExecutor`	Wie `newFixedThreadPool(1)`, aber Pool ist nicht neu konfigurierbar	Einen einzigen geordneten Worker serialisieren
`newScheduledThreadPool(n)`	`n` Core-Threads, scheduled Warteschlange	Periodische Aufgaben
`newWorkStealingPool`	Java 8+: ein `ForkJoinPool` mit Parallelism = Kerne	CPU-gebundene Arbeit, rekursive Teilaufgaben
`newVirtualThreadPerTaskExecutor`	Java 21+: ein Virtual Thread pro Aufgabe	I/O-gebundene Arbeit, Web-Server

newFixedThreadPool und newCachedThreadPool für Produktions-Überlastpfade vermeiden — beide haben unbegrenzte Wachstumsachsen. Stattdessen new ThreadPoolExecutor(...) direkt mit einer begrenzten Warteschlange verwenden.

Die standardmäßige Shutdown-Sequenz

Ein Pool, der nie heruntergefahren wird, hält seine Nicht-Daemon-Worker-Threads am Leben und verhindert den JVM-Exit. Jeder erstellte Pool benötigt dasselbe Bereinigungsmuster:

ExecutorService pool = Executors.newFixedThreadPool(4);
try {
  // ... submit work, gather results ...
} finally {
  pool.shutdown();
  try {
    if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
      pool.shutdownNow();
      pool.awaitTermination(5, TimeUnit.SECONDS);
    }
  } catch (InterruptedException e) {
    pool.shutdownNow();
    Thread.currentThread().interrupt();
  }
}

Oder, seit Java 19, dasselbe über Try-with-Resources:

try (var pool = Executors.newFixedThreadPool(4)) {
  pool.submit(...);
  pool.submit(...);
}                                                    // close() runs shutdown + awaitTermination

Das Java 19 ExecutorService.close() führt das ordentliche Shutdown durch und wartet dann unbegrenzt; es mit einem Watchdog kombinieren, wenn man sich keinen unbegrenzten Wartevorgang leisten kann.

Ein vollständiges Beispiel: das Framework von Anfang bis Ende

Das folgende Programm verwendet alle drei Interfaces — Executor für fire-and-forget, ExecutorService für Ergebnisse und ScheduledExecutorService für periodische Ausführung — alles in einem.

java— editable, runs on the server

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorFrameworkDemo {
  public static void main(String[] args) throws Exception {
    // 1. Executor (the base interface) — fire and forget
    System.out.println("1. Executor.execute (no result):");
    Executor fire = Executors.newSingleThreadExecutor();
    fire.execute(() -> System.out.println("   ran on " + Thread.currentThread().getName()));
    ((ExecutorService) fire).shutdown();

// 2. ExecutorService.submit — returns a Future
    System.out.println("\n2. ExecutorService.submit + Future.get:");
    try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
      Future<Integer> f = pool.submit(() -> {
        Thread.sleep(50);
        return 42;
      });
      System.out.println("   future.get() = " + f.get());
    }

// 3. invokeAll — wait for all
    System.out.println("\n3. invokeAll (run all, wait for all):");
    try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
      List<Callable<Integer>> tasks = List.of(
          () -> { Thread.sleep(50); return 1; },
          () -> { Thread.sleep(80); return 2; },
          () -> { Thread.sleep(20); return 3; }
      );
      long t0 = System.nanoTime();
      List<Future<Integer>> results = pool.invokeAll(tasks);
      long ms = (System.nanoTime() - t0) / 1_000_000;
      int sum = 0;
      for (Future<Integer> r : results) sum += r.get();
      System.out.println("   sum=" + sum + "  in ~" + ms + " ms (limited by slowest task)");
    }

// 4. invokeAny — first success wins; rest cancelled
    System.out.println("\n4. invokeAny (return first success, cancel the rest):");
    try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
      List<Callable<String>> tasks = List.of(
          () -> { Thread.sleep(80); return "slow-A"; },
          () -> { Thread.sleep(50); return "fast-B"; },
          () -> { Thread.sleep(100); return "slowest-C"; }
      );
      String winner = pool.invokeAny(tasks);
      System.out.println("   winner=" + winner);
    }

// 5. ScheduledExecutorService
    System.out.println("\n5. ScheduledExecutorService:");
    ScheduledExecutorService sched = Executors.newScheduledThreadPool(2);
    AtomicInteger ticks = new AtomicInteger();
    var task = sched.scheduleAtFixedRate(() -> {
      try {
        System.out.println("   tick " + ticks.incrementAndGet()
            + " on " + Thread.currentThread().getName());
      } catch (Throwable t) {                          // never let it propagate
        System.out.println("   scheduled task failed: " + t);
      }
    }, 0, 60, TimeUnit.MILLISECONDS);
    Thread.sleep(250);                                  // let it fire ~4 times
    task.cancel(false);
    sched.shutdown();

System.out.println("\ndone.");
  }
}

Was man aus dem Lauf mitnehmen sollte:

Abschnitt 2 verwendete try (ExecutorService pool = ...) — das Java-19-Muster zum Schließen beim Verlassen des Gültigkeitsbereichs. Das close() des Pools führt shutdown() aus und wartet dann. Das ist die sauberste Shutdown-Form; für älteren Code oder strengere Deadlines fällt man auf die Sequenz shutdown + awaitTermination + shutdownNow zurück.
Abschnitt 3 führte drei Aufgaben von 50/80/20 ms auf 4 Workern aus. invokeAll kehrte erst zurück, nachdem die langsamste Aufgabe abgeschlossen war — etwa 80 ms. Das ist der „warte auf alle"-Vertrag. Die sum über die Futures war die Summe der zurückgegebenen Werte in der Einreichungsreihenfolge.
Abschnitt 4 führte dieselbe Struktur mit invokeAny aus. Die schnellste Aufgabe (50 ms) kehrte zuerst zurück; die anderen wurden abgebrochen. invokeAny ist genau die richtige Form für „erste erfolgreiche Antwort"-Muster — DNS-Lookups gegen mehrere Server, Mirror-Downloads, Latenz-Rennen.
Abschnitt 5 verwendete scheduleAtFixedRate mit einer Periode von 60 ms. Jeder Tick feuerte auf einem Scheduled-Pool-Thread. Das try/catch-Wrapper innerhalb des Körpers ist die Produktionsform — wenn eine geplante Aufgabe eine Ausnahme wirft, bricht der Scheduler zukünftige Läufe stillschweigend ab. Jeden Körper in einem übergeordneten Catch zu verpacken verhindert das.
Die geplante Aufgabe wurde explizit mit cancel(false) abgebrochen, bevor das Programm endete. Das Abbrechen und Herunterfahren des Schedulers ermöglicht dem JVM zu beenden; ohne es hält der Scheduler Nicht-Daemon-Threads und das Programm hängt. Dasselbe gilt für jeden erstellten Executor.

Was kommt als Nächstes

Das nächste Kapitel, Java Callable and Future, taucht in die Ergebnisverarbeitung von submit ein — Callable<V>, Future<V>, Abbruch und die Standardmuster zum Abrufen eines Werts aus einer asynchronen Aufgabe.

Übungen

Übung

Du planst eine Aufgabe mit `scheduleAtFixedRate` und sie wirft beim dritten Lauf eine `RuntimeException`. Was passiert mit den nachfolgenden Läufen?

Der Scheduler bricht alle zukünftigen Läufe dieser Aufgabe stillschweigend ab — man muss den Körper in ein `try`/`catch` einwickeln, damit er weiterläuftDer nächste Lauf wird normal fortgesetzt; nur die aktuelle Ausführung wird gemeldetDer Scheduler wiederholt den fehlgeschlagenen Lauf zweimal, bevor er abbrichtDer gesamte `ScheduledExecutorService` wird heruntergefahren und keine anderen geplanten Aufgaben laufen mehr

Die dreistufige Hierarchie

Executor.execute — fire and forget

ExecutorService.submit, die umfangreichere Variante

Massenoperationen: invokeAll und invokeAny

ScheduledExecutorService — Verzögerungen und Wiederholungen