Aufgabenwarteschlangen in Go: Ein praktischer Leitfaden

Einleitung: Warum braucht man Aufgabenwarteschlangen in Go?

In der modernen Entwicklung besteht oft die Notwendigkeit, Aufgaben asynchron zu verarbeiten, ohne den Hauptthread der Anwendung zu blockieren. Aufgabenwarteschlangen (Task Queues) sind ein klassisches Muster, das es ermöglicht, Arbeit auf mehrere Bearbeiter (Worker) zu verteilen, die Last zu steuern und Ausfallsicherheit zu gewährleisten.

Die Sprache Go (Golang) bietet integrierte Mechanismen zur Implementierung von Aufgabenwarteschlangen: Goroutinen (goroutines) und Kanäle (channels). In diesem Artikel werden wir praktisch erarbeiten, wie man mit der Standardbibliothek von Go eine zuverlässige Aufgabenwarteschlange aufbaut, ohne auf externe Nachrichtenbroker zurückzugreifen (obwohl RabbitMQ oder Redis in der Produktion nützlich sein können).

Hauptteil

1. Das grundlegende Worker-Pool-Muster

Ein Worker-Pool ist eine Gruppe einer festgelegten Anzahl von Goroutinen (Workern), die einen gemeinsamen Kanal abhören und eingehende Aufgaben verarbeiten. Dies ermöglicht die Kontrolle des Parallelitätsgrads und vermeidet eine Überlastung des Systems.

Betrachten wir ein einfaches Beispiel: Wir erstellen einen Pool von 3 Workern, die ganze Zahlen verarbeiten (Arbeitssimulation).

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\nfunc worker(id int, jobs <-chan int, wg *sync.WaitGroup) {\n    defer wg.Done()\n    for job := range jobs {\n        fmt.Printf("Worker %d hat Aufgabe %d gestartet\\n", id, job)\n        time.Sleep(time.Second) // Arbeitssimulation\n        fmt.Printf("Worker %d hat Aufgabe %d abgeschlossen\\n", id, job)\n    }\n}\n\nfunc main() {\n    const numJobs = 5\n    const numWorkers = 3\n\n    jobs := make(chan int, numJobs)\n    var wg sync.WaitGroup\n\n    // Worker starten\n    for w := 1; w <= numWorkers; w++ {\n        wg.Add(1)\n        go worker(w, jobs, &wg)\n    }\n\n    // Aufgaben senden\n    for j := 1; j <= numJobs; j++ {\n        jobs <- j\n    }\n    close(jobs) // Kanal schließen, damit die Worker beenden\n\n    wg.Wait()\n    fmt.Println("Alle Aufgaben erledigt")\n}

Code-Analyse:

\n
• jobs — ein gepufferter Kanal, in den wir Aufgaben senden.
\n
• worker — eine Funktion, die aus dem Kanal liest und jede Aufgabe verarbeitet.
\n
• sync.WaitGroup — wartet auf das Ende aller Worker.
\n
• Nach dem Senden aller Aufgaben wird der Kanal geschlossen (close(jobs)), was die for range-Schleife in den Workern beendet.
\n

Dieses Muster ist die Grundlage für komplexere Lösungen.

2. Warteschlange mit Prioritäten und Timeouts

In realen Anwendungen können Aufgaben unterschiedliche Prioritäten haben oder eine Ausführung innerhalb einer bestimmten Zeit erfordern. Wir implementieren eine Warteschlange, bei der Aufgaben mit Priorität zuerst verarbeitet werden, und fügen einen Kontext mit Timeout hinzu, um langlaufende Aufgaben abzubrechen.

package main\n\nimport (\n    "context"\n    "fmt"\n    "time"\n)\n\ntype Task struct {\n    ID       int\n    Priority int // je kleiner die Zahl, desto höher die Priorität\n    Data     string\n}\n\nfunc processTask(ctx context.Context, task Task) error {\n    select {\n    case <-ctx.Done():\n        return ctx.Err()\n    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):\n        fmt.Printf("Aufgabe %d verarbeitet: %s (Priorität %d)\\n", task.ID, task.Data, task.Priority)\n        return nil\n    }\n}\n\nfunc main() {\n    tasks := []Task{\n        {ID: 1, Priority: 3, Data: "Niedrige Priorität"},\n        {ID: 2, Priority: 1, Data: "Hohe Priorität"},\n        {ID: 3, Priority: 2, Data: "Mittlere Priorität"},\n    }\n\n    // Sortierung nach Priorität (höchste zuerst)\n    // Verwenden Sie in einem echten Projekt eine Sortierung oder einen Heap (container/heap)\n    for i := 0; i < len(tasks); i++ {\n        for j := i + 1; j < len(tasks); j++ {\n            if tasks[j].Priority < tasks[i].Priority {\n                tasks[i], tasks[j] = tasks[j], tasks[i]\n            }\n