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 C# ist das BlockingCollection<T> mit boundedCapacity: 3. Add wartet, wenn sie voll ist. GetConsumingEnumerable() ist eine Schleife, die nimmt, solange etwas da ist, und wartet, wenn sie leer ist. Sie liegt in System.Collections.Concurrent, wie ConcurrentBag, in der wir die gepackten Bestellungen sammeln.

Sieh es dir an

Die Website und die Packer

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // wartet, wenn voll
    }
    queue.CompleteAdding(); // Schluss: die Packer hören auf
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // wartet, wenn leer
    foreach (var order in queue.GetConsumingEnumerable())
    {
        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 GetConsumingEnumerable.
  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: Add 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. CompleteAdding sagt der Warteschlange, dass nichts mehr kommt.
  9. Leer und abgeschlossen: GetConsumingEnumerable endet, und jedes foreach hört auf. Ohne CompleteAdding würden die Packer ewig 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, bool Express);

Die Website und die Packer:

static void Front(List<Order> orders,
                  BlockingCollection<Order> queue)
{
    foreach (var order in orders)
    {
        queue.Add(order); // wartet, wenn voll
    }
    queue.CompleteAdding(); // Schluss: die Packer hören auf
}


static void Pack(BlockingCollection<Order> queue,
                 ConcurrentBag<Order> packed)
{
    // wartet, wenn leer
    foreach (var order in queue.GetConsumingEnumerable())
    {
        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.
var orders = new List<Order>
{
    new(1001, 25, false),
    new(1002, 40, true),
    new(1003, 15, false),
    new(1004, 60, false),
    new(1005, 30, true),
    new(1006, 45, false),
    new(1007, 20, true),
    new(1008, 35, false),
};
var queue = new BlockingCollection<Order>(boundedCapacity: 3);
var packed = new ConcurrentBag<Order>();
var threads = new List<Thread>
{
    new Thread(() => Front(orders, queue)),
    new Thread(() => Pack(queue, packed)),
    new Thread(() => Pack(queue, packed)),
};
foreach (var thread in threads) thread.Start();
foreach (var thread in 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, ConcurrentBag<Order> packed)
{
    int distinct = packed.Select(o => o.Id).Distinct().Count();
    Console.WriteLine($"Verpackt: {packed.Count} von {orders.Count}");
    Console.WriteLine($"Jede Bestellung einmal: {(distinct == orders.Count ? "ja" : "nein")}");
    foreach (var express in new[] { true, false })
    {
        var group = packed.Where(o => o.Express == express).ToList();
        string kind = express ? "Express" : "Standard";
        Console.WriteLine($"{kind}: {group.Count} Bestellungen, {group.Sum(o => o.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 GetConsumingEnumerable(), bis eine Bestellung kommt. Wenn die Website einfach endet, warten die Packer ewig, und das Programm endet nie.

Deshalb ruft die Website am Ende CompleteAdding() auf: „Es kommt nichts mehr.“ Danach wirft Add eine Exception, und GetConsumingEnumerable() endet, sobald die Warteschlange leer ist. Jedes foreach darüber hört von selbst auf, egal wie viele Packer es gibt. Die Bestellungen, die schon darin sind, werden zuerst gepackt.

Der andere Weg, dem du in fremdem Code begegnest, ist die Giftpille (Poison Pill): ein besonderer Wert, der keine Bestellung ist, einer pro Consumer. In C# brauchst du sie nicht.

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:

TryAdd(order) gibt false zurück, wenn kein Platz da ist. TryAdd(order, 1000) 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?

var queue = new BlockingCollection<string>(boundedCapacity: 2);
Console.WriteLine(queue.TryAdd("1001"));
Console.WriteLine(queue.TryAdd("1002"));
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Take());
Console.WriteLine(queue.TryAdd("1003"));
Console.WriteLine(queue.Count);
Antwort

Probier es aus

Ergänze

Was ruft die Website am Ende auf, damit die Packer aufhören?
queue.(); // Schluss: die Packer hören auf

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

In neuem C#-Code begegnet dir auch Channel<T> (System.Threading.Channels). Channel.CreateBounded<Order>(3) ist dieselbe Warteschlange mit 3 Plätzen, aber für async-Code: await writer.WriteAsync(order) wartet, ohne einen Thread zu blockieren, writer.Complete() ist CompleteAdding, und await foreach (var o in reader.ReadAllAsync()) ist der Consumer. BlockingCollection blockiert Threads; Channel lässt sie beim Warten los.

Ü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 (CompleteAdding) 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 TryAdd(order, 10) benutzen. Zähle die abgelehnten Bestellungen und gib „angenommen + abgelehnt = 12“ aus.
  • Nach Art. Express-Bestellungen müssen zuerst gepackt werden. Hilft zwei Warteschlangen und ein Packer, der zuerst in die Express-Warteschlange schaut? 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 ·