Zum Inhalt
wannacode

Thread-Pool und Aufgaben

Kurz gesagt: Ein Thread-Pool sind ein paar Worker vor einer Aufgabenwarteschlange statt eines neuen Threads pro Aufgabe; in Java ist das ExecutorService, und jedes Ergebnis kommt als Future zurück.

Problem

Ein Thread für jedes Bild

Der Shop startet einen Sale. Verkäufer laden Bilder ihrer Produkte hoch, und jedes Bild braucht ein Vorschaubild für die Produktliste. Ein Vorschaubild zu erzeugen ist Arbeit, die rechnet: Sie geht durch jeden Pixel (siehe Arbeit, die wartet, und Arbeit, die rechnet).

Die erste Version startet für jedes hochgeladene Bild einen Thread:

List<Thread> threads = new ArrayList<>();
for (String image : uploads) {
    Thread thread = new Thread(() -> saveThumbnail(image));
    thread.start();
    threads.add(thread);
}

Mit 20 Bildern funktioniert alles auf deinem Rechner: „Fertig: 20 Vorschaubilder“. In der ersten Stunde des Sales kommen 10.000 auf einmal.

  • Jeder Thread braucht einen eigenen Stack (meist bis zu 1 MB) und einen Platz im Betriebssystem. Bei ein paar Tausend Threads bricht die JVM mit OutOfMemoryError: unable to create native thread ab.
  • Selbst wenn Speicher da ist, wechseln sich 10.000 Threads auf 8 Kernen nur ab. Der Prozessor verschwendet Zeit mit dem Umschalten, und jedes Bild wird langsamer.

Der Server wird nicht langsamer, er fällt um. Und ein Server, der down ist, bedient auch die Kunden nicht.

Die Idee

Ein paar Worker und eine Warteschlange

Ein Thread-Pool sind ein paar Worker, einmal erzeugt, und davor eine Warteschlange von Aufgaben. Eine Aufgabe ist „verkleinere dieses Bild“. Ein freier Worker nimmt die nächste Aufgabe aus der Warteschlange, erledigt sie und kommt für eine neue zurück.

  • Die Zahl der Worker ist konstant. 10.000 Bilder ergeben nicht 10.000 Threads, sondern 10.000 Aufgaben in der Warteschlange. Eine Aufgabe ist ein kleines Objekt; ein Thread ist teuer.
  • Die Worker werden wiederverwendet. Für jedes Bild wird kein Thread erzeugt und zerstört.
  • Die Warteschlange fängt die Spitze ab. Kommen Uploads schneller, als die Worker sie bearbeiten, warten die Aufgaben, statt den Server zu Fall zu bringen.

Das ist Producer und Consumer: Der Code, der Uploads annimmt, legt Aufgaben in die Warteschlange, und die Worker sind die Consumer.

Den Pool hast du schon in der Lektion über Threads kennengelernt: ExecutorService, submit und Future.get(). Hier sehen wir, was darin passiert, wie man die Ergebnisse ordnet, was mit Fehlern geschieht und wie viele Worker du brauchst.

Sieh es dir an

Drei Worker

Ein Vorschaubild zu erzeugen ist hier eine Pause, 2 ms für jedes KB, und die Größe kommt aus dem Namen des Bildes. So ist die Ausgabe bei jedem Lauf gleich:

// Vorschaubild erzeugen: Hier ist es eine Pause, und die Größe kommt aus dem Namen.
static int thumbnail(String image) throws InterruptedException {
    if (image.startsWith("defekt")) {
        throw new IllegalArgumentException("defekte Datei");
    }
    int kb = 20 + 3 * image.length();
    Thread.sleep(2 * kb);
    return kb;
}

Acht Bilder, ein Pool aus drei Workern:

try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Gesamt: " + total + " KB");
}
Schritt für Schritt, in Worten
  1. 8 Bilder warten. Der Pool besteht aus 3 Threads: Executors.newFixedThreadPool(3).
  2. submit macht aus jedem Bild eine Aufgabe in der Warteschlange. Die drei Threads nehmen die ersten drei.
  3. Thread 3 wird als Erstes fertig: Hut ist das dritte Bild, und sein Ergebnis kommt an dritter Stelle. Der Worker hält nicht an: Er nimmt die nächste Aufgabe.
  4. Die Warteschlange ist leer. Thread 1 ist fertig und hat nichts mehr zu nehmen, also wartet er. Die anderen beenden ihre letzten Bilder.
  5. get() liest die Größen in der Reihenfolge der Bilder, nicht in der Reihenfolge des Fertigwerdens. Gesamt 379 KB; das Ende von try schließt den Pool.
List<String> images = List.of("Kleid.jpg", "Mantel.jpg", "Hut.jpg", "Jacke.jpg", "Rucksack.jpg", "Gürtel.jpg", "Mütze.jpg", "Uhr.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Gesamt: " + total + " KB");
}
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
Hut.jpg: 41 KB
Jacke.jpg: 47 KB
Rucksack.jpg: 56 KB
Gürtel.jpg: 50 KB
Mütze.jpg: 47 KB
Uhr.jpg: 41 KB
Gesamt: 379 KB
  • Executors.newFixedThreadPool(3) erzeugt drei Threads und davor eine endlose Warteschlange.
  • submit legt die Aufgabe in die Warteschlange und gibt sofort ein Future zurück. Die Schleife wartet auf nichts.
  • get() wartet genau auf die eigene Aufgabe. Die Schleife wartet in der Reihenfolge der Bilder darauf, also steht die Ausgabe auch in dieser Reihenfolge.
  • try (var pool = …) schließt den Pool am Ende: close() nimmt keine neuen Aufgaben an, wartet auf die gestarteten und stoppt die Threads. Das gibt es ab Java 19. In älterem Code siehst du shutdown() und awaitTermination in einem finally.
Genauer, für Neugierige

Die Warteschlange von newFixedThreadPool ist endlos. Kommen Uploads immer schneller, als der Pool sie bearbeiten kann, wächst die Warteschlange, bis der Speicher ausgeht. Ein Pool mit begrenzter Warteschlange ist new ThreadPoolExecutor(3, 3, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy()): Ist die Warteschlange voll, läuft die Aufgabe in dem Thread, der sie schickt, sodass nur das Annehmen der Uploads langsamer wird.

Wie es in den anderen Sprachen ist: In C# gibt es einen gemeinsamen Pool (ThreadPool, Task.Run, Parallel), in Python ist es ThreadPoolExecutor, und Node hat keinen fertigen Pool von Workern. In einem Projekt nimmst du dort das Paket Piscina.

Probier es aus

Finde den Fehler

Ein Kollege hat die Vorschaubilder mit einem Pool umgeschrieben, und der Sale wurde nicht schneller. Sechs Bilder brauchen so lange wie ohne Pool:

long start = System.nanoTime();
List<Integer> sizes = new ArrayList<>();
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        Future<Integer> size = pool.submit(() -> thumbnail(image));
        sizes.add(size.get());
    }
}
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("Dauerte über 0,5 s: " + (ms > 500));
Ausgabe
Dauerte über 0,5 s: true

Wähle die Zeile mit dem Fehler.

Die Idee

Ergebnisse in Eingabereihenfolge

Aufgaben werden in einer Reihenfolge fertig, die du nicht kennst: Sie hängt von den Bildern ab, von der Last, davon, welcher Worker frei war. Gib also nie „in der Reihenfolge des Fertigwerdens“ aus, wenn du jedes Mal dieselbe Ausgabe willst.

Zwei Wege liefern bei jedem Lauf dasselbe Ergebnis:

  • In Eingabereihenfolge. Du hältst die Promises in einer Liste in der Reihenfolge der Aufgaben und wartest in dieser Reihenfolge darauf. So funktioniert der Code oben.
  • Eine Summe oder eine Anzahl. Brauchst du nur das Ergebnis insgesamt, ist die Reihenfolge egal: 47 + 50 ist dasselbe wie 50 + 47.

Manchmal willst du die Ergebnisse in der Reihenfolge des Fertigwerdens: für einen Balken „3 von 8 fertig“ oder um die ersten fertigen Bilder zu zeigen. ExecutorCompletionService liefert sie so. Dann sammelst du nur Dinge, die nicht von der Reihenfolge abhängen:

try (var pool = Executors.newFixedThreadPool(3)) {
    var done = new ExecutorCompletionService<Integer>(pool);
    for (String image : images) {
        done.submit(() -> thumbnail(image));
    }
    int count = 0;
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        total += done.take().get();
        count++;
    }
    System.out.println("Fertig: " + count + " von " + images.size() + ", gesamt: " + total + " KB");
}
Ausgabe
Fertig: 8 von 8, gesamt: 379 KB
ExecutorCompletionService führt eine eigene Warteschlange der fertigen Aufgaben. take() wartet auf die nächste fertige, welche auch immer es ist.

Die Idee

Ein Fehler in einer Aufgabe

Eines der Bilder ist defekt, und thumbnail wirft eine Exception. Sie kommt nicht in dem Thread heraus, der die Aufgabe gestartet hat: Der ist längst weitergezogen. Der Pool fängt sie ab und bewahrt sie im Promise des Ergebnisses auf. Sie kommt heraus, wenn du auf dieses Ergebnis wartest:

List<String> images = List.of("Kleid.jpg", "Mantel.jpg", "defekt.jpg", "Hut.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    for (int i = 0; i < images.size(); i++) {
        try {
            int kb = sizes.get(i).get();
            System.out.println(images.get(i) + ": " + kb + " KB");
        } catch (ExecutionException e) {
            String why = e.getCause().getMessage();
            System.out.println(images.get(i) + ": Fehler: " + why);
        }
    }
}
Ausgabe
Kleid.jpg: 47 KB
Mantel.jpg: 50 KB
defekt.jpg: Fehler: defekte Datei
Hut.jpg: 41 KB
get() wirft eine ExecutionException, und die eigentliche Exception steckt in getCause(). Den anderen Bildern passiert nichts: Der Pool macht mit den nächsten Aufgaben weiter.

Probier es aus

Wohin ist der Fehler gegangen

Das zweite Bild ist defekt. Was gibt das Programm aus?

List<String> images = List.of("Kleid.jpg", "defekt.jpg", "Hut.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        pool.submit(() -> thumbnail(image));
    }
}
System.out.println("Fertig: " + images.size() + " Bilder");
Antwort

Die Idee

Wie viele Worker

Das hängt davon ab, was sie tun:

  • Arbeit, die rechnet (Vorschaubilder, PDFs, Komprimierung): so viele wie Kerne. Ein neunter Worker auf 8 Kernen hat keinen Platz zum Laufen; er wechselt sich nur mit den anderen ab.
  • Arbeit, die wartet (E-Mail, Zahlung, Datenbank): mehr. Während ein Worker auf das Netzwerk wartet, ist der Kern für einen anderen frei. Wartet eine Aufgabe W ms und rechnet C ms, passen etwa 1 + W / C Worker auf einen Kern.
static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + waitMs / computeMs);
}


System.out.println("Vorschaubilder (nur Rechnen): " + poolSize(8, 0, 50));
System.out.println("E-Mails (90 ms Warten, 10 ms Rechnen): " + poolSize(8, 90, 10));
Ausgabe
Vorschaubilder (nur Rechnen): 8
E-Mails (90 ms Warten, 10 ms Rechnen): 80

Die Zahl der Kerne kommt aus Runtime.getRuntime().availableProcessors(). Die Formel ist ein Anfang, kein Gesetz: Miss nach. Gehen die E-Mails über einen Server mit einem Limit von 20 Verbindungen, warten 80 Worker nur in einer Schlange davor.

Probier es aus

Ergänze

Wie viele Worker für eine Aufgabe, die waitMs wartet und computeMs rechnet?

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + );
}
Genauer, für Neugierige

Virtuelle Threads (Java 21) ändern die Rechnung für Arbeit, die wartet. Ein virtueller Thread kostet ein paar Hundert Byte, und wartet er (sleep, Netzwerk, Datenbank), nimmt die JVM ihn vom echten Thread und führt einen anderen aus. Executors.newVirtualThreadPerTaskExecutor() liefert für jede Aufgabe einen neuen virtuellen Thread, und 10.000 E-Mails sind wirklich 10.000 Threads, ohne Pool. Für Arbeit, die rechnet, helfen sie nicht: Es sind immer noch 8 Kerne. Und die Ressourcen dahinter (Datenbankverbindungen) sind weiter begrenzt, zum Beispiel mit einem Semaphore.

Übung

Die Rechnungen des Sales

Löse die Aufgabe in deinem Editor.

Nach dem Sale braucht jede Bestellung eine Rechnung. invoice(amount) erzeugt das PDF (hier eine Pause von 50 ms) und gibt den Betrag mit 20 % MwSt. zurück. Eine Bestellung mit einem Betrag von 0 oder weniger ist ungültig, und die Funktion wirft eine Exception mit dem Text „ungültiger Betrag“.

Die Bestellungen sind 1001 bis 1008 mit den Beträgen 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € und 15 €. Erzeuge die Rechnungen in einem Pool aus 4 Workern. Gib sie in der Reihenfolge der Bestellungen aus, die ungültige mit dem Text des Fehlers. Am Ende: die Summe der guten Rechnungen und wie viele gescheitert sind.

Ausgabe
Bestellung 1001: 60 €
Bestellung 1002: 150 €
Bestellung 1003: 96 €
Bestellung 1004: Fehler: ungültiger Betrag
Bestellung 1005: 288 €
Bestellung 1006: 42 €
Bestellung 1007: 108 €
Bestellung 1008: 18 €
Gesamt: 762 €, mit Fehler: 1

Weitere Aufgaben

  • Eine begrenzte Warteschlange. Lass den Pool höchstens 10 Aufgaben in der Warteschlange warten. Was passiert mit der elften? (ArrayBlockingQueue und CallerRunsPolicy)
  • Fortschritt. Gib nach jeder fertigen Rechnung „3 von 8 fertig“ aus, und am Ende noch einmal die geordnete Liste.
  • Ein Timeout. Eine Rechnung, die über 1 Sekunde dauert, gilt als gescheitert, ohne die anderen zu stoppen.

Test

Teste dich

1. Warum ein Pool aus ein paar Threads und nicht ein Thread für jedes hochgeladene Bild?
2. Ein Pool aus 3 Workern verkleinert 8 Bilder. Das dritte wird als Erstes fertig. Wo steht seine Größe, wenn du sie in Listenreihenfolge mit get, result, Promise.all oder sizes[i] liest?
3. Eine Aufgabe im Pool wirft eine Exception. Wann siehst du sie?
4. Der Server hat 8 Kerne. Jede E-Mail wartet 90 ms auf das Netzwerk und rechnet 10 ms. Etwa wie viele Worker halten die Kerne beschäftigt?
5. Wie viele Worker für Vorschaubilder, wenn die Arbeit nur rechnet und der Server 8 Kerne hat?

Erstellt von Ivo Gergov im Rahmen der Kampagne von Evo Tech für zugängliche Programmierausbildung ·