Producer und Consumer

Code in:JavaC#PythonJavaScript

Kurz gesagt: Producer-Consumer ist ein Muster, bei dem einige Threads Aufgaben in eine begrenzte Warteschlange legen und andere sie herausnehmen; eine volle Warteschlange lässt den Producer warten (Backpressure).

Problem

Black Friday

Um 10:00 beginnt der Sale, und die Website nimmt 50 Bestellungen pro Sekunde an. Das Lager packt 10. Es gibt drei schlechte Wege, damit umzugehen.

Die Website packt selbst. Der Kunde klickt auf „Bestellen“ und wartet, bis jemand den Karton gepackt hat. Die Website ist so schnell wie das Lager.

Die Website legt die Bestellungen in eine einfache Liste, und das Lager nimmt sie von dort. Zwei Packer können dieselbe Bestellung nehmen, und eine andere kann verloren gehen: Das ist eine Race Condition. Wenn du ein Lock hinzufügst, muss ein Packer mit leerer Liste immer wieder nachsehen, ob etwas gekommen ist.

Die Liste hat kein Limit. Die Website ist zufrieden: Sie wartet nie. Aber jede Sekunde wächst die Liste um 40 Bestellungen. Nach einer Stunde enthält sie 144.000, die letzte hat eine Stunde gewartet, und der Speicher geht aus.

Die Idee

Eine Warteschlange mit begrenztem Platz

Zwischen die beiden Seiten kommt eine Warteschlange. Der Producer legt Aufgaben hinten hinein, die Consumer nehmen sie vorn heraus. Das ist das Muster Producer und Consumer. Die beiden Seiten kennen einander nicht: Die Website weiß nicht, wie viele Packer es gibt, und die Packer wissen nicht, woher die Bestellungen kommen.

Die Warteschlange ist blockierend und hat begrenzten Platz:

• ist sie leer, wartet der Consumer, ohne den Prozessor zu beschäftigen, bis etwas kommt;
• ist sie voll, wartet der Producer, bis ein Consumer einen Platz frei macht;
• das Lock steckt in der Warteschlange: Jede Bestellung geht an genau einen Consumer.

In Java ist das eine BlockingQueue. ArrayBlockingQueue<>(3) hat 3 Plätze. put wartet, wenn sie voll ist, und take wartet, wenn sie leer ist. Sie liegt in java.util.concurrent, wie ConcurrentLinkedQueue, in der wir die gepackten Bestellungen sammeln.

Sieh es dir an

Die Website und die Packer

static final Order DONE = new Order(0, 0, false);

static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // wartet, wenn voll
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // eine Pille pro Packer
    }
}

static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // wartet, wenn leer
        if (order == DONE) return; // die Pille: stopp
        packed.add(order);
    }
}

Schritt für Schritt, in Worten

1. Die Website (Producer) legt Bestellungen in eine Warteschlange mit 3 Plätzen; zwei Packer (Consumer) nehmen sie vorn heraus. Sie ist leer, also warten sie in take.
2. Die Website legt 1001 und 1002 hinten in die Warteschlange.
3. Jeder Packer nimmt die Bestellung vorn heraus und packt sie. Das Packen ist langsamer als das Annehmen von Bestellungen.
4. Die Website legt 1003, 1004 und 1005 hinein. Alle 3 Plätze sind belegt.
5. 1006 passt nicht hinein: put wartet, und die Website nimmt keine Bestellungen mehr an. Das ist Backpressure: Die langsame Seite hält die schnelle zurück.
6. Packer 1 ist mit 1001 fertig und nimmt 1003. Ein Platz wird frei, und das wartende put lässt 1006 hinein.
7. Und so weiter: Die Packer nehmen vorn heraus, die Website fügt hinten hinzu, und in der Warteschlange sind nie mehr als 3.
8. Keine Bestellungen mehr. Die Website legt nach den Bestellungen eine Pille pro Packer hinein, damit jede Bestellung vorher gepackt wird.
9. Jeder Packer beendet seine Bestellung, nimmt eine Pille und hört auf. Ohne die Pille würde er ewig in take warten.
10. 8 von 8, jede einmal: 3 Express für 90 €, 5 Standard für 180 €. Wer welche packt, ändert sich von Lauf zu Lauf; die Summen nicht.

Die Bestellung:

record Order(int id, int euro, boolean express) {}

Die Website und die Packer:

static final Order DONE = new Order(0, 0, false);

static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // wartet, wenn voll
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // eine Pille pro Packer
    }
}

static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // wartet, wenn leer
        if (order == DONE) return; // die Pille: stopp
        packed.add(order);
    }
}

front und pack laufen in drei Threads: eine Website und zwei Packer. Die Website legt die acht Bestellungen hinein und ist fertig, und die Packer nehmen, bis sie ein Zeichen zum Aufhören bekommen.

List<Order> orders = List.of(
        new Order(1001, 25, false),
        new Order(1002, 40, true),
        new Order(1003, 15, false),
        new Order(1004, 60, false),
        new Order(1005, 30, true),
        new Order(1006, 45, false),
        new Order(1007, 20, true),
        new Order(1008, 35, false));
BlockingQueue<Order> queue = new ArrayBlockingQueue<>(3);
Queue<Order> packed = new ConcurrentLinkedQueue<>();
List<Thread> threads = List.of(
        new Thread(() -> run(() -> front(orders, queue, 2))),
        new Thread(() -> run(() -> pack(queue, packed))),
        new Thread(() -> run(() -> pack(queue, packed))));
for (Thread thread : threads) thread.start();
for (Thread thread : threads) thread.join();
report(orders, packed);

Der Bericht prüft Dinge, die nicht davon abhängen, welcher Packer welche Bestellung genommen hat:

static void report(List<Order> orders, Queue<Order> packed) {
    long distinct = packed.stream().map(Order::id).distinct().count();
    System.out.println("Verpackt: " + packed.size() + " von " + orders.size());
    System.out.println("Jede Bestellung einmal: " + (distinct == orders.size() ? "ja" : "nein"));
    for (boolean express : new boolean[] {true, false}) {
        List<Order> group = packed.stream().filter(o -> o.express() == express).toList();
        int euro = group.stream().mapToInt(Order::euro).sum();
        String kind = express ? "Express" : "Standard";
        System.out.println(kind + ": " + group.size() + " Bestellungen, " + euro + " €");
    }
}

Ausgabe

Verpackt: 8 von 8
Jede Bestellung einmal: ja
Express: 3 Bestellungen, 90 €
Standard: 5 Bestellungen, 180 €

Die Ausgabe ist bei jedem Lauf dieselbe. Wie viele Bestellungen jeder Packer genommen hat, ändert sich, deshalb gibt das Programm es nicht aus.

Die Idee

Wie die Packer aufhören

Der Packer wartet in take(), bis eine Bestellung kommt. Wenn die Website einfach endet, warten die Packer ewig, und das Programm endet nie.

Deshalb legt die Website am Ende eine Giftpille (Poison Pill) hinein: einen besonderen Wert, der keine Bestellung ist. Ein Packer, der die Pille nimmt, hört auf. Es gibt eine Pille pro Packer: Jeder nimmt nur eine und legt sie nicht zurück.

Die Pillen kommen nach den Bestellungen in die Warteschlange, damit jede Bestellung vor ihnen gepackt wird.

Hier ist die Pille DONE: ein Order-Objekt, das mit == verglichen wird, also per Referenz. Keine echte Bestellung ist dasselbe Objekt.

Die Idee

Backpressure

Eine volle Warteschlange lässt die Website warten. Das ist kein Fehler, das ist der Sinn: Die langsame Seite drückt zurück auf die schnelle. Das heißt Backpressure. Die Website kann dem Lager höchstens um die Größe der Warteschlange voraus sein, der Speicher wächst nicht, und die letzte Bestellung wartet höchstens so lange, wie das Packen der 3 vor ihr dauert.

Warten ist nicht die einzige Antwort auf eine volle Warteschlange. Es gibt auch eine Methode, die nicht wartet, sondern sagt, ob es geklappt hat:

offer(order) gibt false zurück, wenn kein Platz da ist. offer(order, 1, SECONDS) wartet höchstens eine Sekunde.

Damit kann die Website „Hohe Nachfrage, versuch es in einer Minute noch einmal“ anzeigen, statt zu hängen. Wichtig ist, dass die Entscheidung bei dir liegt: Die Bestellung verschwindet nicht stillschweigend.

Probier es aus

Sag die Ausgabe voraus

Eine Warteschlange mit 2 Plätzen und ohne Packer. Was gibt sie aus?

BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
System.out.println(queue.offer("1001"));
System.out.println(queue.offer("1002"));
System.out.println(queue.offer("1003"));
System.out.println(queue.poll());
System.out.println(queue.offer("1003"));
System.out.println(queue.size());

Antwort

true
true
true
1001
true
3

true
true
false
1002
true
2

true
true
false
1001
true
2

Prüfen

Probier es aus

Ergänze

Wie viele Pillen legt die Website hinein?

for (int i = 0; i < ; i++) {
    queue.put(DONE);
}

Prüfen

Probier es aus

Finde den Fehler

Ein Kollege will nicht, dass die Website wartet, und ändert eine Zeile. Jetzt hängt das Programm nie mehr, aber der Bericht sagt manchmal „Verpackt: 5 von 8“. In welcher Zeile steckt der Fehler?

Wähle die Zeile mit dem Fehler.

1static void front(List<Order> orders,2 BlockingQueue<Order> queue, int packers)3 throws InterruptedException {4 for (Order order : orders) {5 queue.offer(order);6 }7 for (int i = 0; i < packers; i++) {8 queue.put(DONE);9 }10}

Prüfen

Sieh es dir an

Threads, Worker und Nachrichten

Eine Warteschlange kann auch ohne Limit sein: new LinkedBlockingQueue<>(). Dann wartet put nie, und das Problem vom Anfang der Lektion ist zurück. Ein Pool fester Größe (Executors.newFixedThreadPool) ist genau so eine Warteschlange ohne Limit im Inneren: siehe die nächste Lektion.

Übung

Zwei Eingänge, drei Packer

Löse die Aufgabe in deinem Editor.

Bestellungen kommen aus zwei Eingängen: der Website (2001–2006) und der Mobile-App (3001–3006). Jeder Eingang ist ein eigener Producer, und es gibt drei Packer. Die Warteschlange hat 4 Plätze. Die Pillen (eine pro Packer) kommen erst, wenn beide Eingänge fertig sind. Gib aus:

Ausgabe

Verpackt: 12 von 12
Jede Bestellung einmal: ja
Gesamt: 325 €

Weitere Aufgaben

• Ablehnung. Mach die Warteschlange 1 Platz groß und lass die Website offer(order, 10, MILLISECONDS) benutzen. Zähle die abgelehnten Bestellungen und gib „angenommen + abgelehnt = 12“ aus.
• Nach Art. Express-Bestellungen müssen zuerst gepackt werden. Hilft PriorityBlockingQueue? Was passiert mit den Pillen?
• Eine Kette. Füge nach den Packern eine zweite Warteschlange hinzu: Sie legen die gepackten Kartons hinein, und ein Kurier nimmt sie heraus. Wie viele Pillen braucht man jetzt, und wer legt sie hinein?

Test

Teste dich

1. Die Warteschlange zwischen der Website und den Packern ist voll. Was macht put?

Es wirft die Bestellung wegEs wartet, bis ein Packer eine Bestellung nimmt und einen Platz frei machtEs vergrößert die Warteschlange um einen weiteren Platz

2. Warum hat die Warteschlange überhaupt ein Limit? Ohne eines wartet die Website nie.

Kein Grund: Das Limit macht das Programm nur langsamerOhne Limit bremst das langsame Lager die schnelle Website nicht: Die Warteschlange wächst, bis der Speicher ausgeht, und die letzte Bestellung wartet immer längerWeil Warteschlangen ohne Limit zwischen Threads nicht sicher sind

3. Zwei Websites legen Bestellungen hinein, drei Packer nehmen sie heraus. Wie viele Pillen braucht man, und wann werden sie hineingelegt?

Drei, eine pro Packer, nachdem beide Websites fertig sind (nach join)Eine: Der erste Packer, der sie sieht, sagt es den anderenZwei, eine von jeder Website, wenn sie fertig ist

4. Woher weißt du, dass das Programm mit der Warteschlange richtig ist, wenn die Reihenfolge des Packens bei jedem Lauf anders ist?

Du prüfst Dinge, die nicht von der Reihenfolge abhängen: Jede Bestellung wird genau einmal gepackt, die Anzahl und die Summen nach ArtDu lässt es laufen, bis zweimal hintereinander dieselbe Reihenfolge herauskommtDu fügst sleep hinzu, damit die Reihenfolge immer dieselbe ist

Prüfen

Lektion von vorn beginnen

Verwandte Lektionen

• Baut auf: Race Condition · Sperren · Atomare Operationen und unveränderliche Objekte · Deadlock
• Geht weiter in: Thread-Pool und Aufgaben · Ereignisse und Warteschlangen
• Aufgaben zum Thema: Nebenläufigkeit und Race Conditions