Aufgabenwarteschlangen in Go: Erstellung, Verarbeitung und Optimierung

Einführung: Warum braucht man Aufgabenwarteschlangen in Go?

In der modernen Backend-Entwicklung treten oft Aufgaben auf, die keine sofortige Antwort an den Benutzer erfordern: E-Mail-Versand, Berichtserstellung, Bildverarbeitung, Synchronisation mit externen APIs. Solche Operationen synchron innerhalb einer HTTP-Anfrage auszuführen, ist eine schlechte Praxis, da dies die Antwort verlangsamt und die Serverlast erhöht.

Aufgabenwarteschlangen (Job Queues) lösen dieses Problem. In Go (Golang) ermöglichen die eingebauten Konkurrenzmechanismen – Goroutinen und Kanäle – den Aufbau zuverlässiger und leistungsfähiger Warteschlangen ohne externe Abhängigkeiten. In diesem Artikel werden wir uns ansehen, wie man eine Aufgabenwarteschlange von Grund auf erstellt, sie mit einem Worker Pool optimiert und typische Fehler vermeidet.

Grundlagen: Goroutinen und Kanäle

Go bietet zwei wichtige Primitive für die Arbeit mit Warteschlangen:

• Goroutinen – leichtgewichtige Threads, die mit dem Schlüsselwort go gestartet werden.
• Kanäle – typisierte „Röhren“ zur Datenübertragung zwischen Goroutinen.

Die einfachste Warteschlange ist eine Goroutine, die aus einem Kanal liest und Aufgaben verarbeitet:

package main

import (    "fmt"    "time")

func worker(jobs <-chan int) {    for job := range jobs {        fmt.Printf("Verarbeite Aufgabe %d", job)        time.Sleep(1 * time.Second) // Arbeitsimulation    }}

func main() {    jobs := make(chan int, 10)

    // Worker starten    go worker(jobs)

    // Aufgaben senden    for i := 1; i <= 5; i++ {        jobs <- i    }

    close(jobs) // Kanal schließen nach dem Senden aller Aufgaben    time.Sleep(2 * time.Second)}

Dieser Code erstellt einen einzelnen Worker, der Aufgaben nacheinander verarbeitet. Für reale Projekte wird jedoch Skalierbarkeit benötigt – mehrere Worker.

Implementierung eines Worker Pools in Go

Ein Worker Pool ist ein Entwurfsmuster, bei dem wir eine feste Anzahl von Goroutine-Workern starten, die jeweils Aufgaben aus einem gemeinsamen Kanal nehmen. Dies ermöglicht die Kontrolle über den Parallelitätsgrad und die Systemlast.

Basisimplementierung

package main

import (    "fmt"    "sync"    "time")

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {    defer wg.Done()    for job := range jobs {        fmt.Printf("Worker %d: Aufgabe %d", id, job)        time.Sleep(500 * time.Millisecond)    }}

func main() {    const numWorkers = 3    const numJobs = 10

    jobs := make(chan int, numJobs)    var wg sync.WaitGroup

    // Worker starten    for i := 1; i <= numWorkers; i++ {        wg.Add(1)        go worker(i, jobs, &wg)    }

    // Aufgaben senden    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs)

    wg.Wait() // Auf Abschluss aller Worker warten    fmt.Println("Alle Aufgaben erledigt")}

Hier verwenden wir sync.WaitGroup, um auf den Abschluss aller Worker zu warten. Der Kanal mit dem Puffer numJobs ermöglicht das Senden aller Aufgaben ohne Blockierung, falls die Worker sie noch nicht abgeholt haben.

Dynamisches Hinzufügen von Aufgaben

In realen Anwendungen kommen Aufgaben asynchron an – zum Beispiel per HTTP. Hier ist ein Beispiel mit einem Handler:

package main

import (    "fmt"    "net/http"    "sync"    "time")

type Task struct {    ID   int    Data string}

var (    taskQueue = make(chan Task, 100)    wg        sync.WaitGroup)

func worker(id int) {    defer wg.Done()    for task := range taskQueue {        fmt.Printf("[Worker %d] Verarbeitung: %s (ID=%d)", id, task.Data, task.ID)        time.Sleep(1 * time.Second)    }}

func handler(w http.ResponseWriter, r *http.Request) {    // Aufgabe aus der Anfrage holen (vereinfacht)    task := Task{ID: time.Now().Nanosecond(), Data: "Neue Aufgabe"}    taskQueue <- task    w.W