Zum Inhalt
wannacode

Producer und Consumer

Kurz gesagt: Producer-Consumer ist ein Muster, bei dem einige Teile eines Programms Aufgaben in eine begrenzte Warteschlange legen und andere sie herausnehmen; in JavaScript sind es async-Funktionen in einem Thread, und eine volle Warteschlange lässt den Producer warten.

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.
JavaScript hat keine solche fertige Warteschlange. In einem Thread mit async-Funktionen sind es mit Promises weniger als 40 Zeilen: put und take geben ein Promise zurück, das erfüllt wird, wenn ein Platz oder eine Bestellung da ist. Ein Lock ist nicht nötig, weil der Code zwischen zwei awaits nicht unterbrochen werden kann.
BoundedQueue
class BoundedQueue {
  #items = [];
  #putters = []; // die, die etwas hineinlegen wollen
  #takers = []; // die, die etwas herausnehmen wollen


  constructor(capacity) {
    this.capacity = capacity;
  }


  get size() {
    return this.#items.length;
  }


  async put(item) {
    while (this.#items.length >= this.capacity) {
      await new Promise((resolve) => this.#putters.push(resolve));
    }
    this.#items.push(item);
    this.#takers.shift()?.();
  }


  tryPut(item) {
    if (this.#items.length >= this.capacity) return false;
    this.#items.push(item);
    this.#takers.shift()?.();
    return true;
  }


  async take() {
    while (this.#items.length === 0) {
      await new Promise((resolve) => this.#takers.push(resolve));
    }
    const item = this.#items.shift();
    this.#putters.shift()?.();
    return item;
  }
}

#putters und #takers sind Warteschlangen von resolve-Funktionen: wer auf einen Platz wartet und wer auf eine Bestellung. take weckt einen wartenden Producer, put weckt einen wartenden Consumer. Wer aufwacht, prüft noch einmal (while), weil jemand anderes schneller gewesen sein kann.

Sieh es dir an

Die Website und die Packer

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // wartet, wenn voll
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // eine Pille pro Packer
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // wartet, wenn leer
    if (order === DONE) return; // die Pille: stopp
    packed.push(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 await 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: await 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 await 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:

// eine Bestellung ist { id, euro, express }

Die Website und die Packer:

const DONE = null;


async function front(orders, queue, packers) {
  for (const order of orders) {
    await queue.put(order); // wartet, wenn voll
  }
  for (let i = 0; i < packers; i++) {
    await queue.put(DONE); // eine Pille pro Packer
  }
}


async function pack(queue, packed) {
  while (true) {
    const order = await queue.take(); // wartet, wenn leer
    if (order === DONE) return; // die Pille: stopp
    packed.push(order);
  }
}
front und die zwei packs sind drei async-Funktionen in einem Thread. await ist die Stelle, an der jede warten und die anderen laufen lassen kann. Das ist abwechselndes Arbeiten, nicht paralleles: Für Packen, das viel rechnet, brauchst du Worker (siehe unten).
const orders = [
  { id: 1001, euro: 25, express: false },
  { id: 1002, euro: 40, express: true },
  { id: 1003, euro: 15, express: false },
  { id: 1004, euro: 60, express: false },
  { id: 1005, euro: 30, express: true },
  { id: 1006, euro: 45, express: false },
  { id: 1007, euro: 20, express: true },
  { id: 1008, euro: 35, express: false },
];
const queue = new BoundedQueue(3);
const packed = [];
await Promise.all([
  front(orders, queue, 2),
  pack(queue, packed),
  pack(queue, packed),
]);
report(orders, packed);

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

function report(orders, packed) {
  const distinct = new Set(packed.map((order) => order.id)).size;
  console.log(`Verpackt: ${packed.length} von ${orders.length}`);
  console.log(`Jede Bestellung einmal: ${distinct === orders.length ? "ja" : "nein"}`);
  for (const express of [true, false]) {
    const group = packed.filter((order) => order.express === express);
    const euro = group.reduce((sum, order) => sum + order.euro, 0);
    const kind = express ? "Express" : "Standard";
    console.log(`${kind}: ${group.length} 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 await queue.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 = null.

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:

tryPut(order) in unserer Warteschlange gibt false zurück, wenn kein Platz da ist.

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?

const queue = new BoundedQueue(2);
console.log(queue.tryPut("1001"));
console.log(queue.tryPut("1002"));
console.log(queue.tryPut("1003"));
console.log(await queue.take());
console.log(queue.tryPut("1003"));
console.log(queue.size);
Antwort

Probier es aus

Ergänze

Wie viele Pillen legt die Website hinein?
for (let i = 0; i < ; i++) {
  await queue.put(DONE);
}

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.

Sieh es dir an

Threads, Worker und Nachrichten

Wenn das Packen viel rechnet, brauchst du echte Threads: Worker. Zwischen ihnen gibt es keine Warteschlange gemeinsamer Objekte. Der Haupt-Thread schickt jede Bestellung mit postMessage an einen Worker, und der Worker empfängt sie in on("message"):

import { Worker, isMainThread, parentPort } from "node:worker_threads";
import { once } from "node:events";


if (isMainThread) {
  const orders = [
    { id: 1001, euro: 25, express: false },
    { id: 1002, euro: 40, express: true },
    { id: 1003, euro: 15, express: false },
    { id: 1004, euro: 60, express: false },
    { id: 1005, euro: 30, express: true },
    { id: 1006, euro: 45, express: false },
    { id: 1007, euro: 20, express: true },
    { id: 1008, euro: 35, express: false },
  ];
  const packers = [0, 1].map(() => new Worker(new URL(import.meta.url)));
  const packed = [];
  const reports = packers.map((packer) => once(packer, "message"));
  orders.forEach((order, i) => packers[i % 2].postMessage(order)); // wartet nie
  packers.forEach((packer) => packer.postMessage(null));
  for (const [done] of await Promise.all(reports)) packed.push(...done);
  report(orders, packed);
} else {
  const done = [];
  parentPort.on("message", (order) => {
    if (order === null) {
      parentPort.postMessage(done);
      parentPort.close();
    } else {
      done.push(order);
    }
  });
}
Ausgabe
Verpackt: 8 von 8
Jede Bestellung einmal: ja
Express: 3 Bestellungen, 90 €
Standard: 5 Bestellungen, 180 €

Das ist ebenfalls Producer und Consumer, mit zwei Unterschieden. Erstens gibt es kein Limit: postMessage wartet nie, die Nachrichten stauen sich im Speicher des Workers, und Backpressure musst du selbst bauen (der Worker antwortet „bereit“ und bekommt erst dann die nächste). Zweitens werden die Bestellungen nicht aus einer gemeinsamen Warteschlange genommen; der Haupt-Thread entscheidet, welche Bestellung an welchen Worker geht (hier abwechselnd). Die Bestellung wird beim Senden kopiert, also gibt es dafür keine Race Condition.

Ü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 tryPut benutzen. Zähle die abgelehnten Bestellungen und gib „angenommen + abgelehnt = 12“ aus.
  • Nach Art. Express-Bestellungen müssen zuerst gepackt werden. Hilft zwei Warteschlangen? 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?
2. Warum hat die Warteschlange überhaupt ein Limit? Ohne eines wartet die Website nie.
3. Zwei Websites legen Bestellungen hinein, drei Packer nehmen sie heraus. Wie viele Pillen braucht man, und wann werden sie hineingelegt?
4. Woher weißt du, dass das Programm mit der Warteschlange richtig ist, wenn die Reihenfolge des Packens bei jedem Lauf anders ist?

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