Zum Inhalt
wannacode

Producer und Consumer

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 Python ist das queue.Queue(maxsize=3). put wartet, wenn sie voll ist, und get wartet, wenn sie leer ist. (Das Modul queue ist für Threads. Für async-Code gibt es auch asyncio.Queue mit denselben Methoden.)

Sieh es dir an

Die Website und die Packer

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # wartet, wenn voll
    for _ in range(packers):
        queue.put(DONE)  # eine Pille pro Packer




def pack(queue, packed):
    while True:
        order = queue.get()  # wartet, wenn leer
        if order is DONE:
            return  # die Pille: stopp
        packed.append(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 get.
  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 get 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:

@dataclass(frozen=True)
class Order:
    id: int
    euro: int
    express: bool

Die Website und die Packer:

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # wartet, wenn voll
    for _ in range(packers):
        queue.put(DONE)  # eine Pille pro Packer




def pack(queue, packed):
    while True:
        order = queue.get()  # wartet, wenn leer
        if order is DONE:
            return  # die Pille: stopp
        packed.append(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.
orders = [
    Order(1001, 25, False),
    Order(1002, 40, True),
    Order(1003, 15, False),
    Order(1004, 60, False),
    Order(1005, 30, True),
    Order(1006, 45, False),
    Order(1007, 20, True),
    Order(1008, 35, False),
]
queue = Queue(maxsize=3)
packed = []  # list.append ist von vielen Threads aus sicher
threads = [threading.Thread(target=front, args=(orders, queue, 2)),
           threading.Thread(target=pack, args=(queue, packed)),
           threading.Thread(target=pack, args=(queue, packed))]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
report(orders, packed)

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

def report(orders, packed):
    distinct = len({order.id for order in packed})
    print(f"Verpackt: {len(packed)} von {len(orders)}")
    print("Jede Bestellung einmal:", "ja" if distinct == len(orders) else "nein")
    for express in (True, False):
        group = [order for order in packed if order.express == express]
        kind = "Express" if express else "Standard"
        print(f"{kind}: {len(group)} Bestellungen, {sum(o.euro for o in group)} €")
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 get(), 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 = None, und is DONE vergleicht sie nach Identität.

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:

put_nowait(order) löst queue.Full aus, wenn kein Platz da ist. put(order, timeout=1) wartet höchstens eine Sekunde und löst dann Full aus.

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?

def offer(queue, item):
    try:
        queue.put_nowait(item)
        return True
    except Full:
        return False




queue = Queue(maxsize=2)
print(offer(queue, "1001"))
print(offer(queue, "1002"))
print(offer(queue, "1003"))
print(queue.get())
print(offer(queue, "1003"))
print(queue.qsize())
Antwort

Probier es aus

Ergänze

Wie viele Pillen legt die Website hinein?
for _ in range():
    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

Queue() ohne maxsize hat kein Limit: put wartet nie, und das Problem vom Anfang der Lektion ist zurück. In async-Code ist die Warteschlange asyncio.Queue(maxsize=3) mit await queue.put(...) und await queue.get(): dasselbe, aber in einem Thread.

Ü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 put(order, timeout=0.01) benutzen. Zähle die abgelehnten Bestellungen und gib „angenommen + abgelehnt = 12“ aus.
  • Nach Art. Express-Bestellungen müssen zuerst gepackt werden. Hilft queue.PriorityQueue? 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 ·