Redis-Warteschlange und Cron in Go

Der Originalbeitrag ist hier

In diesem Tutorial werden wir mit einer Warteschlange interagieren und sie auf einen Redis-Server stellen
Verwenden Sie das Paket github.com/hibiken/asynq und erstellen Sie einen Planer für einen
Geplante Aufgabe mit dem Paket github.com/robfig/cron. Dies Schritt für Schritt
Der Leitfaden erklärt, wie man eine Warteschlange einrichtet, Aufgaben plant und ordnungsgemäß damit umgeht
Abschaltungen.

Initialisieren Sie das Modul

Erstellen Sie zunächst ein neues Go-Modul für das Projekt:

go mod init learn_queue_and_cron

Erstellen Sie cron.go

Die Datei cron.go ist für die Planung und Ausführung von Aufgaben zu bestimmten Zeiten verantwortlich
Intervalle. Unten ist die Implementierung:

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/robfig/cron/v3"
)

func runCron(c *cron.Cron) {

    // Schedule a task to run every minute
    _, err := c.AddFunc("@every 1m", func() {
        fmt.Printf("Task executed every minute at: %v \n", time.Now().Local())
    })
    if err != nil {
        log.Fatal(err)
    }

    // Start the cron scheduler
    c.Start()
    log.Println("Cron scheduler started")

    // Keep the main goroutine running
    select {}
}

Dieser Code plant die Ausführung einer Aufgabe jede Minute und hält die Anwendung am Laufen
um sicherzustellen, dass der Planer kontinuierlich funktioniert.

Erstellen Sie queue.go

Die Datei queue.go verwaltet die Aufgabenverarbeitung mit Asynq. Hier ist der Code:

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"

    "github.com/hibiken/asynq"
)

func runQueue(server *asynq.Server) {
    mux := asynq.NewServeMux()
    mux.HandleFunc("send_email", emailHandler)
    mux.HandleFunc("generate_report", reportHandler)

    if err := server.Run(mux); err != nil {
        log.Fatalf("Failed to run Asynq server: %v", err)
    }
}

func emailHandler(ctx context.Context, task *asynq.Task) error {
    var payload struct {
        To string `json:"to"`
    }
    if err := json.Unmarshal(task.Payload(), &payload); err != nil {
        return fmt.Errorf("failed to unmarshal payload: %w", err)
    }
    fmt.Printf("Sending email to: %s\n", payload.To)
    return nil
}

func reportHandler(ctx context.Context, task *asynq.Task) error {
    var payload struct {
        ReportID int `json:"report_id"`
    }
    if err := json.Unmarshal(task.Payload(), &payload); err != nil {
        return fmt.Errorf("failed to unmarshal payload: %w", err)
    }
    fmt.Printf("Generating report for ID: %d\n", payload.ReportID)
    return nil
}

Erläuterung

• Handler: emailHandler und reportHandler verarbeiten Aufgaben durch Parsen ihre Nutzlasten und die Ausführung der entsprechenden Aktionen.
• Aufgabenwarteschlange:Aufgaben wie „send_email“ und „generate_report“ sind definiert und über die Aufgabenwarteschlange von Asynq verarbeitet.

Erstellen Sie router.go

Die Datei router.go richtet HTTP-Endpunkte ein, um Aufgaben in die Warteschlange zu stellen:

package main

import (
    "encoding/json"
    "net/http"

    "github.com/gin-gonic/gin"
    "github.com/hibiken/asynq"
)

func setupRouter(client *asynq.Client) *gin.Engine {
    r := gin.Default()

    r.POST("/enqueue/email", func(c *gin.Context) {
        var payload struct {
            To string `json:"to"`
        }
        if err := c.ShouldBindJSON(&payload); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
            return
        }

        jsonPayload, err := json.Marshal(payload)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to marshal payload"})
            return
        }

        task := asynq.NewTask("send_email", jsonPayload)
        _, err = client.Enqueue(task)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to enqueue task"})
            return
        }

        c.JSON(http.StatusOK, gin.H{"message": "Email job enqueued"})
    })

    r.POST("/enqueue/report", func(c *gin.Context) {
        var payload struct {
            ReportID int `json:"report_id"`
        }
        if err := c.ShouldBindJSON(&payload); err != nil {
            c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
            return
        }

        jsonPayload, err := json.Marshal(payload)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to marshal payload"})
            return
        }

        task := asynq.NewTask("generate_report", jsonPayload)
        _, err = client.Enqueue(task)
        if err != nil {
            c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to enqueue task"})
            return
        }

        c.JSON(http.StatusOK, gin.H{"message": "Report job enqueued"})
    })

    return r
}

Dieser Code verwendet das Gin-Framework, um zwei Endpunkte für das Einreihen von Aufgaben bereitzustellen.

Erstellen Sie main.go

Die main.go-Datei integriert alles zusammen:

package main

import (
    "context"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/hibiken/asynq"
    "github.com/robfig/cron/v3"
)

func main() {
    c := cron.New()

    server := asynq.NewServer(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        asynq.Config{
            Concurrency: 10,
        },
    )

    client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
    defer client.Close()

    router := setupRouter(client)

    httpServer := &http.Server{
        Addr:    ":8080",
        Handler: router,
    }

    // Prepare shutdown context
    ctx, stop := context.WithCancel(context.Background())
    defer stop()
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt, syscall.SIGTERM)

    go runQueue(server)
    go runCron(c)
    go func() {
        if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Failed to run HTTP server: %v", err)
        }
    }()

    appShutdown(ctx, httpServer, c, server, quit)
}

func appShutdown(ctx context.Context, httpServer *http.Server, c *cron.Cron, server *asynq.Server, quit chan os.Signal) {
    // Wait for termination signal
    <-quit
    log.Println("Shutting down gracefully...")

    httpCtx, httpCancel := context.WithTimeout(ctx, 5*time.Second)
    defer httpCancel()
    if err := httpServer.Shutdown(httpCtx); err != nil {
        log.Printf("HTTP server shutdown error: %v", err)
    }

    server.Shutdown()
    c.Stop()

    log.Println("Application stopped")
}

Diese Datei kombiniert die Warteschlange, den Cron, den HTTP-Server und die Logik zum Herunterfahren.

Abhängigkeiten installieren

Installieren Sie alle erforderlichen Abhängigkeiten:

go mod tidy

Erstellen Sie die Anwendung und führen Sie sie aus

Erstellen und führen Sie die Anwendung aus mit:

go build -o run *.go && ./run

Testen Sie die Anwendung

Besuchen Sie die folgenden Endpunkte, um Aufgaben in die Warteschlange zu stellen:

• http://localhost:8080/enqueue/email
• http://localhost:8080/enqueue/report

Beobachten Sie das Terminal auf die Protokolle zur Aufgabenausführung.

Kanonische URL

Weitere Informationen finden Sie im Originalbeitrag auf meinem Blog.

Das obige ist der detaillierte Inhalt vonRedis-Warteschlange und Cron in Go. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!